Fix assertion for proper Huffman merge pattern: 0 == 1 modulo 1.
[pspp] / 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 "error.h"
23 #include <stdio.h>
24 #include <stdlib.h>
25 #include <errno.h>
26 #include "algorithm.h"
27 #include "alloc.h"
28 #include "case.h"
29 #include "casefile.h"
30 #include "command.h"
31 #include "error.h"
32 #include "expr.h"
33 #include "lexer.h"
34 #include "misc.h"
35 #include "settings.h"
36 #include "str.h"
37 #include "var.h"
38 #include "vfm.h"
39 #include "vfmP.h"
40
41 #if HAVE_UNISTD_H
42 #include <unistd.h>
43 #endif
44
45 #if HAVE_SYS_TYPES_H
46 #include <sys/types.h>
47 #endif
48
49 #if HAVE_SYS_STAT_H
50 #include <sys/stat.h>
51 #endif
52
53 #include "debug-print.h"
54
55 /* Sort direction. */
56 enum sort_direction
57   {
58     SRT_ASCEND,                 /* A, B, C, ..., X, Y, Z. */
59     SRT_DESCEND                 /* Z, Y, X, ..., C, B, A. */
60   };
61
62 /* A sort criterion. */
63 struct sort_criterion
64   {
65     int fv;                     /* Variable data index. */
66     int width;                  /* 0=numeric, otherwise string widthe. */
67     enum sort_direction dir;    /* Sort direction. */
68   };
69
70 /* A set of sort criteria. */
71 struct sort_criteria 
72   {
73     struct sort_criterion *crits;
74     size_t crit_cnt;
75   };
76
77 static int compare_case_dblptrs (const void *, const void *, void *);
78 static int compare_record (const struct ccase *, const struct ccase *,
79                            const struct sort_criteria *);
80 static struct casefile *do_internal_sort (struct casereader *,
81                                           const struct sort_criteria *);
82 static struct casefile *do_external_sort (struct casereader *,
83                                           const struct sort_criteria *);
84
85 /* Performs the SORT CASES procedures. */
86 int
87 cmd_sort_cases (void)
88 {
89   struct sort_criteria *criteria;
90   int success;
91
92   lex_match (T_BY);
93
94   criteria = sort_parse_criteria (default_dict, NULL, NULL);
95   if (criteria == NULL)
96     return CMD_FAILURE;
97
98   success = sort_active_file_in_place (criteria);
99   sort_destroy_criteria (criteria);
100   return success ? lex_end_of_command () : CMD_FAILURE;
101 }
102
103 /* Gets ready to sort the active file, either in-place or to a
104    separate casefile. */
105 static void
106 prepare_to_sort_active_file (void) 
107 {
108   /* Cancel temporary transformations and PROCESS IF. */
109   if (temporary != 0)
110     cancel_temporary (); 
111   expr_free (process_if_expr);
112   process_if_expr = NULL;
113
114   /* Make sure source cases are in a storage source. */
115   procedure (NULL, NULL);
116   assert (case_source_is_class (vfm_source, &storage_source_class));
117 }
118
119 /* Sorts the active file in-place according to CRITERIA.
120    Returns nonzero if successful. */
121 int
122 sort_active_file_in_place (const struct sort_criteria *criteria) 
123 {
124   struct casefile *src, *dst;
125   
126   prepare_to_sort_active_file ();
127
128   src = storage_source_get_casefile (vfm_source);
129   dst = sort_execute (casefile_get_destructive_reader (src), criteria);
130   free_case_source (vfm_source);
131   vfm_source = NULL;
132
133   if (dst == NULL) 
134     return 0;
135
136   vfm_source = storage_source_create (dst, default_dict);
137   return 1;
138 }
139
140 /* Sorts the active file to a separate casefile.  If successful,
141    returns the sorted casefile.  Returns a null pointer on
142    failure. */
143 struct casefile *
144 sort_active_file_to_casefile (const struct sort_criteria *criteria) 
145 {
146   struct casefile *src;
147   
148   prepare_to_sort_active_file ();
149
150   src = storage_source_get_casefile (vfm_source);
151   return sort_execute (casefile_get_reader (src), criteria);
152 }
153
154 /* Parses a list of sort keys and returns a struct sort_cases_pgm
155    based on it.  Returns a null pointer on error. */
156 struct sort_criteria *
157 sort_parse_criteria (const struct dictionary *dict,
158                      struct variable ***vars, int *var_cnt)
159 {
160   struct sort_criteria *criteria;
161   struct variable **local_vars = NULL;
162   size_t local_var_cnt;
163
164   assert ((vars == NULL) == (var_cnt == NULL));
165   if (vars == NULL) 
166     {
167       vars = &local_vars;
168       var_cnt = &local_var_cnt;
169     }
170
171   criteria = xmalloc (sizeof *criteria);
172   criteria->crits = NULL;
173   criteria->crit_cnt = 0;
174
175   *vars = NULL;
176   *var_cnt = 0;
177
178   do
179     {
180       int prev_var_cnt = *var_cnt;
181       enum sort_direction direction;
182
183       /* Variables. */
184       if (!parse_variables (dict, vars, var_cnt,
185                             PV_NO_DUPLICATE | PV_APPEND | PV_NO_SCRATCH))
186         goto error;
187
188       /* Sort direction. */
189       if (lex_match ('('))
190         {
191           if (lex_match_id ("D") || lex_match_id ("DOWN"))
192             direction = SRT_DESCEND;
193           else if (lex_match_id ("A") || lex_match_id ("UP"))
194             direction = SRT_ASCEND;
195           else
196             {
197               msg (SE, _("`A' or `D' expected inside parentheses."));
198               goto error;
199             }
200           if (!lex_match (')'))
201             {
202               msg (SE, _("`)' expected."));
203               goto error;
204             }
205         }
206       else
207         direction = SRT_ASCEND;
208
209       criteria->crits = xrealloc (criteria->crits,
210                                   sizeof *criteria->crits * *var_cnt);
211       criteria->crit_cnt = *var_cnt;
212       for (; prev_var_cnt < criteria->crit_cnt; prev_var_cnt++) 
213         {
214           struct sort_criterion *c = &criteria->crits[prev_var_cnt];
215           c->fv = (*vars)[prev_var_cnt]->fv;
216           c->width = (*vars)[prev_var_cnt]->width;
217           c->dir = direction;
218         }
219     }
220   while (token != '.' && token != '/');
221
222   free (local_vars);
223   return criteria;
224
225  error:
226   free (local_vars);
227   sort_destroy_criteria (criteria);
228   return NULL;
229 }
230
231 /* Destroys a SORT CASES program. */
232 void
233 sort_destroy_criteria (struct sort_criteria *criteria) 
234 {
235   if (criteria != NULL) 
236     {
237       free (criteria->crits);
238       free (criteria);
239     }
240 }
241
242 /* Reads all the cases from READER, which is destroyed.  Sorts
243    the cases according to CRITERIA.  Returns the sorted cases in
244    a newly created casefile. */
245 struct casefile *
246 sort_execute (struct casereader *reader, const struct sort_criteria *criteria)
247 {
248   struct casefile *output;
249
250   output = do_internal_sort (reader, criteria);
251   if (output == NULL)
252     output = do_external_sort (reader, criteria);
253   casereader_destroy (reader);
254   return output;
255 }
256 \f
257 /* If the data is in memory, do an internal sort and return a new
258    casefile for the data. */
259 static struct casefile *
260 do_internal_sort (struct casereader *reader,
261                   const struct sort_criteria *criteria)
262 {
263   const struct casefile *src;
264   struct casefile *dst;
265   struct ccase *cases, **case_ptrs;
266   unsigned long case_cnt;
267
268   src = casereader_get_casefile (reader);
269   if (casefile_get_case_cnt (src) > 1 && !casefile_in_core (src))
270     return NULL;
271       
272   case_cnt = casefile_get_case_cnt (src);
273   cases = malloc (sizeof *cases * case_cnt);
274   case_ptrs = malloc (sizeof *case_ptrs * case_cnt);
275   if ((cases != NULL && case_ptrs != NULL) || case_cnt == 0) 
276     {
277       unsigned long case_idx;
278       
279       for (case_idx = 0; case_idx < case_cnt; case_idx++) 
280         {
281           int success = casereader_read_xfer (reader, &cases[case_idx]);
282           assert (success);
283           case_ptrs[case_idx] = &cases[case_idx];
284         }
285
286       sort (case_ptrs, case_cnt, sizeof *case_ptrs, compare_case_dblptrs,
287             (void *) criteria);
288       
289       dst = casefile_create (casefile_get_value_cnt (src));
290       for (case_idx = 0; case_idx < case_cnt; case_idx++) 
291         casefile_append_xfer (dst, case_ptrs[case_idx]);
292     }
293   else
294     dst = NULL;
295   
296   free (case_ptrs);
297   free (cases);
298
299   return dst;
300 }
301
302 /* Compares the variables specified by CRITERIA between the cases
303    at A and B, and returns a strcmp()-type result. */
304 static int
305 compare_case_dblptrs (const void *a_, const void *b_, void *criteria_)
306 {
307   struct sort_criteria *criteria = criteria_;
308   struct ccase *const *pa = a_;
309   struct ccase *const *pb = b_;
310   struct ccase *a = *pa;
311   struct ccase *b = *pb;
312  
313   return compare_record (a, b, criteria);
314 }
315 \f
316 /* External sort. */
317
318 /* Maximum order of merge.  If you increase this, then you should
319    use a heap for comparing cases during merge.  */
320 #define MAX_MERGE_ORDER         7
321
322 /* Minimum total number of records for buffers. */
323 #define MIN_BUFFER_TOTAL_SIZE_RECS      64
324
325 /* Minimum single input buffer size, in bytes and records. */
326 #define MIN_BUFFER_SIZE_BYTES   4096
327 #define MIN_BUFFER_SIZE_RECS    16
328
329 #if MIN_BUFFER_SIZE_RECS * 2 + 16 > MIN_BUFFER_TOTAL_SIZE_RECS
330 #error MIN_BUFFER_SIZE_RECS and MIN_BUFFER_TOTAL_SIZE_RECS do not make sense.
331 #endif
332
333 /* Sorts initial runs A and B in decending order by length. */
334 static int
335 compare_initial_runs (const void *a_, const void *b_, void *aux UNUSED) 
336 {
337   const struct casefile *a = a_;
338   const struct casefile *b = b_;
339   unsigned long a_case_cnt = casefile_get_case_cnt (a);
340   unsigned long b_case_cnt = casefile_get_case_cnt (b);
341   
342   return a_case_cnt > b_case_cnt ? -1 : a_case_cnt < b_case_cnt;
343 }
344
345 /* Results of an external sort. */
346 struct external_sort 
347   {
348     const struct sort_criteria *criteria; /* Sort criteria. */
349     size_t value_cnt;                 /* Size of data in `union value's. */
350     struct casefile **initial_runs;   /* Array of initial runs. */
351     size_t run_cnt, run_cap;          /* Number of runs, allocated capacity. */
352   };
353
354 /* Prototypes for helper functions. */
355 static int write_initial_runs (struct external_sort *, struct casereader *);
356 static int merge (struct external_sort *);
357 static void destroy_external_sort (struct external_sort *);
358
359 /* Performs an external sort of the active file according to the
360    specification in SCP.  Forms initial runs using a heap as a
361    reservoir.  Determines the optimum merge pattern via Huffman's
362    method (see Knuth vol. 3, 2nd edition, p. 365-366), and merges
363    according to that pattern. */
364 static struct casefile *
365 do_external_sort (struct casereader *reader,
366                   const struct sort_criteria *criteria)
367 {
368   struct external_sort *xsrt;
369
370   casefile_to_disk (casereader_get_casefile (reader));
371
372   xsrt = xmalloc (sizeof *xsrt);
373   xsrt->criteria = criteria;
374   xsrt->value_cnt = casefile_get_value_cnt (casereader_get_casefile (reader));
375   xsrt->run_cap = 512;
376   xsrt->run_cnt = 0;
377   xsrt->initial_runs = xmalloc (sizeof *xsrt->initial_runs * xsrt->run_cap);
378   if (write_initial_runs (xsrt, reader) && merge (xsrt))
379     {
380       struct casefile *output = xsrt->initial_runs[0];
381       xsrt->initial_runs[0] = NULL;
382       destroy_external_sort (xsrt);
383       return output;
384     }
385   else
386     {
387       destroy_external_sort (xsrt);
388       return NULL;
389     }
390 }
391
392 /* Destroys XSRT. */
393 static void
394 destroy_external_sort (struct external_sort *xsrt) 
395 {
396   if (xsrt != NULL) 
397     {
398       int i;
399       
400       for (i = 0; i < xsrt->run_cnt; i++)
401         casefile_destroy (xsrt->initial_runs[i]);
402       free (xsrt->initial_runs);
403       free (xsrt);
404     }
405 }
406 \f
407 /* Replacement selection. */
408
409 /* Pairs a record with a run number. */
410 struct record_run
411   {
412     int run;                    /* Run number of case. */
413     struct ccase record;        /* Case data. */
414   };
415
416 /* Represents a set of initial runs during an external sort. */
417 struct initial_run_state 
418   {
419     struct external_sort *xsrt;
420
421     /* Reservoir. */
422     struct record_run *records; /* Records arranged as a heap. */
423     size_t record_cnt;          /* Current number of records. */
424     size_t record_cap;          /* Capacity for records. */
425     
426     /* Run currently being output. */
427     int run;                    /* Run number. */
428     size_t case_cnt;            /* Number of cases so far. */
429     struct casefile *casefile;  /* Output file. */
430     struct ccase last_output;   /* Record last output. */
431
432     int okay;                   /* Zero if an error has been encountered. */
433   };
434
435 static const struct case_sink_class sort_sink_class;
436
437 static void destroy_initial_run_state (struct initial_run_state *);
438 static void process_case (struct initial_run_state *, const struct ccase *);
439 static int allocate_cases (struct initial_run_state *);
440 static void output_record (struct initial_run_state *);
441 static void start_run (struct initial_run_state *);
442 static void end_run (struct initial_run_state *);
443 static int compare_record_run (const struct record_run *,
444                                const struct record_run *,
445                                struct initial_run_state *);
446 static int compare_record_run_minheap (const void *, const void *, void *);
447
448 /* Reads cases from READER and composes initial runs in XSRT. */
449 static int
450 write_initial_runs (struct external_sort *xsrt, struct casereader *reader)
451 {
452   struct initial_run_state *irs;
453   struct ccase c;
454   int success = 0;
455
456   /* Allocate memory for cases. */
457   irs = xmalloc (sizeof *irs);
458   irs->xsrt = xsrt;
459   irs->records = NULL;
460   irs->record_cnt = irs->record_cap = 0;
461   irs->run = 0;
462   irs->case_cnt = 0;
463   irs->casefile = NULL;
464   case_nullify (&irs->last_output);
465   irs->okay = 1;
466   if (!allocate_cases (irs)) 
467     goto done;
468
469   /* Create initial runs. */
470   start_run (irs);
471   for (; irs->okay && casereader_read (reader, &c); case_destroy (&c))
472     process_case (irs, &c);
473   while (irs->okay && irs->record_cnt > 0)
474     output_record (irs);
475   end_run (irs);
476
477   success = irs->okay;
478
479  done:
480   destroy_initial_run_state (irs);
481
482   return success;
483 }
484
485 /* Add a single case to an initial run. */
486 static void
487 process_case (struct initial_run_state *irs, const struct ccase *c)
488 {
489   struct record_run *new_record_run;
490
491   /* Compose record_run for this run and add to heap. */
492   assert (irs->record_cnt < irs->record_cap - 1);
493   new_record_run = irs->records + irs->record_cnt++;
494   case_copy (&new_record_run->record, 0, c, 0, irs->xsrt->value_cnt);
495   new_record_run->run = irs->run;
496   if (!case_is_null (&irs->last_output)
497       && compare_record (c, &irs->last_output, irs->xsrt->criteria) < 0)
498     new_record_run->run = irs->run + 1;
499   push_heap (irs->records, irs->record_cnt, sizeof *irs->records,
500              compare_record_run_minheap, irs);
501
502   /* Output a record if the reservoir is full. */
503   if (irs->record_cnt == irs->record_cap - 1 && irs->okay)
504     output_record (irs);
505 }
506
507 /* Destroys the initial run state represented by IRS. */
508 static void
509 destroy_initial_run_state (struct initial_run_state *irs) 
510 {
511   int i;
512
513   if (irs == NULL)
514     return;
515
516   for (i = 0; i < irs->record_cap; i++)
517     case_destroy (&irs->records[i].record);
518   free (irs->records);
519
520   if (irs->casefile != NULL)
521     casefile_sleep (irs->casefile);
522
523   free (irs);
524 }
525
526 /* Allocates room for lots of cases as a buffer. */
527 static int
528 allocate_cases (struct initial_run_state *irs)
529 {
530   int approx_case_cost; /* Approximate memory cost of one case in bytes. */
531   int max_cases;        /* Maximum number of cases to allocate. */
532   int i;
533
534   /* Allocate as many cases as we can within the workspace
535      limit. */
536   approx_case_cost = (sizeof *irs->records
537                       + irs->xsrt->value_cnt * sizeof (union value)
538                       + 4 * sizeof (void *));
539   max_cases = get_max_workspace() / approx_case_cost;
540   irs->records = malloc (sizeof *irs->records * max_cases);
541   for (i = 0; i < max_cases; i++)
542     if (!case_try_create (&irs->records[i].record, irs->xsrt->value_cnt))
543       {
544         max_cases = i;
545         break;
546       }
547   irs->record_cap = max_cases;
548
549   /* Fail if we didn't allocate an acceptable number of cases. */
550   if (irs->records == NULL || max_cases < MIN_BUFFER_TOTAL_SIZE_RECS)
551     {
552       msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "
553                  "cases of %d bytes each.  (PSPP workspace is currently "
554                  "restricted to a maximum of %d KB.)"),
555            MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, get_max_workspace() / 1024);
556       return 0;
557     }
558   return 1;
559 }
560
561 /* Compares the VAR_CNT variables in VARS[] between the `value's at
562    A and B, and returns a strcmp()-type result. */
563 static int
564 compare_record (const struct ccase *a, const struct ccase *b,
565                 const struct sort_criteria *criteria)
566 {
567   int i;
568
569   assert (a != NULL);
570   assert (b != NULL);
571   
572   for (i = 0; i < criteria->crit_cnt; i++)
573     {
574       const struct sort_criterion *c = &criteria->crits[i];
575       int result;
576       
577       if (c->width == 0)
578         {
579           double af = case_num (a, c->fv);
580           double bf = case_num (b, c->fv);
581           
582           result = af < bf ? -1 : af > bf;
583         }
584       else
585         result = memcmp (case_str (a, c->fv), case_str (b, c->fv), c->width);
586
587       if (result != 0)
588         return c->dir == SRT_ASCEND ? result : -result;
589     }
590
591   return 0;
592 }
593
594 /* Compares record-run tuples A and B on run number first, then
595    on the current record according to SCP. */
596 static int
597 compare_record_run (const struct record_run *a,
598                     const struct record_run *b,
599                     struct initial_run_state *irs)
600 {
601   if (a->run != b->run)
602     return a->run > b->run ? 1 : -1;
603   else
604     return compare_record (&a->record, &b->record, irs->xsrt->criteria);
605 }
606
607 /* Compares record-run tuples A and B on run number first, then
608    on the current record according to SCP, but in descending
609    order. */
610 static int
611 compare_record_run_minheap (const void *a, const void *b, void *irs) 
612 {
613   return -compare_record_run (a, b, irs);
614 }
615
616 /* Begins a new initial run, specifically its output file. */
617 static void
618 start_run (struct initial_run_state *irs)
619 {
620   irs->run++;
621   irs->case_cnt = 0;
622   irs->casefile = casefile_create (irs->xsrt->value_cnt);
623   casefile_to_disk (irs->casefile);
624   case_nullify (&irs->last_output); 
625 }
626
627 /* Ends the current initial run.  */
628 static void
629 end_run (struct initial_run_state *irs)
630 {
631   struct external_sort *xsrt = irs->xsrt;
632
633   /* Record initial run. */
634   if (irs->casefile != NULL) 
635     {
636       if (xsrt->run_cnt >= xsrt->run_cap) 
637         {
638           xsrt->run_cap *= 2;
639           xsrt->initial_runs
640             = xrealloc (xsrt->initial_runs,
641                         sizeof *xsrt->initial_runs * xsrt->run_cap);
642         }
643       xsrt->initial_runs[xsrt->run_cnt++] = irs->casefile;
644       irs->casefile = NULL; 
645     }
646 }
647
648 /* Writes a record to the current initial run. */
649 static void
650 output_record (struct initial_run_state *irs)
651 {
652   struct record_run *record_run;
653   struct ccase case_tmp;
654   
655   /* Extract minimum case from heap. */
656   assert (irs->record_cnt > 0);
657   pop_heap (irs->records, irs->record_cnt--, sizeof *irs->records,
658             compare_record_run_minheap, irs);
659   record_run = irs->records + irs->record_cnt;
660
661   /* Bail if an error has occurred. */
662   if (!irs->okay)
663     return;
664
665   /* Start new run if necessary. */
666   assert (record_run->run == irs->run
667           || record_run->run == irs->run + 1);
668   if (record_run->run != irs->run)
669     {
670       end_run (irs);
671       start_run (irs);
672     }
673   assert (record_run->run == irs->run);
674   irs->case_cnt++;
675
676   /* Write to disk. */
677   if (irs->casefile != NULL)
678     casefile_append (irs->casefile, &record_run->record);
679
680   /* This record becomes last_output. */
681   irs->last_output = case_tmp = record_run->record;
682   record_run->record = irs->records[irs->record_cap - 1].record;
683   irs->records[irs->record_cap - 1].record = case_tmp;
684 }
685 \f
686 /* Merging. */
687
688 /* State of merging initial runs. */
689 struct merge_state 
690   {
691     struct external_sort *xsrt; /* External sort state. */
692     struct ccase *cases;        /* Buffers. */
693     size_t case_cnt;            /* Number of buffers. */
694   };
695
696 struct run;
697 static struct casefile *merge_once (struct merge_state *,
698                                     struct casefile *[], size_t);
699 static int mod (int, int);
700
701 /* Performs a series of P-way merges of initial runs. */
702 static int
703 merge (struct external_sort *xsrt)
704 {
705   struct merge_state mrg;       /* State of merge. */
706   size_t approx_case_cost;      /* Approximate memory cost of one case. */
707   int max_order;                /* Maximum order of merge. */
708   size_t dummy_run_cnt;         /* Number of dummy runs to insert. */
709   int success = 0;
710   int i;
711
712   mrg.xsrt = xsrt;
713
714   /* Allocate as many cases as possible into cases. */
715   approx_case_cost = (sizeof *mrg.cases
716                       + xsrt->value_cnt * sizeof (union value)
717                       + 4 * sizeof (void *));
718   mrg.case_cnt = get_max_workspace() / approx_case_cost;
719   mrg.cases = malloc (sizeof *mrg.cases * mrg.case_cnt);
720   if (mrg.cases == NULL)
721     goto done;
722   for (i = 0; i < mrg.case_cnt; i++) 
723     if (!case_try_create (&mrg.cases[i], xsrt->value_cnt)) 
724       {
725         mrg.case_cnt = i;
726         break;
727       }
728   if (mrg.case_cnt < MIN_BUFFER_TOTAL_SIZE_RECS)
729     {
730       msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "
731                  "cases of %d bytes each.  (PSPP workspace is currently "
732                  "restricted to a maximum of %d KB.)"),
733            MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, get_max_workspace() / 1024);
734       return 0;
735     }
736
737   /* Determine maximum order of merge. */
738   max_order = MAX_MERGE_ORDER;
739   if (mrg.case_cnt / max_order < MIN_BUFFER_SIZE_RECS)
740     max_order = mrg.case_cnt / MIN_BUFFER_SIZE_RECS;
741   else if (mrg.case_cnt / max_order * xsrt->value_cnt * sizeof (union value)
742            < MIN_BUFFER_SIZE_BYTES)
743     max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES
744                                 / (xsrt->value_cnt * sizeof (union value)));
745   if (max_order < 2)
746     max_order = 2;
747   if (max_order > xsrt->run_cnt)
748     max_order = xsrt->run_cnt;
749
750   /* Repeatedly merge the P shortest existing runs until only one run
751      is left. */
752   make_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
753              compare_initial_runs, NULL);
754   dummy_run_cnt = mod (1 - (int) xsrt->run_cnt, max_order - 1);
755   assert (max_order == 2
756           || (xsrt->run_cnt + dummy_run_cnt) % (max_order - 1) == 1);
757   while (xsrt->run_cnt > 1)
758     {
759       struct casefile *output_run;
760       int order;
761       int i;
762
763       /* Choose order of merge (max_order after first merge). */
764       order = max_order - dummy_run_cnt;
765       dummy_run_cnt = 0;
766
767       /* Choose runs to merge. */
768       assert (xsrt->run_cnt >= order);
769       for (i = 0; i < order; i++) 
770         pop_heap (xsrt->initial_runs, xsrt->run_cnt--,
771                   sizeof *xsrt->initial_runs,
772                   compare_initial_runs, NULL); 
773           
774       /* Merge runs. */
775       output_run = merge_once (&mrg,
776                                xsrt->initial_runs + xsrt->run_cnt, order);
777       if (output_run == NULL)
778         goto done;
779       
780       /* Add output run to heap. */
781       xsrt->initial_runs[xsrt->run_cnt++] = output_run;
782       push_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
783                  compare_initial_runs, NULL);
784     }
785
786   /* Exactly one run is left, which contains the entire sorted
787      file.  We could use it to find a total case count. */
788   assert (xsrt->run_cnt == 1);
789
790   success = 1;
791
792  done:
793   for (i = 0; i < mrg.case_cnt; i++)
794     case_destroy (&mrg.cases[i]);
795   free (mrg.cases);
796
797   return success;
798 }
799
800 /* Modulo function as defined by Knuth. */
801 static int
802 mod (int x, int y)
803 {
804   if (y == 0)
805     return x;
806   else if (x == 0)
807     return 0;
808   else if (x > 0 && y > 0)
809     return x % y;
810   else if (x < 0 && y > 0)
811     return y - (-x) % y;
812
813   abort ();
814 }
815
816 /* Merges the RUN_CNT initial runs specified in INPUT_RUNS into a
817    new run.  Returns nonzero only if successful.  Adds an entry
818    to MRG->xsrt->runs for the output file if and only if the
819    output file is actually created.  Always deletes all the input
820    files. */
821 static struct casefile *
822 merge_once (struct merge_state *mrg,
823             struct casefile *input_runs[],
824             size_t run_cnt)
825 {
826   struct casereader *input_readers[MAX_MERGE_ORDER];
827   struct ccase input_cases[MAX_MERGE_ORDER];
828   struct casefile *output_casefile = NULL;
829   int i;
830
831   for (i = 0; i < run_cnt; i++) 
832     {
833       input_readers[i] = casefile_get_reader (input_runs[i]);
834       if (!casereader_read_xfer (input_readers[i], &input_cases[i]))
835         {
836           run_cnt--;
837           i--;
838         }
839     }
840   
841   output_casefile = casefile_create (mrg->xsrt->value_cnt);
842   casefile_to_disk (output_casefile);
843
844   /* Merge. */
845   while (run_cnt > 0) 
846     {
847       size_t min_idx;
848
849       /* Find minimum. */
850       min_idx = 0;
851       for (i = 1; i < run_cnt; i++)
852         if (compare_record (&input_cases[i], &input_cases[min_idx],
853                             mrg->xsrt->criteria) < 0)
854           min_idx = i;
855
856       /* Write minimum to output file. */
857       casefile_append_xfer (output_casefile, &input_cases[i]);
858
859       if (!casereader_read_xfer (input_readers[i], &input_cases[i]))
860         {
861           casereader_destroy (input_readers[i]);
862           casefile_destroy (input_runs[i]);
863
864           run_cnt--;
865           input_runs[i] = input_runs[run_cnt--];
866           input_readers[i] = input_readers[run_cnt--];
867           input_cases[i] = input_cases[run_cnt--];
868         } 
869     }
870
871   casefile_sleep (output_casefile);
872
873   return output_casefile;
874 }