src/sort.c

   1 /* PSPP - computes sample statistics.
   2    Copyright (C) 1997-9, 2000 Free Software Foundation, Inc.
   3    Written by Ben Pfaff <blp@gnu.org>.
   4
   5    This program is free software; you can redistribute it and/or
   6    modify it under the terms of the GNU General Public License as
   7    published by the Free Software Foundation; either version 2 of the
   8    License, or (at your option) any later version.
   9
  10    This program is distributed in the hope that it will be useful, but
  11    WITHOUT ANY WARRANTY; without even the implied warranty of
  12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13    General Public License for more details.
  14
  15    You should have received a copy of the GNU General Public License
  16    along with this program; if not, write to the Free Software
  17    Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
  18    02111-1307, USA. */
  19
  20 #include <config.h>
  21 #include "sort.h"
  22 #include <assert.h>
  23 #include <stdio.h>
  24 #include <stdlib.h>
  25 #include <errno.h>
  26 #include "algorithm.h"
  27 #include "alloc.h"
  28 #include "command.h"
  29 #include "error.h"
  30 #include "expr.h"
  31 #include "lexer.h"
  32 #include "misc.h"
  33 #include "settings.h"
  34 #include "str.h"
  35 #include "var.h"
  36 #include "vfm.h"
  37 #include "vfmP.h"
  38
  39 #if HAVE_UNISTD_H
  40 #include <unistd.h>
  41 #endif
  42
  43 #if HAVE_SYS_TYPES_H
  44 #include <sys/types.h>
  45 #endif
  46
  47 #if HAVE_SYS_STAT_H
  48 #include <sys/stat.h>
  49 #endif
  50
  51 #include "debug-print.h"
  52
  53 /* Other prototypes. */
  54 static int compare_record (const union value *, const union value *,
  55                            const struct sort_cases_pgm *);
  56 static int compare_case_lists (const void *, const void *, void *);
  57 static struct internal_sort *do_internal_sort (struct sort_cases_pgm *,
  58                                                int separate);
  59 static void destroy_internal_sort (struct internal_sort *);
  60 static struct external_sort *do_external_sort (struct sort_cases_pgm *,
  61                                                int separate);
  62 static void destroy_external_sort (struct external_sort *);
  63 struct sort_cases_pgm *parse_sort (void);
  64
  65 /* Performs the SORT CASES procedures. */
  66 int
  67 cmd_sort_cases (void)
  68 {
  69   struct sort_cases_pgm *scp;
  70   int success;
  71
  72   lex_match_id ("SORT");
  73   lex_match_id ("CASES");
  74   lex_match (T_BY);
  75
  76   scp = parse_sort ();
  77   if (scp == NULL)
  78     return CMD_FAILURE;
  79
  80   cancel_temporary ();
  81
  82   success = sort_cases (scp, 0);
  83   destroy_sort_cases_pgm (scp);
  84   if (success)
  85     return lex_end_of_command ();
  86   else
  87     return CMD_FAILURE;
  88 }
  89
  90 /* Parses a list of sort keys and returns a struct sort_cases_pgm
  91    based on it.  Returns a null pointer on error. */
  92 struct sort_cases_pgm *
  93 parse_sort (void)
  94 {
  95   struct sort_cases_pgm *scp;
  96
  97   scp = xmalloc (sizeof *scp);
  98   scp->ref_cnt = 1;
  99   scp->vars = NULL;
 100   scp->dirs = NULL;
 101   scp->var_cnt = 0;
 102   scp->isrt = NULL;
 103   scp->xsrt = NULL;
 104
 105   do
 106     {
 107       int prev_var_cnt = scp->var_cnt;
 108       enum sort_direction direction = SRT_ASCEND;
 109
 110       /* Variables. */
 111       if (!parse_variables (default_dict, &scp->vars, &scp->var_cnt,
 112                             PV_NO_DUPLICATE | PV_APPEND | PV_NO_SCRATCH))
 113         goto error;
 114
 115       /* Sort direction. */
 116       if (lex_match ('('))
 117         {
 118           if (lex_match_id ("D") || lex_match_id ("DOWN"))
 119             direction = SRT_DESCEND;
 120           else if (!lex_match_id ("A") && !lex_match_id ("UP"))
 121             {
 122               msg (SE, _("`A' or `D' expected inside parentheses."));
 123               goto error;
 124             }
 125           if (!lex_match (')'))
 126             {
 127               msg (SE, _("`)' expected."));
 128               goto error;
 129             }
 130         }
 131       scp->dirs = xrealloc (scp->dirs, sizeof *scp->dirs * scp->var_cnt);
 132       for (; prev_var_cnt < scp->var_cnt; prev_var_cnt++)
 133         scp->dirs[prev_var_cnt] = direction;
 134     }
 135   while (token != '.' && token != '/');
 136
 137   return scp;
 138
 139  error:
 140   destroy_sort_cases_pgm (scp);
 141   return NULL;
 142 }
 143
 144 void
 145 destroy_sort_cases_pgm (struct sort_cases_pgm *scp)
 146 {
 147   if (scp != NULL)
 148     {
 149       assert (scp->ref_cnt > 0);
 150       if (--scp->ref_cnt > 0)
 151         return;
 152
 153       free (scp->vars);
 154       free (scp->dirs);
 155       destroy_internal_sort (scp->isrt);
 156       destroy_external_sort (scp->xsrt);
 157       free (scp);
 158     }
 159 }
 160
 161 /* Sorts the active file based on the key variables specified in
 162    global variables vars and var_cnt.  The output is either to the
 163    active file, if SEPARATE is zero, or to a separate file, if
 164    SEPARATE is nonzero.  In the latter case the output cases can be
 165    read with a call to read_sort_output().  (In the former case the
 166    output cases should be dealt with through the usual vfm interface.)
 167
 168    The caller is responsible for freeing vars[]. */
 169 int
 170 sort_cases (struct sort_cases_pgm *scp, int separate)
 171 {
 172   /* Not sure this is necessary but it's good to be safe. */
 173   if (separate && case_source_is_class (vfm_source, &sort_source_class))
 174     procedure (NULL, NULL, NULL, NULL);
 175
 176   /* SORT CASES cancels PROCESS IF. */
 177   expr_free (process_if_expr);
 178   process_if_expr = NULL;
 179
 180   /* Try an internal sort first. */
 181   scp->isrt = do_internal_sort (scp, separate);
 182   if (scp->isrt != NULL)
 183     return 1;
 184
 185   /* Fall back to an external sort. */
 186   write_active_file_to_disk ();
 187   scp->xsrt = do_external_sort (scp, separate);
 188   if (scp->xsrt != NULL)
 189     return 1;
 190
 191   destroy_sort_cases_pgm (scp);
 192   return 0;
 193 }
 194 \f
 195 struct internal_sort
 196   {
 197     struct case_list **results;
 198   };
 199
 200 /* If a reasonable situation is set up, do an internal sort of the
 201    data.  Return success. */
 202 static struct internal_sort *
 203 do_internal_sort (struct sort_cases_pgm *scp, int separate)
 204 {
 205   struct internal_sort *isrt;
 206
 207   isrt = xmalloc (sizeof *isrt);
 208   isrt->results = NULL;
 209
 210   if (!case_source_is_class (vfm_source, &disk_source_class))
 211     {
 212       if (!case_source_is_class (vfm_source, &memory_source_class))
 213         procedure (NULL, NULL, NULL, NULL);
 214
 215       if (case_source_is_class (vfm_source, &memory_source_class))
 216         {
 217           struct case_list *case_list;
 218           struct case_list **case_array;
 219           size_t case_cnt;
 220           int i;
 221
 222           case_cnt = vfm_source_info.ncases;
 223           if (case_cnt == 0)
 224             return isrt;
 225
 226           if (case_cnt > set_max_workspace / sizeof *case_array)
 227             goto error;
 228
 229           case_list = memory_source_get_cases (vfm_source);
 230           case_array = malloc (sizeof *case_array * (case_cnt + 1));
 231           if (case_array == NULL)
 232             goto error;
 233
 234           for (i = 0; case_list != NULL; i++)
 235             {
 236               case_array[i] = case_list;
 237               case_list = case_list->next;
 238             }
 239           assert (i == case_cnt);
 240           case_array[case_cnt] = NULL;
 241
 242           sort (case_array, case_cnt, sizeof *case_array,
 243                 compare_case_lists, scp);
 244
 245           if (!separate)
 246             {
 247               memory_source_set_cases (vfm_source, case_array[0]);
 248               for (i = 1; i <= case_cnt; i++)
 249                 case_array[i - 1]->next = case_array[i];
 250               free (case_array);
 251             }
 252           else
 253             isrt->results = case_array;
 254
 255           return isrt;
 256         }
 257     }
 258
 259  error:
 260   free (isrt);
 261   return NULL;
 262 }
 263
 264 static void
 265 destroy_internal_sort (struct internal_sort *isrt)
 266 {
 267   if (isrt != NULL)
 268     {
 269       free (isrt->results);
 270       free (isrt);
 271     }
 272 }
 273
 274 /* Compares the VAR_CNT variables in VARS[] between the
 275    `case_list's at A and B, and returns a strcmp()-type
 276    result. */
 277 static int
 278 compare_case_lists (const void *a_, const void *b_, void *scp_)
 279 {
 280   struct sort_cases_pgm *scp = scp_;
 281   struct case_list *const *pa = a_;
 282   struct case_list *const *pb = b_;
 283   struct case_list *a = *pa;
 284   struct case_list *b = *pb;
 285
 286   return compare_record (a->c.data, b->c.data, scp);
 287 }
 288 \f
 289 /* External sort. */
 290
 291 /* Maximum order of merge.  If you increase this, then you should
 292    use a heap for comparing cases during merge.  */
 293 #define MAX_MERGE_ORDER         7
 294
 295 /* Minimum total number of records for buffers. */
 296 #define MIN_BUFFER_TOTAL_SIZE_RECS      64
 297
 298 /* Minimum single input buffer size, in bytes and records. */
 299 #define MIN_BUFFER_SIZE_BYTES   4096
 300 #define MIN_BUFFER_SIZE_RECS    16
 301
 302 #if MIN_BUFFER_SIZE_RECS * 2 + 16 > MIN_BUFFER_TOTAL_SIZE_RECS
 303 #error MIN_BUFFER_SIZE_RECS and MIN_BUFFER_TOTAL_SIZE_RECS do not make sense.
 304 #endif
 305
 306 /* An initial run and its length. */
 307 struct initial_run
 308   {
 309     int file_idx;                     /* File index. */
 310     size_t case_cnt;                  /* Number of cases. */
 311   };
 312
 313 /* Sorts initial runs A and B in decending order by length. */
 314 static int
 315 compare_initial_runs (const void *a_, const void *b_, void *aux UNUSED)
 316 {
 317   const struct initial_run *a = a_;
 318   const struct initial_run *b = b_;
 319
 320   return a->case_cnt > b->case_cnt ? -1 : a->case_cnt <b->case_cnt;
 321 }
 322
 323 struct external_sort
 324   {
 325     struct sort_cases_pgm *scp;       /* SORT CASES info. */
 326     struct initial_run *initial_runs; /* Array of initial runs. */
 327     size_t run_cnt, run_cap;          /* Number of runs, allocated capacity. */
 328     char *temp_dir;                   /* Temporary file directory name. */
 329     int next_file_idx;                /* Lowest unused file index. */
 330   };
 331
 332 /* Prototypes for helper functions. */
 333 static void sort_sink_write (struct case_sink *, struct ccase *);
 334 static int write_initial_runs (struct external_sort *, int separate);
 335 static int init_external_sort (struct external_sort *);
 336 static int merge (struct external_sort *);
 337 static void rmdir_temp_dir (struct external_sort *);
 338 static void remove_temp_file (struct external_sort *xsrt, int file_idx);
 339
 340 /* Performs an external sort of the active file according to the
 341    specification in SCP.  Forms initial runs using a heap as a
 342    reservoir.  Determines the optimum merge pattern via Huffman's
 343    method (see Knuth vol. 3, 2nd edition, p. 365-366), and merges
 344    according to that pattern. */
 345 static struct external_sort *
 346 do_external_sort (struct sort_cases_pgm *scp, int separate)
 347 {
 348   struct external_sort *xsrt;
 349   int success = 0;
 350
 351   xsrt = xmalloc (sizeof *xsrt);
 352   xsrt->scp = scp;
 353   if (!init_external_sort (xsrt))
 354     goto done;
 355   if (!write_initial_runs (xsrt, separate))
 356     goto done;
 357   if (!merge (xsrt))
 358     goto done;
 359
 360   success = 1;
 361
 362  done:
 363   if (success)
 364     {
 365       /* Don't destroy anything because we'll need it for reading
 366          the output. */
 367       return xsrt;
 368     }
 369   else
 370     {
 371       destroy_external_sort (xsrt);
 372       return NULL;
 373     }
 374 }
 375
 376 /* Destroys XSRT. */
 377 static void
 378 destroy_external_sort (struct external_sort *xsrt)
 379 {
 380   if (xsrt != NULL)
 381     {
 382       int i;
 383
 384       for (i = 0; i < xsrt->run_cnt; i++)
 385         remove_temp_file (xsrt, xsrt->initial_runs[i].file_idx);
 386       rmdir_temp_dir (xsrt);
 387       free (xsrt->initial_runs);
 388       free (xsrt);
 389     }
 390 }
 391
 392 #ifdef HAVE_MKDTEMP
 393 /* Creates and returns the name of a temporary directory. */
 394 static char *
 395 make_temp_dir (void)
 396 {
 397   const char *parent_dir;
 398   char *temp_dir;
 399
 400   if (getenv ("TMPDIR") != NULL)
 401     parent_dir = getenv ("TMPDIR");
 402   else
 403     parent_dir = P_tmpdir;
 404
 405   temp_dir = xmalloc (strlen (parent_dir) + 32);
 406   sprintf (temp_dir, "%s%cpsppXXXXXX", parent_dir, DIR_SEPARATOR);
 407   if (mkdtemp (temp_dir) == NULL)
 408     {
 409       msg (SE, _("%s: Creating temporary directory: %s."),
 410            temp_dir, strerror (errno));
 411       free (temp_dir);
 412       return NULL;
 413     }
 414   else
 415     return temp_dir;
 416 }
 417 #else /* !HAVE_MKDTEMP */
 418 /* Creates directory DIR. */
 419 static int
 420 do_mkdir (const char *dir)
 421 {
 422 #ifndef __MSDOS__
 423   return mkdir (dir, S_IRWXU);
 424 #else
 425   return mkdir (dir);
 426 #endif
 427 }
 428
 429 /* Creates and returns the name of a temporary directory. */
 430 static char *
 431 make_temp_dir (void)
 432 {
 433   int i;
 434
 435   for (i = 0; i < 100; i++)
 436     {
 437       char temp_dir[L_tmpnam + 1];
 438       if (tmpnam (temp_dir) == NULL)
 439         {
 440           msg (SE, _("Generating temporary directory name failed: %s."),
 441                strerror (errno));
 442           return NULL;
 443         }
 444       else if (do_mkdir (temp_dir) == 0)
 445         return xstrdup (temp_dir);
 446     }
 447
 448   msg (SE, _("Creating temporary directory failed: %s."), strerror (errno));
 449   return NULL;
 450 }
 451 #endif /* !HAVE_MKDTEMP */
 452
 453 /* Sets up to open temporary files. */
 454 static int
 455 init_external_sort (struct external_sort *xsrt)
 456 {
 457   /* Zero. */
 458   xsrt->temp_dir = NULL;
 459   xsrt->next_file_idx = 0;
 460
 461   /* Huffman queue. */
 462   xsrt->run_cap = 512;
 463   xsrt->run_cnt = 0;
 464   xsrt->initial_runs = xmalloc (sizeof *xsrt->initial_runs * xsrt->run_cap);
 465
 466   /* Temporary directory. */
 467   xsrt->temp_dir = make_temp_dir ();
 468   if (xsrt->temp_dir == NULL)
 469     return 0;
 470
 471   return 1;
 472 }
 473
 474
 475 static int
 476 simulate_error (void)
 477 {
 478   static int op_err_cnt = -1;
 479   static int op_cnt;
 480
 481   if (op_err_cnt == -1 || op_cnt++ < op_err_cnt)
 482     return 0;
 483   else
 484     {
 485       errno = 0;
 486       return 1;
 487     }
 488 }
 489
 490 /* Removes the directory created for temporary files, if one
 491    exists. */
 492 static void
 493 rmdir_temp_dir (struct external_sort *xsrt)
 494 {
 495   if (xsrt->temp_dir != NULL && rmdir (xsrt->temp_dir) == -1)
 496     {
 497       msg (SE, _("%s: Error removing directory for temporary files: %s."),
 498            xsrt->temp_dir, strerror (errno));
 499       xsrt->temp_dir = NULL;
 500     }
 501 }
 502
 503 #define TEMP_FILE_NAME_SIZE (L_tmpnam + 32)
 504 static void
 505 get_temp_file_name (struct external_sort *xsrt, int file_idx,
 506                     char filename[TEMP_FILE_NAME_SIZE])
 507 {
 508   assert (xsrt->temp_dir != NULL);
 509   sprintf (filename, "%s%c%04d", xsrt->temp_dir, DIR_SEPARATOR, file_idx);
 510 }
 511
 512 static FILE *
 513 open_temp_file (struct external_sort *xsrt, int file_idx, const char *mode)
 514 {
 515   char temp_file[TEMP_FILE_NAME_SIZE];
 516   FILE *file;
 517
 518   get_temp_file_name (xsrt, file_idx, temp_file);
 519
 520   file = fopen (temp_file, mode);
 521   if (simulate_error () || file == NULL)
 522     msg (SE, _("%s: Error opening temporary file for %s: %s."),
 523          temp_file, mode[0] == 'r' ? "reading" : "writing",
 524          strerror (errno));
 525
 526   return file;
 527 }
 528
 529 static int
 530 close_temp_file (struct external_sort *xsrt, int file_idx, FILE *file)
 531 {
 532   if (file != NULL)
 533     {
 534       char temp_file[TEMP_FILE_NAME_SIZE];
 535       get_temp_file_name (xsrt, file_idx, temp_file);
 536       if (simulate_error () || fclose (file) == EOF)
 537         {
 538           msg (SE, _("%s: Error closing temporary file: %s."),
 539                temp_file, strerror (errno));
 540           return 0;
 541         }
 542     }
 543   return 1;
 544 }
 545
 546 static void
 547 remove_temp_file (struct external_sort *xsrt, int file_idx)
 548 {
 549   if (file_idx != -1)
 550     {
 551       char temp_file[TEMP_FILE_NAME_SIZE];
 552       get_temp_file_name (xsrt, file_idx, temp_file);
 553       if (simulate_error () || remove (temp_file) != 0)
 554         msg (SE, _("%s: Error removing temporary file: %s."),
 555              temp_file, strerror (errno));
 556     }
 557 }
 558
 559 static int
 560 write_temp_file (struct external_sort *xsrt, int file_idx,
 561                  FILE *file, const void *data, size_t size)
 562 {
 563   if (!simulate_error () && fwrite (data, size, 1, file) == 1)
 564     return 1;
 565   else
 566     {
 567       char temp_file[TEMP_FILE_NAME_SIZE];
 568       get_temp_file_name (xsrt, file_idx, temp_file);
 569       msg (SE, _("%s: Error writing temporary file: %s."),
 570            temp_file, strerror (errno));
 571       return 0;
 572     }
 573 }
 574
 575 static int
 576 read_temp_file (struct external_sort *xsrt, int file_idx,
 577                 FILE *file, void *data, size_t size)
 578 {
 579   if (!simulate_error () && fread (data, size, 1, file) == 1)
 580     return 1;
 581   else
 582     {
 583       char temp_file[TEMP_FILE_NAME_SIZE];
 584       get_temp_file_name (xsrt, file_idx, temp_file);
 585       if (ferror (file))
 586         msg (SE, _("%s: Error reading temporary file: %s."),
 587              temp_file, strerror (errno));
 588       else
 589         msg (SE, _("%s: Unexpected end of temporary file."),
 590              temp_file);
 591       return 0;
 592     }
 593 }
 594 \f
 595 /* Replacement selection. */
 596
 597 /* Pairs a record with a run number. */
 598 struct record_run
 599   {
 600     int run;                    /* Run number of case. */
 601     struct case_list *record;   /* Case data. */
 602   };
 603
 604 struct initial_run_state
 605   {
 606     struct external_sort *xsrt;
 607
 608     /* Reservoir. */
 609     struct record_run *records; /* Records arranged as a heap. */
 610     size_t record_cnt;          /* Current number of records. */
 611     size_t record_cap;          /* Capacity for records. */
 612     struct case_list *free_list;/* Cases not in heap. */
 613
 614     /* Run currently being output. */
 615     int file_idx;               /* Temporary file number. */
 616     size_t case_cnt;            /* Number of cases so far. */
 617     FILE *output_file;          /* Output file. */
 618     struct case_list *last_output;/* Record last output. */
 619
 620     int okay;                   /* Zero if an error has been encountered. */
 621   };
 622
 623 static void destroy_initial_run_state (struct initial_run_state *irs);
 624 static int allocate_cases (struct initial_run_state *);
 625 static struct case_list *grab_case (struct initial_run_state *);
 626 static void release_case (struct initial_run_state *, struct case_list *);
 627 static void output_record (struct initial_run_state *irs);
 628 static void start_run (struct initial_run_state *irs);
 629 static void end_run (struct initial_run_state *irs);
 630 static int compare_record_run (const struct record_run *,
 631                                const struct record_run *,
 632                                struct sort_cases_pgm *);
 633 static int compare_record_run_minheap (const void *, const void *, void *);
 634
 635 static int
 636 write_initial_runs (struct external_sort *xsrt, int separate)
 637 {
 638   struct initial_run_state *irs;
 639   int success = 0;
 640
 641   /* Allocate memory for cases. */
 642   irs = xmalloc (sizeof *irs);
 643   irs->xsrt = xsrt;
 644   irs->records = NULL;
 645   irs->record_cnt = irs->record_cap = 0;
 646   irs->free_list = NULL;
 647   irs->output_file = NULL;
 648   irs->last_output = NULL;
 649   irs->file_idx = 0;
 650   irs->case_cnt = 0;
 651   irs->okay = 1;
 652   if (!allocate_cases (irs))
 653     goto done;
 654
 655   /* Create case sink. */
 656   if (!separate)
 657     {
 658       if (vfm_sink)
 659         vfm_sink->class->destroy (vfm_sink);
 660       vfm_sink = create_case_sink (&sort_sink_class, irs);
 661       xsrt->scp->ref_cnt++;
 662     }
 663
 664   /* Create initial runs. */
 665   start_run (irs);
 666   procedure (NULL, NULL, NULL, NULL);
 667   while (irs->record_cnt > 0 && irs->okay)
 668     output_record (irs);
 669   end_run (irs);
 670
 671   success = irs->okay;
 672
 673  done:
 674   destroy_initial_run_state (irs);
 675
 676   return success;
 677 }
 678
 679 /* Add a single case to an initial run. */
 680 static void
 681 sort_sink_write (struct case_sink *sink, struct ccase *c)
 682 {
 683   struct initial_run_state *irs = sink->aux;
 684   struct record_run *new_record_run;
 685
 686   if (!irs->okay)
 687     return;
 688
 689   /* Compose record_run for this run and add to heap. */
 690   assert (irs->record_cnt < irs->record_cap);
 691   new_record_run = irs->records + irs->record_cnt++;
 692   new_record_run->record = grab_case (irs);
 693   memcpy (new_record_run->record->c.data, c->data, vfm_sink_info.case_size);
 694   new_record_run->run = irs->file_idx;
 695   if (irs->last_output != NULL
 696       && compare_record (c->data, irs->last_output->c.data,
 697                          irs->xsrt->scp) < 0)
 698     new_record_run->run = irs->xsrt->next_file_idx;
 699   push_heap (irs->records, irs->record_cnt, sizeof *irs->records,
 700              compare_record_run_minheap, irs->xsrt->scp);
 701
 702   /* Output a record if the reservoir is full. */
 703   if (irs->record_cnt == irs->record_cap && irs->okay)
 704     output_record (irs);
 705 }
 706
 707 static void
 708 destroy_initial_run_state (struct initial_run_state *irs)
 709 {
 710   struct case_list *iter, *next;
 711   int i;
 712
 713   if (irs == NULL)
 714     return;
 715
 716   /* Release cases to free list. */
 717   for (i = 0; i < irs->record_cnt; i++)
 718     release_case (irs, irs->records[i].record);
 719   if (irs->last_output != NULL)
 720     release_case (irs, irs->last_output);
 721
 722   /* Free cases in free list. */
 723   for (iter = irs->free_list; iter != NULL; iter = next)
 724     {
 725       next = iter->next;
 726       free (iter);
 727     }
 728
 729   free (irs->records);
 730   free (irs);
 731
 732   if (irs->output_file != NULL)
 733     close_temp_file (irs->xsrt, irs->file_idx, irs->output_file);
 734 }
 735
 736 /* Allocates room for lots of cases as a buffer. */
 737 static int
 738 allocate_cases (struct initial_run_state *irs)
 739 {
 740   size_t case_size;     /* Size of one case, in bytes. */
 741   int approx_case_cost; /* Approximate memory cost of one case in bytes. */
 742   int max_cases;        /* Maximum number of cases to allocate. */
 743   int i;
 744
 745   /* Allocate as many cases as we can within the workspace
 746      limit. */
 747   case_size = dict_get_case_size (default_dict);
 748   approx_case_cost = (sizeof *irs->records
 749                       + sizeof *irs->free_list
 750                       + case_size
 751                       + 4 * sizeof (void *));
 752   max_cases = set_max_workspace / approx_case_cost;
 753   irs->records = malloc (sizeof *irs->records * max_cases);
 754   for (i = 0; i < max_cases; i++)
 755     {
 756       struct case_list *c;
 757       c = malloc (sizeof *c + case_size - sizeof (union value));
 758       if (c == NULL)
 759         {
 760           max_cases = i;
 761           break;
 762         }
 763       release_case (irs, c);
 764     }
 765
 766   /* irs->records gets all but one of the allocated cases.
 767      The extra is used for last_output. */
 768   irs->record_cap = max_cases - 1;
 769
 770   /* Fail if we didn't allocate an acceptable number of cases. */
 771   if (irs->records == NULL || max_cases < MIN_BUFFER_TOTAL_SIZE_RECS)
 772     {
 773       msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "
 774                  "cases of %d bytes each.  (PSPP workspace is currently "
 775                  "restricted to a maximum of %d KB.)"),
 776            MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, set_max_workspace / 1024);
 777       return 0;
 778     }
 779   return 1;
 780 }
 781
 782 /* Compares the VAR_CNT variables in VARS[] between the `value's at
 783    A and B, and returns a strcmp()-type result. */
 784 static int
 785 compare_record (const union value *a, const union value *b,
 786                 const struct sort_cases_pgm *scp)
 787 {
 788   int i;
 789
 790   assert (a != NULL);
 791   assert (b != NULL);
 792
 793   for (i = 0; i < scp->var_cnt; i++)
 794     {
 795       struct variable *v = scp->vars[i];
 796       int fv = v->fv;
 797       int result;
 798
 799       if (v->type == NUMERIC)
 800         {
 801           double af = a[fv].f;
 802           double bf = b[fv].f;
 803
 804           result = af < bf ? -1 : af > bf;
 805         }
 806       else
 807         result = memcmp (a[fv].s, b[fv].s, v->width);
 808
 809       if (result != 0)
 810         {
 811           if (scp->dirs[i] == SRT_DESCEND)
 812             result = -result;
 813           return result;
 814         }
 815     }
 816
 817   return 0;
 818 }
 819
 820 static int
 821 compare_record_run (const struct record_run *a,
 822                     const struct record_run *b,
 823                     struct sort_cases_pgm *scp)
 824 {
 825   if (a->run != b->run)
 826     return a->run > b->run ? 1 : -1;
 827   else
 828     return compare_record (a->record->c.data, b->record->c.data, scp);
 829 }
 830
 831 static int
 832 compare_record_run_minheap (const void *a, const void *b, void *scp)
 833 {
 834   return -compare_record_run (a, b, scp);
 835 }
 836
 837 /* Begins a new initial run, specifically its output file. */
 838 static void
 839 start_run (struct initial_run_state *irs)
 840 {
 841   irs->file_idx = irs->xsrt->next_file_idx++;
 842   irs->case_cnt = 0;
 843   irs->output_file = open_temp_file (irs->xsrt, irs->file_idx, "wb");
 844   if (irs->output_file == NULL)
 845     irs->okay = 0;
 846   if (irs->last_output != NULL)
 847     {
 848       release_case (irs, irs->last_output);
 849       irs->last_output = NULL;
 850     }
 851 }
 852
 853 /* Ends the current initial run.  */
 854 static void
 855 end_run (struct initial_run_state *irs)
 856 {
 857   struct external_sort *xsrt = irs->xsrt;
 858
 859   /* Record initial run. */
 860   if (xsrt->run_cnt >= xsrt->run_cap)
 861     {
 862       xsrt->run_cap *= 2;
 863       xsrt->initial_runs
 864         = xrealloc (xsrt->initial_runs,
 865                     sizeof *xsrt->initial_runs * xsrt->run_cap);
 866     }
 867   xsrt->initial_runs[xsrt->run_cnt].file_idx = irs->file_idx;
 868   xsrt->initial_runs[xsrt->run_cnt].case_cnt = irs->case_cnt;
 869   xsrt->run_cnt++;
 870
 871   /* Close file handle. */
 872   if (irs->output_file != NULL
 873       && !close_temp_file (irs->xsrt, irs->file_idx, irs->output_file))
 874     irs->okay = 0;
 875   irs->output_file = NULL;
 876 }
 877
 878 static void
 879 output_record (struct initial_run_state *irs)
 880 {
 881   struct record_run *record_run;
 882   struct ccase *out_case;
 883
 884   /* Extract minimum case from heap. */
 885   assert (irs->record_cnt > 0);
 886   pop_heap (irs->records, irs->record_cnt--, sizeof *irs->records,
 887             compare_record_run_minheap, irs->xsrt->scp);
 888   record_run = irs->records + irs->record_cnt;
 889
 890   /* Bail if an error has occurred. */
 891   if (!irs->okay)
 892     return;
 893
 894   /* Obtain case data to write to disk. */
 895   out_case = &record_run->record->c;
 896   if (compaction_necessary)
 897     {
 898       compact_case (compaction_case, out_case);
 899       out_case = compaction_case;
 900     }
 901
 902   /* Start new run if necessary. */
 903   assert (record_run->run == irs->file_idx
 904           || record_run->run == irs->xsrt->next_file_idx);
 905   if (record_run->run != irs->file_idx)
 906     {
 907       end_run (irs);
 908       start_run (irs);
 909     }
 910   assert (record_run->run == irs->file_idx);
 911   irs->case_cnt++;
 912
 913   /* Write to disk. */
 914   if (irs->output_file != NULL
 915       && !write_temp_file (irs->xsrt, irs->file_idx, irs->output_file,
 916                            out_case->data,
 917                            sizeof *out_case->data * compaction_nval))
 918     irs->okay = 0;
 919
 920   /* This record becomes last_output. */
 921   if (irs->last_output != NULL)
 922     release_case (irs, irs->last_output);
 923   irs->last_output = record_run->record;
 924 }
 925
 926 static struct case_list *
 927 grab_case (struct initial_run_state *irs)
 928 {
 929   struct case_list *c;
 930
 931   assert (irs != NULL);
 932   assert (irs->free_list != NULL);
 933
 934   c = irs->free_list;
 935   irs->free_list = c->next;
 936   return c;
 937 }
 938
 939 static void
 940 release_case (struct initial_run_state *irs, struct case_list *c)
 941 {
 942   assert (irs != NULL);
 943   assert (c != NULL);
 944
 945   c->next = irs->free_list;
 946   irs->free_list = c;
 947 }
 948 \f
 949 /* Merging. */
 950
 951 struct merge_state
 952   {
 953     struct external_sort *xsrt; /* External sort state. */
 954     struct ccase **cases;       /* Buffers. */
 955     size_t case_cnt;            /* Number of buffers. */
 956   };
 957
 958 struct run;
 959 static int merge_once (struct merge_state *,
 960                        const struct initial_run[], size_t,
 961                        struct initial_run *);
 962 static int fill_run_buffer (struct merge_state *, struct run *);
 963 static int mod (int, int);
 964
 965 /* Performs a series of P-way merges of initial runs
 966    method. */
 967 static int
 968 merge (struct external_sort *xsrt)
 969 {
 970   struct merge_state mrg;       /* State of merge. */
 971   size_t case_size;             /* Size of one case, in bytes. */
 972   size_t approx_case_cost;      /* Approximate memory cost of one case. */
 973   int max_order;                /* Maximum order of merge. */
 974   size_t dummy_run_cnt;         /* Number of dummy runs to insert. */
 975   int success = 0;
 976   int i;
 977
 978   mrg.xsrt = xsrt;
 979
 980   /* Allocate as many cases as possible into cases. */
 981   case_size = dict_get_case_size (default_dict);
 982   approx_case_cost = sizeof *mrg.cases + case_size + 4 * sizeof (void *);
 983   mrg.case_cnt = set_max_workspace / approx_case_cost;
 984   mrg.cases = malloc (sizeof *mrg.cases * mrg.case_cnt);
 985   if (mrg.cases == NULL)
 986     goto done;
 987   for (i = 0; i < mrg.case_cnt; i++)
 988     {
 989       mrg.cases[i] = malloc (case_size);
 990       if (mrg.cases[i] == NULL)
 991         {
 992           mrg.case_cnt = i;
 993           break;
 994         }
 995     }
 996   if (mrg.case_cnt < MIN_BUFFER_TOTAL_SIZE_RECS)
 997     {
 998       msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "
 999                  "cases of %d bytes each.  (PSPP workspace is currently "
1000                  "restricted to a maximum of %d KB.)"),
1001            MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, set_max_workspace / 1024);
1002       return 0;
1003     }
1004
1005   /* Determine maximum order of merge. */
1006   max_order = MAX_MERGE_ORDER;
1007   if (mrg.case_cnt / max_order < MIN_BUFFER_SIZE_RECS)
1008     max_order = mrg.case_cnt / MIN_BUFFER_SIZE_RECS;
1009   else if (mrg.case_cnt / max_order * case_size < MIN_BUFFER_SIZE_BYTES)
1010     max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES / case_size);
1011   if (max_order < 2)
1012     max_order = 2;
1013   if (max_order > xsrt->run_cnt)
1014     max_order = xsrt->run_cnt;
1015
1016   /* Repeatedly merge the P shortest existing runs until only one run
1017      is left. */
1018   make_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
1019              compare_initial_runs, NULL);
1020   dummy_run_cnt = mod (1 - (int) xsrt->run_cnt, max_order - 1);
1021   assert (max_order == 1
1022           || (xsrt->run_cnt + dummy_run_cnt) % (max_order - 1) == 1);
1023   while (xsrt->run_cnt > 1)
1024     {
1025       struct initial_run output_run;
1026       int order;
1027       int i;
1028
1029       /* Choose order of merge (max_order after first merge). */
1030       order = max_order - dummy_run_cnt;
1031       dummy_run_cnt = 0;
1032
1033       /* Choose runs to merge. */
1034       assert (xsrt->run_cnt >= order);
1035       for (i = 0; i < order; i++)
1036         pop_heap (xsrt->initial_runs, xsrt->run_cnt--,
1037                   sizeof *xsrt->initial_runs,
1038                   compare_initial_runs, NULL);
1039
1040       /* Merge runs. */
1041       if (!merge_once (&mrg, xsrt->initial_runs + xsrt->run_cnt, order,
1042                        &output_run))
1043         goto done;
1044
1045       /* Add output run to heap. */
1046       xsrt->initial_runs[xsrt->run_cnt++] = output_run;
1047       push_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
1048                  compare_initial_runs, NULL);
1049     }
1050
1051   /* Exactly one run is left, which contains the entire sorted
1052      file.  We could use it to find a total case count. */
1053   assert (xsrt->run_cnt == 1);
1054
1055   success = 1;
1056
1057  done:
1058   for (i = 0; i < mrg.case_cnt; i++)
1059     free (mrg.cases[i]);
1060   free (mrg.cases);
1061
1062   return success;
1063 }
1064
1065 /* Modulo function as defined by Knuth. */
1066 static int
1067 mod (int x, int y)
1068 {
1069   if (y == 0)
1070     return x;
1071   else if (x == 0)
1072     return 0;
1073   else if (x > 0 && y > 0)
1074     return x % y;
1075   else if (x < 0 && y > 0)
1076     return y - (-x) % y;
1077
1078   assert (0);
1079 }
1080
1081 /* A run of data for use in merging. */
1082 struct run
1083   {
1084     FILE *file;                 /* File that contains run. */
1085     int file_idx;               /* Index of file that contains run. */
1086     struct ccase **buffer;      /* Case buffer. */
1087     struct ccase **buffer_head; /* First unconsumed case in buffer. */
1088     struct ccase **buffer_tail; /* One past last unconsumed case in buffer. */
1089     size_t buffer_cap;          /* Number of cases buffer can hold. */
1090     size_t unread_case_cnt;     /* Number of cases not yet read. */
1091   };
1092
1093 /* Merges the RUN_CNT initial runs specified in INPUT_RUNS into a
1094    new run.  Returns nonzero only if successful.  Adds an entry
1095    to MRG->xsrt->runs for the output file if and only if the
1096    output file is actually created.  Always deletes all the input
1097    files. */
1098 static int
1099 merge_once (struct merge_state *mrg,
1100             const struct initial_run input_runs[],
1101             size_t run_cnt,
1102             struct initial_run *output_run)
1103 {
1104   struct run runs[MAX_MERGE_ORDER];
1105   FILE *output_file = NULL;
1106   size_t case_size;
1107   int success = 0;
1108   int i;
1109
1110   /* Initialize runs[]. */
1111   for (i = 0; i < run_cnt; i++)
1112     {
1113       runs[i].file = NULL;
1114       runs[i].file_idx = input_runs[i].file_idx;
1115       runs[i].buffer = mrg->cases + mrg->case_cnt / run_cnt * i;
1116       runs[i].buffer_head = runs[i].buffer;
1117       runs[i].buffer_tail = runs[i].buffer;
1118       runs[i].buffer_cap = mrg->case_cnt / run_cnt;
1119       runs[i].unread_case_cnt = input_runs[i].case_cnt;
1120     }
1121
1122   /* Open input files. */
1123   for (i = 0; i < run_cnt; i++)
1124     {
1125       runs[i].file = open_temp_file (mrg->xsrt, runs[i].file_idx, "rb");
1126       if (runs[i].file == NULL)
1127         goto error;
1128     }
1129
1130   /* Create output file and count cases to be output. */
1131   output_run->file_idx = mrg->xsrt->next_file_idx++;
1132   output_run->case_cnt = 0;
1133   for (i = 0; i < run_cnt; i++)
1134     output_run->case_cnt += input_runs[i].case_cnt;
1135   output_file = open_temp_file (mrg->xsrt, output_run->file_idx, "wb");
1136   if (output_file == NULL)
1137     goto error;
1138
1139   /* Prime buffers. */
1140   for (i = 0; i < run_cnt; i++)
1141     if (!fill_run_buffer (mrg, runs + i))
1142       goto error;
1143
1144   /* Merge. */
1145   case_size = dict_get_case_size (default_dict);
1146   while (run_cnt > 0)
1147     {
1148       struct run *min_run;
1149
1150       /* Find minimum. */
1151       min_run = runs;
1152       for (i = 1; i < run_cnt; i++)
1153         if (compare_record ((*runs[i].buffer_head)->data,
1154                             (*min_run->buffer_head)->data,
1155                             mrg->xsrt->scp) < 0)
1156           min_run = runs + i;
1157
1158       /* Write minimum to output file. */
1159       if (!write_temp_file (mrg->xsrt, min_run->file_idx, output_file,
1160                             (*min_run->buffer_head)->data, case_size))
1161         goto error;
1162
1163       /* Remove case from buffer. */
1164       if (++min_run->buffer_head >= min_run->buffer_tail)
1165         {
1166           /* Buffer is empty.  Fill from file. */
1167           if (!fill_run_buffer (mrg, min_run))
1168             goto error;
1169
1170           /* If buffer is still empty, delete its run. */
1171           if (min_run->buffer_head >= min_run->buffer_tail)
1172             {
1173               close_temp_file (mrg->xsrt, min_run->file_idx, min_run->file);
1174               remove_temp_file (mrg->xsrt, min_run->file_idx);
1175               *min_run = runs[--run_cnt];
1176
1177               /* We could donate the now-unused buffer space to
1178                  other runs. */
1179             }
1180         }
1181     }
1182
1183   /* Close output file.  */
1184   close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1185
1186   return 1;
1187
1188  error:
1189   /* Close and remove output file.  */
1190   if (output_file != NULL)
1191     {
1192       close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1193       remove_temp_file (mrg->xsrt, output_run->file_idx);
1194     }
1195
1196   /* Close and remove any remaining input runs. */
1197   for (i = 0; i < run_cnt; i++)
1198     {
1199       close_temp_file (mrg->xsrt, runs[i].file_idx, runs[i].file);
1200       remove_temp_file (mrg->xsrt, runs[i].file_idx);
1201     }
1202
1203   return success;
1204 }
1205
1206 /* Reads as many cases as possible into RUN's buffer.
1207    Reads nonzero unless a disk error occurs. */
1208 static int
1209 fill_run_buffer (struct merge_state *mrg, struct run *run)
1210 {
1211   run->buffer_head = run->buffer_tail = run->buffer;
1212   while (run->unread_case_cnt > 0
1213          && run->buffer_tail < run->buffer + run->buffer_cap)
1214     {
1215       if (!read_temp_file (mrg->xsrt, run->file_idx, run->file,
1216                            (*run->buffer_tail)->data,
1217                            dict_get_case_size (default_dict)))
1218         return 0;
1219
1220       run->unread_case_cnt--;
1221       run->buffer_tail++;
1222     }
1223
1224   return 1;
1225 }
1226 \f
1227 static void
1228 sort_sink_destroy (struct case_sink *sink UNUSED)
1229 {
1230   assert (0);
1231 }
1232
1233 static struct case_source *
1234 sort_sink_make_source (struct case_sink *sink)
1235 {
1236   struct initial_run_state *irs = sink->aux;
1237
1238   return create_case_source (&sort_source_class, irs->xsrt->scp);
1239 }
1240
1241 const struct case_sink_class sort_sink_class =
1242   {
1243     "SORT CASES",
1244     NULL,
1245     sort_sink_write,
1246     sort_sink_destroy,
1247     sort_sink_make_source,
1248   };
1249 \f
1250 /* Reads all the records from the source stream and passes them
1251    to write_case(). */
1252 static void
1253 sort_source_read (struct case_source *source,
1254                   write_case_func *write_case, write_case_data wc_data)
1255 {
1256   struct sort_cases_pgm *scp = source->aux;
1257
1258   read_sort_output (scp, write_case, wc_data);
1259 }
1260
1261 void read_internal_sort_output (struct internal_sort *isrt,
1262                                 write_case_func *write_case,
1263                                 write_case_data wc_data);
1264 void read_external_sort_output (struct external_sort *xsrt,
1265                                 write_case_func *write_case,
1266                                 write_case_data wc_data);
1267
1268 /* Reads all the records from the output stream and passes them to the
1269    function provided, which must have an interface identical to
1270    write_case(). */
1271 void
1272 read_sort_output (struct sort_cases_pgm *scp,
1273                   write_case_func *write_case, write_case_data wc_data)
1274 {
1275   assert ((scp->isrt != NULL) + (scp->xsrt != NULL) <= 1);
1276   if (scp->isrt != NULL)
1277     read_internal_sort_output (scp->isrt, write_case, wc_data);
1278   else if (scp->xsrt != NULL)
1279     read_external_sort_output (scp->xsrt, write_case, wc_data);
1280   else
1281     {
1282       /* No results.  Probably an external sort that failed. */
1283     }
1284 }
1285
1286 void
1287 read_internal_sort_output (struct internal_sort *isrt,
1288                            write_case_func *write_case,
1289                            write_case_data wc_data)
1290 {
1291   struct ccase *save_temp_case = temp_case;
1292   struct case_list **p;
1293
1294   for (p = isrt->results; *p; p++)
1295     {
1296       temp_case = &(*p)->c;
1297       write_case (wc_data);
1298     }
1299   free (isrt->results);
1300
1301   temp_case = save_temp_case;
1302 }
1303
1304 void
1305 read_external_sort_output (struct external_sort *xsrt,
1306                            write_case_func *write_case,
1307                            write_case_data wc_data)
1308 {
1309   FILE *file;
1310   int file_idx;
1311   int i;
1312
1313   assert (xsrt->run_cnt == 1);
1314   file_idx = xsrt->initial_runs[0].file_idx;
1315
1316   file = open_temp_file (xsrt, file_idx, "rb");
1317   if (file == NULL)
1318     {
1319       err_failure ();
1320       return;
1321     }
1322
1323   for (i = 0; i < vfm_source_info.ncases; i++)
1324     {
1325       if (!read_temp_file (xsrt, file_idx, file,
1326                           temp_case, vfm_source_info.case_size))
1327         {
1328           err_failure ();
1329           break;
1330         }
1331
1332       if (!write_case (wc_data))
1333         break;
1334     }
1335 }
1336
1337 static void
1338 sort_source_destroy (struct case_source *source)
1339 {
1340   struct sort_cases_pgm *scp = source->aux;
1341
1342   destroy_sort_cases_pgm (scp);
1343 }
1344
1345 const struct case_source_class sort_source_class =
1346   {
1347     "SORT CASES",
1348     sort_source_read,
1349     sort_source_destroy,
1350   };