/[pspp]/pspp/src/sort.c
ViewVC logotype

Diff of /pspp/src/sort.c

Parent Directory Parent Directory | Revision Log Revision Log | View Patch Patch

revision 1.26 by blp, Mon Mar 14 06:54:40 2005 UTC revision 1.27 by blp, Tue Mar 15 06:04:10 2005 UTC
# Line 20  Line 20 
20  #include <config.h>  #include <config.h>
21  #include "sort.h"  #include "sort.h"
22  #include "error.h"  #include "error.h"
23    #include <limits.h>
24  #include <stdio.h>  #include <stdio.h>
25  #include <stdlib.h>  #include <stdlib.h>
26  #include <errno.h>  #include <errno.h>
# Line 31  Line 32 
32  #include "command.h"  #include "command.h"
33  #include "error.h"  #include "error.h"
34  #include "expressions/public.h"  #include "expressions/public.h"
35    #include "glob.h"
36  #include "lexer.h"  #include "lexer.h"
37  #include "misc.h"  #include "misc.h"
38  #include "settings.h"  #include "settings.h"
# Line 39  Line 41 
41  #include "vfm.h"  #include "vfm.h"
42  #include "vfmP.h"  #include "vfmP.h"
43    
 #if HAVE_UNISTD_H  
 #include <unistd.h>  
 #endif  
   
 #if HAVE_SYS_TYPES_H  
 #include <sys/types.h>  
 #endif  
   
 #if HAVE_SYS_STAT_H  
 #include <sys/stat.h>  
 #endif  
   
44  #include "debug-print.h"  #include "debug-print.h"
45    
46  /* Sort direction. */  /* Sort direction. */
# Line 75  struct sort_criteria Line 65  struct sort_criteria
65      size_t crit_cnt;      size_t crit_cnt;
66    };    };
67    
68    /* These should only be changed for testing purposes. */
69    static int min_buffers = 64;
70    static int max_buffers = INT_MAX;
71    static bool allow_internal_sort = true;
72    
73  static int compare_record (const struct ccase *, const struct ccase *,  static int compare_record (const struct ccase *, const struct ccase *,
74                             const struct sort_criteria *);                             const struct sort_criteria *);
75  static struct casefile *do_internal_sort (struct casereader *,  static struct casefile *do_internal_sort (struct casereader *,
# Line 87  int Line 82  int
82  cmd_sort_cases (void)  cmd_sort_cases (void)
83  {  {
84    struct sort_criteria *criteria;    struct sort_criteria *criteria;
85    int success;    bool success = false;
86    
87    lex_match (T_BY);    lex_match (T_BY);
88    
# Line 95  cmd_sort_cases (void) Line 90  cmd_sort_cases (void)
90    if (criteria == NULL)    if (criteria == NULL)
91      return CMD_FAILURE;      return CMD_FAILURE;
92    
93      if (test_mode && lex_match ('/'))
94        {
95          if (!lex_force_match_id ("BUFFERS") || !lex_match ('=')
96              || !lex_force_int ())
97            goto done;
98    
99          min_buffers = max_buffers = lex_integer ();
100          allow_internal_sort = false;
101          if (max_buffers < 2)
102            {
103              msg (SE, _("Buffer limit must be at least 2."));
104              goto done;
105            }
106    
107          lex_get ();
108        }
109    
110    success = sort_active_file_in_place (criteria);    success = sort_active_file_in_place (criteria);
111    
112     done:
113      min_buffers = 64;
114      max_buffers = INT_MAX;
115      allow_internal_sort = true;
116      
117    sort_destroy_criteria (criteria);    sort_destroy_criteria (criteria);
118    return success ? lex_end_of_command () : CMD_FAILURE;    return success ? lex_end_of_command () : CMD_FAILURE;
119  }  }
# Line 252  sort_destroy_criteria (struct sort_crite Line 270  sort_destroy_criteria (struct sort_crite
270  struct casefile *  struct casefile *
271  sort_execute (struct casereader *reader, const struct sort_criteria *criteria)  sort_execute (struct casereader *reader, const struct sort_criteria *criteria)
272  {  {
273    struct casefile *output;    struct casefile *output = do_internal_sort (reader, criteria);
   
   output = do_internal_sort (reader, criteria);  
274    if (output == NULL)    if (output == NULL)
275      output = do_external_sort (reader, criteria);      output = do_external_sort (reader, criteria);
276    casereader_destroy (reader);    casereader_destroy (reader);
# Line 271  struct indexed_case Line 287  struct indexed_case
287  static int compare_indexed_cases (const void *, const void *, void *);  static int compare_indexed_cases (const void *, const void *, void *);
288    
289  /* If the data is in memory, do an internal sort and return a new  /* If the data is in memory, do an internal sort and return a new
290     casefile for the data. */     casefile for the data.  Otherwise, return a null pointer. */
291  static struct casefile *  static struct casefile *
292  do_internal_sort (struct casereader *reader,  do_internal_sort (struct casereader *reader,
293                    const struct sort_criteria *criteria)                    const struct sort_criteria *criteria)
# Line 280  do_internal_sort (struct casereader *rea Line 296  do_internal_sort (struct casereader *rea
296    struct casefile *dst;    struct casefile *dst;
297    unsigned long case_cnt;    unsigned long case_cnt;
298    
299      if (!allow_internal_sort)
300        return NULL;
301    
302    src = casereader_get_casefile (reader);    src = casereader_get_casefile (reader);
303    if (casefile_get_case_cnt (src) > 1 && !casefile_in_core (src))    if (casefile_get_case_cnt (src) > 1 && !casefile_in_core (src))
304      return NULL;      return NULL;
# Line 335  compare_indexed_cases (const void *a_, c Line 354  compare_indexed_cases (const void *a_, c
354    
355  /* External sort. */  /* External sort. */
356    
357  /* Maximum order of merge.  If you increase this, then you should  /* Maximum order of merge (external sort only).  The maximum
358     use a heap for comparing cases during merge.  */     reasonable value is about 7.  Above that, it would be a good
359  #define MAX_MERGE_ORDER         7     idea to use a heap in merge_once() to select the minimum. */
360    #define MAX_MERGE_ORDER 7
 /* Minimum total number of records for buffers. */  
 #define MIN_BUFFER_TOTAL_SIZE_RECS      64  
   
 /* Minimum single input buffer size, in bytes and records. */  
 #define MIN_BUFFER_SIZE_BYTES   4096  
 #define MIN_BUFFER_SIZE_RECS    16  
   
 #if MIN_BUFFER_SIZE_RECS * 2 + 16 > MIN_BUFFER_TOTAL_SIZE_RECS  
 #error MIN_BUFFER_SIZE_RECS and MIN_BUFFER_TOTAL_SIZE_RECS do not make sense.  
 #endif  
   
 /* Sorts initial runs A and B in decending order by length. */  
 static int  
 compare_initial_runs (const void *a_, const void *b_, void *aux UNUSED)  
 {  
   struct casefile *const *a = a_;  
   struct casefile *const *b = b_;  
   unsigned long a_case_cnt = casefile_get_case_cnt (*a);  
   unsigned long b_case_cnt = casefile_get_case_cnt (*b);  
     
   return a_case_cnt > b_case_cnt ? -1 : a_case_cnt < b_case_cnt;  
 }  
361    
362  /* Results of an external sort. */  /* Results of an external sort. */
363  struct external_sort  struct external_sort
364    {    {
365      const struct sort_criteria *criteria; /* Sort criteria. */      const struct sort_criteria *criteria; /* Sort criteria. */
366      size_t value_cnt;                 /* Size of data in `union value's. */      size_t value_cnt;                 /* Size of data in `union value's. */
367      struct casefile **initial_runs;   /* Array of initial runs. */      struct casefile **runs;           /* Array of initial runs. */
368      size_t run_cnt, run_cap;          /* Number of runs, allocated capacity. */      size_t run_cnt, run_cap;          /* Number of runs, allocated capacity. */
369    };    };
370    
371  /* Prototypes for helper functions. */  /* Prototypes for helper functions. */
372  static int write_initial_runs (struct external_sort *, struct casereader *);  static int write_runs (struct external_sort *, struct casereader *);
373  static int merge (struct external_sort *);  static struct casefile *merge (struct external_sort *);
374  static void destroy_external_sort (struct external_sort *);  static void destroy_external_sort (struct external_sort *);
375    
376  /* Performs an external sort of the active file according to the  /* Performs a stable external sort of the active file according
377     specification in SCP.  Forms initial runs using a heap as a     to the specification in SCP.  Forms initial runs using a heap
378     reservoir.  Determines the optimum merge pattern via Huffman's     as a reservoir.  Merges the initial runs according to a
379     method (see Knuth vol. 3, 2nd edition, p. 365-366), and merges     pattern that assures stability. */
    according to that pattern. */  
380  static struct casefile *  static struct casefile *
381  do_external_sort (struct casereader *reader,  do_external_sort (struct casereader *reader,
382                    const struct sort_criteria *criteria)                    const struct sort_criteria *criteria)
# Line 394  do_external_sort (struct casereader *rea Line 390  do_external_sort (struct casereader *rea
390    xsrt->value_cnt = casefile_get_value_cnt (casereader_get_casefile (reader));    xsrt->value_cnt = casefile_get_value_cnt (casereader_get_casefile (reader));
391    xsrt->run_cap = 512;    xsrt->run_cap = 512;
392    xsrt->run_cnt = 0;    xsrt->run_cnt = 0;
393    xsrt->initial_runs = xmalloc (sizeof *xsrt->initial_runs * xsrt->run_cap);    xsrt->runs = xmalloc (sizeof *xsrt->runs * xsrt->run_cap);
394    if (write_initial_runs (xsrt, reader) && merge (xsrt))    if (write_runs (xsrt, reader))
395      {      {
396        struct casefile *output = xsrt->initial_runs[0];        struct casefile *output = merge (xsrt);
       xsrt->initial_runs[0] = NULL;  
397        destroy_external_sort (xsrt);        destroy_external_sort (xsrt);
398        return output;        return output;
399      }      }
# Line 418  destroy_external_sort (struct external_s Line 413  destroy_external_sort (struct external_s
413        int i;        int i;
414                
415        for (i = 0; i < xsrt->run_cnt; i++)        for (i = 0; i < xsrt->run_cnt; i++)
416          casefile_destroy (xsrt->initial_runs[i]);          casefile_destroy (xsrt->runs[i]);
417        free (xsrt->initial_runs);        free (xsrt->runs);
418        free (xsrt);        free (xsrt);
419      }      }
420  }  }
# Line 431  struct record_run Line 426  struct record_run
426    {    {
427      int run;                    /* Run number of case. */      int run;                    /* Run number of case. */
428      struct ccase record;        /* Case data. */      struct ccase record;        /* Case data. */
429        size_t idx;                 /* Case number (for stability). */
430    };    };
431    
432  /* Represents a set of initial runs during an external sort. */  /* Represents a set of initial runs during an external sort. */
# Line 455  struct initial_run_state Line 451  struct initial_run_state
451  static const struct case_sink_class sort_sink_class;  static const struct case_sink_class sort_sink_class;
452    
453  static void destroy_initial_run_state (struct initial_run_state *);  static void destroy_initial_run_state (struct initial_run_state *);
454  static void process_case (struct initial_run_state *, const struct ccase *);  static void process_case (struct initial_run_state *, const struct ccase *,
455                              size_t);
456  static int allocate_cases (struct initial_run_state *);  static int allocate_cases (struct initial_run_state *);
457  static void output_record (struct initial_run_state *);  static void output_record (struct initial_run_state *);
458  static void start_run (struct initial_run_state *);  static void start_run (struct initial_run_state *);
# Line 467  static int compare_record_run_minheap (c Line 464  static int compare_record_run_minheap (c
464    
465  /* Reads cases from READER and composes initial runs in XSRT. */  /* Reads cases from READER and composes initial runs in XSRT. */
466  static int  static int
467  write_initial_runs (struct external_sort *xsrt, struct casereader *reader)  write_runs (struct external_sort *xsrt, struct casereader *reader)
468  {  {
469    struct initial_run_state *irs;    struct initial_run_state *irs;
470    struct ccase c;    struct ccase c;
471      size_t idx = 0;
472    int success = 0;    int success = 0;
473    
474    /* Allocate memory for cases. */    /* Allocate memory for cases. */
# Line 489  write_initial_runs (struct external_sort Line 487  write_initial_runs (struct external_sort
487    /* Create initial runs. */    /* Create initial runs. */
488    start_run (irs);    start_run (irs);
489    for (; irs->okay && casereader_read (reader, &c); case_destroy (&c))    for (; irs->okay && casereader_read (reader, &c); case_destroy (&c))
490      process_case (irs, &c);      process_case (irs, &c, idx++);
491    while (irs->okay && irs->record_cnt > 0)    while (irs->okay && irs->record_cnt > 0)
492      output_record (irs);      output_record (irs);
493    end_run (irs);    end_run (irs);
# Line 504  write_initial_runs (struct external_sort Line 502  write_initial_runs (struct external_sort
502    
503  /* Add a single case to an initial run. */  /* Add a single case to an initial run. */
504  static void  static void
505  process_case (struct initial_run_state *irs, const struct ccase *c)  process_case (struct initial_run_state *irs, const struct ccase *c, size_t idx)
506  {  {
507    struct record_run *new_record_run;    struct record_run *rr;
508    
509    /* Compose record_run for this run and add to heap. */    /* Compose record_run for this run and add to heap. */
510    assert (irs->record_cnt < irs->record_cap - 1);    assert (irs->record_cnt < irs->record_cap - 1);
511    new_record_run = irs->records + irs->record_cnt++;    rr = irs->records + irs->record_cnt++;
512    case_copy (&new_record_run->record, 0, c, 0, irs->xsrt->value_cnt);    case_copy (&rr->record, 0, c, 0, irs->xsrt->value_cnt);
513    new_record_run->run = irs->run;    rr->run = irs->run;
514      rr->idx = idx;
515    if (!case_is_null (&irs->last_output)    if (!case_is_null (&irs->last_output)
516        && compare_record (c, &irs->last_output, irs->xsrt->criteria) < 0)        && compare_record (c, &irs->last_output, irs->xsrt->criteria) < 0)
517      new_record_run->run = irs->run + 1;      rr->run = irs->run + 1;
518    push_heap (irs->records, irs->record_cnt, sizeof *irs->records,    push_heap (irs->records, irs->record_cnt, sizeof *irs->records,
519               compare_record_run_minheap, irs);               compare_record_run_minheap, irs);
520    
# Line 557  allocate_cases (struct initial_run_state Line 556  allocate_cases (struct initial_run_state
556                        + irs->xsrt->value_cnt * sizeof (union value)                        + irs->xsrt->value_cnt * sizeof (union value)
557                        + 4 * sizeof (void *));                        + 4 * sizeof (void *));
558    max_cases = get_max_workspace() / approx_case_cost;    max_cases = get_max_workspace() / approx_case_cost;
559      if (max_cases > max_buffers)
560        max_cases = max_buffers;
561    irs->records = malloc (sizeof *irs->records * max_cases);    irs->records = malloc (sizeof *irs->records * max_cases);
562    for (i = 0; i < max_cases; i++)    for (i = 0; i < max_cases; i++)
563      if (!case_try_create (&irs->records[i].record, irs->xsrt->value_cnt))      if (!case_try_create (&irs->records[i].record, irs->xsrt->value_cnt))
# Line 567  allocate_cases (struct initial_run_state Line 568  allocate_cases (struct initial_run_state
568    irs->record_cap = max_cases;    irs->record_cap = max_cases;
569    
570    /* Fail if we didn't allocate an acceptable number of cases. */    /* Fail if we didn't allocate an acceptable number of cases. */
571    if (irs->records == NULL || max_cases < MIN_BUFFER_TOTAL_SIZE_RECS)    if (irs->records == NULL || max_cases < min_buffers)
572      {      {
573        msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "        msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "
574                   "cases of %d bytes each.  (PSPP workspace is currently "                   "cases of %d bytes each.  (PSPP workspace is currently "
575                   "restricted to a maximum of %d KB.)"),                   "restricted to a maximum of %d KB.)"),
576             MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, get_max_workspace() / 1024);             min_buffers, approx_case_cost, get_max_workspace() / 1024);
577        return 0;        return 0;
578      }      }
579    return 1;    return 1;
# Line 612  compare_record (const struct ccase *a, c Line 613  compare_record (const struct ccase *a, c
613  }  }
614    
615  /* Compares record-run tuples A and B on run number first, then  /* Compares record-run tuples A and B on run number first, then
616     on the current record according to SCP. */     on record, then on case index. */
617  static int  static int
618  compare_record_run (const struct record_run *a,  compare_record_run (const struct record_run *a,
619                      const struct record_run *b,                      const struct record_run *b,
620                      struct initial_run_state *irs)                      struct initial_run_state *irs)
621  {  {
622    if (a->run != b->run)    int result = a->run < b->run ? -1 : a->run > b->run;
623      return a->run > b->run ? 1 : -1;    if (result == 0)
624    else      result = compare_record (&a->record, &b->record, irs->xsrt->criteria);
625      return compare_record (&a->record, &b->record, irs->xsrt->criteria);    if (result == 0)
626        result = a->idx < b->idx ? -1 : a->idx > b->idx;
627      return result;
628  }  }
629    
630  /* Compares record-run tuples A and B on run number first, then  /* Compares record-run tuples A and B on run number first, then
# Line 653  end_run (struct initial_run_state *irs) Line 656  end_run (struct initial_run_state *irs)
656    /* Record initial run. */    /* Record initial run. */
657    if (irs->casefile != NULL)    if (irs->casefile != NULL)
658      {      {
659          casefile_sleep (irs->casefile);
660        if (xsrt->run_cnt >= xsrt->run_cap)        if (xsrt->run_cnt >= xsrt->run_cap)
661          {          {
662            xsrt->run_cap *= 2;            xsrt->run_cap *= 2;
663            xsrt->initial_runs            xsrt->runs = xrealloc (xsrt->runs,
664              = xrealloc (xsrt->initial_runs,                                   sizeof *xsrt->runs * xsrt->run_cap);
                         sizeof *xsrt->initial_runs * xsrt->run_cap);  
665          }          }
666        xsrt->initial_runs[xsrt->run_cnt++] = irs->casefile;        xsrt->runs[xsrt->run_cnt++] = irs->casefile;
667        irs->casefile = NULL;        irs->casefile = NULL;
668      }      }
669  }  }
# Line 705  output_record (struct initial_run_state Line 708  output_record (struct initial_run_state
708    
709  /* Merging. */  /* Merging. */
710    
711  /* State of merging initial runs. */  static int choose_merge (struct casefile *runs[], int run_cnt, int order);
712  struct merge_state  static struct casefile *merge_once (struct external_sort *,
   {  
     struct external_sort *xsrt; /* External sort state. */  
     struct ccase *cases;        /* Buffers. */  
     size_t case_cnt;            /* Number of buffers. */  
   };  
   
 struct run;  
 static struct casefile *merge_once (struct merge_state *,  
713                                      struct casefile *[], size_t);                                      struct casefile *[], size_t);
 static int mod (int, int);  
714    
715  /* Performs a series of P-way merges of initial runs. */  /* Repeatedly merges run until only one is left,
716  static int     and returns the final casefile.  */
717    static struct casefile *
718  merge (struct external_sort *xsrt)  merge (struct external_sort *xsrt)
719  {  {
   struct merge_state mrg;       /* State of merge. */  
   size_t approx_case_cost;      /* Approximate memory cost of one case. */  
   int max_order;                /* Maximum order of merge. */  
   size_t dummy_run_cnt;         /* Number of dummy runs to insert. */  
   int success = 0;  
   int i;  
   
   mrg.xsrt = xsrt;  
   
   /* Allocate as many cases as possible into cases. */  
   approx_case_cost = (sizeof *mrg.cases  
                       + xsrt->value_cnt * sizeof (union value)  
                       + 4 * sizeof (void *));  
   mrg.case_cnt = get_max_workspace() / approx_case_cost;  
   mrg.cases = malloc (sizeof *mrg.cases * mrg.case_cnt);  
   if (mrg.cases == NULL)  
     goto done;  
   for (i = 0; i < mrg.case_cnt; i++)  
     if (!case_try_create (&mrg.cases[i], xsrt->value_cnt))  
       {  
         mrg.case_cnt = i;  
         break;  
       }  
   if (mrg.case_cnt < MIN_BUFFER_TOTAL_SIZE_RECS)  
     {  
       msg (SE, _("Out of memory.  Could not allocate room for minimum of %d "  
                  "cases of %d bytes each.  (PSPP workspace is currently "  
                  "restricted to a maximum of %d KB.)"),  
            MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, get_max_workspace() / 1024);  
       return 0;  
     }  
   
   /* Determine maximum order of merge. */  
   max_order = MAX_MERGE_ORDER;  
   if (mrg.case_cnt / max_order < MIN_BUFFER_SIZE_RECS)  
     max_order = mrg.case_cnt / MIN_BUFFER_SIZE_RECS;  
   else if (mrg.case_cnt / max_order * xsrt->value_cnt * sizeof (union value)  
            < MIN_BUFFER_SIZE_BYTES)  
     max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES  
                                 / (xsrt->value_cnt * sizeof (union value)));  
   if (max_order < 2)  
     max_order = 2;  
   if (max_order > xsrt->run_cnt)  
     max_order = xsrt->run_cnt;  
   
   /* Repeatedly merge the P shortest existing runs until only one run  
      is left. */  
   make_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,  
              compare_initial_runs, NULL);  
   dummy_run_cnt = mod (1 - (int) xsrt->run_cnt, max_order - 1);  
   
   assert (max_order > 0);  
   assert (max_order <= 2  
           || (xsrt->run_cnt + dummy_run_cnt) % (max_order - 1) == 1);  
720    while (xsrt->run_cnt > 1)    while (xsrt->run_cnt > 1)
721      {      {
722        struct casefile *output_run;        int order = min (MAX_MERGE_ORDER, xsrt->run_cnt);
723        int order;        int idx = choose_merge (xsrt->runs, xsrt->run_cnt, order);
724        int i;        xsrt->runs[idx] = merge_once (xsrt, xsrt->runs + idx, order);
725          remove_range (xsrt->runs, xsrt->run_cnt, sizeof *xsrt->runs,
726        /* Choose order of merge (max_order after first merge). */                      idx + 1, order - 1);
727        order = max_order - dummy_run_cnt;        xsrt->run_cnt -= order - 1;
       dummy_run_cnt = 0;  
   
       /* Choose runs to merge. */  
       assert (xsrt->run_cnt >= order);  
       for (i = 0; i < order; i++)  
         pop_heap (xsrt->initial_runs, xsrt->run_cnt--,  
                   sizeof *xsrt->initial_runs,  
                   compare_initial_runs, NULL);  
             
       /* Merge runs. */  
       output_run = merge_once (&mrg,  
                                xsrt->initial_runs + xsrt->run_cnt, order);  
       if (output_run == NULL)  
         goto done;  
         
       /* Add output run to heap. */  
       xsrt->initial_runs[xsrt->run_cnt++] = output_run;  
       push_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,  
                  compare_initial_runs, NULL);  
728      }      }
   
   /* Exactly one run is left, which contains the entire sorted  
      file.  We could use it to find a total case count. */  
729    assert (xsrt->run_cnt == 1);    assert (xsrt->run_cnt == 1);
730      xsrt->run_cnt = 0;
731    success = 1;    return xsrt->runs[0];
   
  done:  
   for (i = 0; i < mrg.case_cnt; i++)  
     case_destroy (&mrg.cases[i]);  
   free (mrg.cases);  
   
   return success;  
732  }  }
733    
734  /* Modulo function as defined by Knuth. */  /* Chooses ORDER runs out of the RUN_CNT runs in RUNS to merge,
735       and returns the index of the first one.
736    
737       For stability, we must merge only consecutive runs.  For
738       efficiency, we choose the shortest consecutive sequence of
739       runs. */
740  static int  static int
741  mod (int x, int y)  choose_merge (struct casefile *runs[], int run_cnt, int order)
742  {  {
743    if (y == 0)    int min_idx, min_sum;
744      return x;    int cur_idx, cur_sum;
745    else if (x == 0)    int i;
746      return 0;  
747    else if (x > 0 && y > 0)    /* Sum up the length of the first ORDER runs. */
748      return x % y;    cur_sum = 0;
749    else if (x < 0 && y > 0)    for (i = 0; i < order; i++)
750      return y - (-x) % y;      cur_sum += casefile_get_case_cnt (runs[i]);
751    
752    abort ();    /* Find the shortest group of ORDER runs,
753         using a running total for efficiency. */
754      min_idx = 0;
755      min_sum = cur_sum;
756      for (cur_idx = 1; cur_idx + order <= run_cnt; cur_idx++)
757        {
758          cur_sum -= casefile_get_case_cnt (runs[cur_idx - 1]);
759          cur_sum += casefile_get_case_cnt (runs[cur_idx + order - 1]);
760          if (cur_sum < min_sum)
761            {
762              min_sum = cur_sum;
763              min_idx = cur_idx;
764            }
765        }
766    
767      return min_idx;
768  }  }
769    
770  /* Merges the RUN_CNT initial runs specified in INPUT_RUNS into a  /* Merges the RUN_CNT initial runs specified in INPUT_FILES into a
771     new run.  Returns nonzero only if successful.  Adds an entry     new run, and returns the new run. */
    to MRG->xsrt->runs for the output file if and only if the  
    output file is actually created.  Always deletes all the input  
    files. */  
772  static struct casefile *  static struct casefile *
773  merge_once (struct merge_state *mrg,  merge_once (struct external_sort *xsrt,
774              struct casefile *input_runs[],              struct casefile **const input_files,
775              size_t run_cnt)              size_t run_cnt)
776  {  {
777    struct casereader *input_readers[MAX_MERGE_ORDER];    struct run
778    struct ccase input_cases[MAX_MERGE_ORDER];      {
779    struct casefile *output_casefile = NULL;        struct casefile *file;
780          struct casereader *reader;
781          struct ccase ccase;
782        }
783      *runs;
784    
785      struct casefile *output = NULL;
786    int i;    int i;
787    
788      /* Open input files. */
789      runs = xmalloc (sizeof *runs * run_cnt);
790    for (i = 0; i < run_cnt; i++)    for (i = 0; i < run_cnt; i++)
791      {      {
792        input_readers[i] = casefile_get_destructive_reader (input_runs[i]);        struct run *r = &runs[i];
793        if (!casereader_read_xfer (input_readers[i], &input_cases[i]))        r->file = input_files[i];
794          r->reader = casefile_get_destructive_reader (r->file);
795          if (!casereader_read_xfer (r->reader, &r->ccase))
796          {          {
797            run_cnt--;            run_cnt--;
798            i--;            i--;
799          }          }
800      }      }
801      
802    output_casefile = casefile_create (mrg->xsrt->value_cnt);    /* Create output file. */
803    casefile_to_disk (output_casefile);    output = casefile_create (xsrt->value_cnt);
804      casefile_to_disk (output);
805    
806    /* Merge. */    /* Merge. */
807    while (run_cnt > 0)    while (run_cnt > 0)
808      {      {
809        size_t min_idx;        struct run *min_run, *run;
810          
811        /* Find minimum. */        /* Find minimum. */
812        min_idx = 0;        min_run = runs;
813        for (i = 1; i < run_cnt; i++)        for (run = runs + 1; run < runs + run_cnt; run++)
814          if (compare_record (&input_cases[i], &input_cases[min_idx],          if (compare_record (&run->ccase, &min_run->ccase, xsrt->criteria) < 0)
815                              mrg->xsrt->criteria) < 0)            min_run = run;
           min_idx = i;  
816    
817        /* Write minimum to output file. */        /* Write minimum to output file. */
818        casefile_append_xfer (output_casefile, &input_cases[min_idx]);        casefile_append_xfer (output, &min_run->ccase);
819    
820        if (!casereader_read_xfer (input_readers[min_idx],        /* Read another case from minimum run. */
821                                   &input_cases[min_idx]))        if (!casereader_read_xfer (min_run->reader, &min_run->ccase))
822          {          {
823            casereader_destroy (input_readers[min_idx]);            casereader_destroy (min_run->reader);
824            casefile_destroy (input_runs[min_idx]);            casefile_destroy (min_run->file);
825    
826              remove_element (runs, run_cnt, sizeof *runs, min_run - runs);
827            run_cnt--;            run_cnt--;
           input_runs[min_idx] = input_runs[run_cnt];  
           input_readers[min_idx] = input_readers[run_cnt];  
           input_cases[min_idx] = input_cases[run_cnt];  
828          }          }
829      }      }
830    
831    casefile_sleep (output_casefile);    casefile_sleep (output);
832      free (runs);
833    
834    return output_casefile;    return output;
835  }  }

Legend:
Removed from v.1.26  
changed lines
  Added in v.1.27

savannah-hackers-public@gnu.org
ViewVC Help
Powered by ViewVC 1.1.26