1 /* PSPP - computes sample statistics.
2 Copyright (C) 1997-9, 2000 Free Software Foundation, Inc.
3 Written by Ben Pfaff <blp@gnu.org>.
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.
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.
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
26 #include "algorithm.h"
44 #include <sys/types.h>
51 #include "debug-print.h"
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 *,
59 static void destroy_internal_sort (struct internal_sort *);
60 static struct external_sort *do_external_sort (struct sort_cases_pgm *,
62 static void destroy_external_sort (struct external_sort *);
63 struct sort_cases_pgm *parse_sort (void);
65 /* Performs the SORT CASES procedures. */
69 struct sort_cases_pgm *scp;
72 lex_match_id ("SORT");
73 lex_match_id ("CASES");
82 success = sort_cases (scp, 0);
83 destroy_sort_cases_pgm (scp);
85 return lex_end_of_command ();
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 *
95 struct sort_cases_pgm *scp;
97 scp = xmalloc (sizeof *scp);
107 int prev_var_cnt = scp->var_cnt;
108 enum sort_direction direction = SRT_ASCEND;
111 if (!parse_variables (default_dict, &scp->vars, &scp->var_cnt,
112 PV_NO_DUPLICATE | PV_APPEND | PV_NO_SCRATCH))
115 /* Sort direction. */
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"))
122 msg (SE, _("`A' or `D' expected inside parentheses."));
125 if (!lex_match (')'))
127 msg (SE, _("`)' expected."));
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;
135 while (token != '.' && token != '/');
140 destroy_sort_cases_pgm (scp);
145 destroy_sort_cases_pgm (struct sort_cases_pgm *scp)
149 assert (scp->ref_cnt > 0);
150 if (--scp->ref_cnt > 0)
155 destroy_internal_sort (scp->isrt);
156 destroy_external_sort (scp->xsrt);
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.)
168 The caller is responsible for freeing vars[]. */
170 sort_cases (struct sort_cases_pgm *scp, int separate)
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);
176 /* SORT CASES cancels PROCESS IF. */
177 expr_free (process_if_expr);
178 process_if_expr = NULL;
180 /* Try an internal sort first. */
181 scp->isrt = do_internal_sort (scp, separate);
182 if (scp->isrt != NULL)
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)
191 destroy_sort_cases_pgm (scp);
197 struct case_list **results;
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)
205 struct internal_sort *isrt;
207 isrt = xmalloc (sizeof *isrt);
208 isrt->results = NULL;
210 if (!case_source_is_class (vfm_source, &disk_source_class))
212 if (!case_source_is_class (vfm_source, &memory_source_class))
213 procedure (NULL, NULL, NULL, NULL);
215 if (case_source_is_class (vfm_source, &memory_source_class))
217 struct case_list *case_list;
218 struct case_list **case_array;
222 case_cnt = vfm_source_info.ncases;
226 if (case_cnt > set_max_workspace / sizeof *case_array)
229 case_list = memory_source_get_cases (vfm_source);
230 case_array = malloc (sizeof *case_array * (case_cnt + 1));
231 if (case_array == NULL)
234 for (i = 0; case_list != NULL; i++)
236 case_array[i] = case_list;
237 case_list = case_list->next;
239 assert (i == case_cnt);
240 case_array[case_cnt] = NULL;
242 sort (case_array, case_cnt, sizeof *case_array,
243 compare_case_lists, scp);
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];
253 isrt->results = case_array;
265 destroy_internal_sort (struct internal_sort *isrt)
269 free (isrt->results);
274 /* Compares the VAR_CNT variables in VARS[] between the
275 `case_list's at A and B, and returns a strcmp()-type
278 compare_case_lists (const void *a_, const void *b_, void *scp_)
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;
286 return compare_record (a->c.data, b->c.data, scp);
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
295 /* Minimum total number of records for buffers. */
296 #define MIN_BUFFER_TOTAL_SIZE_RECS 64
298 /* Minimum single input buffer size, in bytes and records. */
299 #define MIN_BUFFER_SIZE_BYTES 4096
300 #define MIN_BUFFER_SIZE_RECS 16
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.
306 /* An initial run and its length. */
309 int file_idx; /* File index. */
310 size_t case_cnt; /* Number of cases. */
313 /* Sorts initial runs A and B in decending order by length. */
315 compare_initial_runs (const void *a_, const void *b_, void *aux UNUSED)
317 const struct initial_run *a = a_;
318 const struct initial_run *b = b_;
320 return a->case_cnt > b->case_cnt ? -1 : a->case_cnt <b->case_cnt;
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. */
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);
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)
348 struct external_sort *xsrt;
351 xsrt = xmalloc (sizeof *xsrt);
353 if (!init_external_sort (xsrt))
355 if (!write_initial_runs (xsrt, separate))
365 /* Don't destroy anything because we'll need it for reading
371 destroy_external_sort (xsrt);
378 destroy_external_sort (struct external_sort *xsrt)
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);
393 /* Creates and returns the name of a temporary directory. */
397 const char *parent_dir;
400 if (getenv ("TMPDIR") != NULL)
401 parent_dir = getenv ("TMPDIR");
403 parent_dir = P_tmpdir;
405 temp_dir = xmalloc (strlen (parent_dir) + 32);
406 sprintf (temp_dir, "%s%cpsppXXXXXX", parent_dir, DIR_SEPARATOR);
407 if (mkdtemp (temp_dir) == NULL)
409 msg (SE, _("%s: Creating temporary directory: %s."),
410 temp_dir, strerror (errno));
417 #else /* !HAVE_MKDTEMP */
418 /* Creates directory DIR. */
420 do_mkdir (const char *dir)
423 return mkdir (dir, S_IRWXU);
429 /* Creates and returns the name of a temporary directory. */
435 for (i = 0; i < 100; i++)
437 char temp_dir[L_tmpnam + 1];
438 if (tmpnam (temp_dir) == NULL)
440 msg (SE, _("Generating temporary directory name failed: %s."),
444 else if (do_mkdir (temp_dir) == 0)
445 return xstrdup (temp_dir);
448 msg (SE, _("Creating temporary directory failed: %s."), strerror (errno));
451 #endif /* !HAVE_MKDTEMP */
453 /* Sets up to open temporary files. */
455 init_external_sort (struct external_sort *xsrt)
458 xsrt->temp_dir = NULL;
459 xsrt->next_file_idx = 0;
464 xsrt->initial_runs = xmalloc (sizeof *xsrt->initial_runs * xsrt->run_cap);
466 /* Temporary directory. */
467 xsrt->temp_dir = make_temp_dir ();
468 if (xsrt->temp_dir == NULL)
476 simulate_error (void)
478 static int op_err_cnt = -1;
481 if (op_err_cnt == -1 || op_cnt++ < op_err_cnt)
490 /* Removes the directory created for temporary files, if one
493 rmdir_temp_dir (struct external_sort *xsrt)
495 if (xsrt->temp_dir != NULL && rmdir (xsrt->temp_dir) == -1)
497 msg (SE, _("%s: Error removing directory for temporary files: %s."),
498 xsrt->temp_dir, strerror (errno));
499 xsrt->temp_dir = NULL;
503 #define TEMP_FILE_NAME_SIZE (L_tmpnam + 32)
505 get_temp_file_name (struct external_sort *xsrt, int file_idx,
506 char filename[TEMP_FILE_NAME_SIZE])
508 assert (xsrt->temp_dir != NULL);
509 sprintf (filename, "%s%c%04d", xsrt->temp_dir, DIR_SEPARATOR, file_idx);
513 open_temp_file (struct external_sort *xsrt, int file_idx, const char *mode)
515 char temp_file[TEMP_FILE_NAME_SIZE];
518 get_temp_file_name (xsrt, file_idx, temp_file);
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",
530 close_temp_file (struct external_sort *xsrt, int file_idx, FILE *file)
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)
538 msg (SE, _("%s: Error closing temporary file: %s."),
539 temp_file, strerror (errno));
547 remove_temp_file (struct external_sort *xsrt, int file_idx)
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));
560 write_temp_file (struct external_sort *xsrt, int file_idx,
561 FILE *file, const void *data, size_t size)
563 if (!simulate_error () && fwrite (data, size, 1, file) == 1)
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));
576 read_temp_file (struct external_sort *xsrt, int file_idx,
577 FILE *file, void *data, size_t size)
579 if (!simulate_error () && fread (data, size, 1, file) == 1)
583 char temp_file[TEMP_FILE_NAME_SIZE];
584 get_temp_file_name (xsrt, file_idx, temp_file);
586 msg (SE, _("%s: Error reading temporary file: %s."),
587 temp_file, strerror (errno));
589 msg (SE, _("%s: Unexpected end of temporary file."),
595 /* Replacement selection. */
597 /* Pairs a record with a run number. */
600 int run; /* Run number of case. */
601 struct case_list *record; /* Case data. */
604 struct initial_run_state
606 struct external_sort *xsrt;
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. */
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. */
620 int okay; /* Zero if an error has been encountered. */
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 *);
636 write_initial_runs (struct external_sort *xsrt, int separate)
638 struct initial_run_state *irs;
641 /* Allocate memory for cases. */
642 irs = xmalloc (sizeof *irs);
645 irs->record_cnt = irs->record_cap = 0;
646 irs->free_list = NULL;
647 irs->output_file = NULL;
648 irs->last_output = NULL;
652 if (!allocate_cases (irs))
655 /* Create case sink. */
659 vfm_sink->class->destroy (vfm_sink);
660 vfm_sink = create_case_sink (&sort_sink_class, irs);
661 xsrt->scp->ref_cnt++;
664 /* Create initial runs. */
666 procedure (NULL, NULL, NULL, NULL);
667 while (irs->record_cnt > 0 && irs->okay)
674 destroy_initial_run_state (irs);
679 /* Add a single case to an initial run. */
681 sort_sink_write (struct case_sink *sink, struct ccase *c)
683 struct initial_run_state *irs = sink->aux;
684 struct record_run *new_record_run;
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,
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);
702 /* Output a record if the reservoir is full. */
703 if (irs->record_cnt == irs->record_cap && irs->okay)
708 destroy_initial_run_state (struct initial_run_state *irs)
710 struct case_list *iter, *next;
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);
722 /* Free cases in free list. */
723 for (iter = irs->free_list; iter != NULL; iter = next)
732 if (irs->output_file != NULL)
733 close_temp_file (irs->xsrt, irs->file_idx, irs->output_file);
736 /* Allocates room for lots of cases as a buffer. */
738 allocate_cases (struct initial_run_state *irs)
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. */
745 /* Allocate as many cases as we can within the workspace
747 case_size = dict_get_case_size (default_dict);
748 approx_case_cost = (sizeof *irs->records
749 + sizeof *irs->free_list
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++)
757 c = malloc (sizeof *c + case_size - sizeof (union value));
763 release_case (irs, c);
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;
770 /* Fail if we didn't allocate an acceptable number of cases. */
771 if (irs->records == NULL || max_cases < MIN_BUFFER_TOTAL_SIZE_RECS)
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);
782 /* Compares the VAR_CNT variables in VARS[] between the `value's at
783 A and B, and returns a strcmp()-type result. */
785 compare_record (const union value *a, const union value *b,
786 const struct sort_cases_pgm *scp)
793 for (i = 0; i < scp->var_cnt; i++)
795 struct variable *v = scp->vars[i];
799 if (v->type == NUMERIC)
804 result = af < bf ? -1 : af > bf;
807 result = memcmp (a[fv].s, b[fv].s, v->width);
811 if (scp->dirs[i] == SRT_DESCEND)
821 compare_record_run (const struct record_run *a,
822 const struct record_run *b,
823 struct sort_cases_pgm *scp)
825 if (a->run != b->run)
826 return a->run > b->run ? 1 : -1;
828 return compare_record (a->record->c.data, b->record->c.data, scp);
832 compare_record_run_minheap (const void *a, const void *b, void *scp)
834 return -compare_record_run (a, b, scp);
837 /* Begins a new initial run, specifically its output file. */
839 start_run (struct initial_run_state *irs)
841 irs->file_idx = irs->xsrt->next_file_idx++;
843 irs->output_file = open_temp_file (irs->xsrt, irs->file_idx, "wb");
844 if (irs->output_file == NULL)
846 if (irs->last_output != NULL)
848 release_case (irs, irs->last_output);
849 irs->last_output = NULL;
853 /* Ends the current initial run. */
855 end_run (struct initial_run_state *irs)
857 struct external_sort *xsrt = irs->xsrt;
859 /* Record initial run. */
860 if (xsrt->run_cnt >= xsrt->run_cap)
864 = xrealloc (xsrt->initial_runs,
865 sizeof *xsrt->initial_runs * xsrt->run_cap);
867 xsrt->initial_runs[xsrt->run_cnt].file_idx = irs->file_idx;
868 xsrt->initial_runs[xsrt->run_cnt].case_cnt = irs->case_cnt;
871 /* Close file handle. */
872 if (irs->output_file != NULL
873 && !close_temp_file (irs->xsrt, irs->file_idx, irs->output_file))
875 irs->output_file = NULL;
879 output_record (struct initial_run_state *irs)
881 struct record_run *record_run;
882 struct ccase *out_case;
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;
890 /* Bail if an error has occurred. */
894 /* Obtain case data to write to disk. */
895 out_case = &record_run->record->c;
896 if (compaction_necessary)
898 compact_case (compaction_case, out_case);
899 out_case = compaction_case;
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)
910 assert (record_run->run == irs->file_idx);
914 if (irs->output_file != NULL
915 && !write_temp_file (irs->xsrt, irs->file_idx, irs->output_file,
917 sizeof *out_case->data * compaction_nval))
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;
926 static struct case_list *
927 grab_case (struct initial_run_state *irs)
931 assert (irs != NULL);
932 assert (irs->free_list != NULL);
935 irs->free_list = c->next;
940 release_case (struct initial_run_state *irs, struct case_list *c)
942 assert (irs != NULL);
945 c->next = irs->free_list;
953 struct external_sort *xsrt; /* External sort state. */
954 struct ccase **cases; /* Buffers. */
955 size_t case_cnt; /* Number of buffers. */
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);
965 /* Performs a series of P-way merges of initial runs
968 merge (struct external_sort *xsrt)
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. */
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)
987 for (i = 0; i < mrg.case_cnt; i++)
989 mrg.cases[i] = malloc (case_size);
990 if (mrg.cases[i] == NULL)
996 if (mrg.case_cnt < MIN_BUFFER_TOTAL_SIZE_RECS)
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);
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);
1013 if (max_order > xsrt->run_cnt)
1014 max_order = xsrt->run_cnt;
1016 /* Repeatedly merge the P shortest existing runs until only one run
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)
1025 struct initial_run output_run;
1029 /* Choose order of merge (max_order after first merge). */
1030 order = max_order - dummy_run_cnt;
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);
1041 if (!merge_once (&mrg, xsrt->initial_runs + xsrt->run_cnt, order,
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);
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);
1058 for (i = 0; i < mrg.case_cnt; i++)
1059 free (mrg.cases[i]);
1065 /* Modulo function as defined by Knuth. */
1073 else if (x > 0 && y > 0)
1075 else if (x < 0 && y > 0)
1076 return y - (-x) % y;
1081 /* A run of data for use in merging. */
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. */
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
1099 merge_once (struct merge_state *mrg,
1100 const struct initial_run input_runs[],
1102 struct initial_run *output_run)
1104 struct run runs[MAX_MERGE_ORDER];
1105 FILE *output_file = NULL;
1110 /* Initialize runs[]. */
1111 for (i = 0; i < run_cnt; i++)
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;
1122 /* Open input files. */
1123 for (i = 0; i < run_cnt; i++)
1125 runs[i].file = open_temp_file (mrg->xsrt, runs[i].file_idx, "rb");
1126 if (runs[i].file == NULL)
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)
1139 /* Prime buffers. */
1140 for (i = 0; i < run_cnt; i++)
1141 if (!fill_run_buffer (mrg, runs + i))
1145 case_size = dict_get_case_size (default_dict);
1148 struct run *min_run;
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)
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))
1163 /* Remove case from buffer. */
1164 if (++min_run->buffer_head >= min_run->buffer_tail)
1166 /* Buffer is empty. Fill from file. */
1167 if (!fill_run_buffer (mrg, min_run))
1170 /* If buffer is still empty, delete its run. */
1171 if (min_run->buffer_head >= min_run->buffer_tail)
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];
1177 /* We could donate the now-unused buffer space to
1183 /* Close output file. */
1184 close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1189 /* Close and remove output file. */
1190 if (output_file != NULL)
1192 close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1193 remove_temp_file (mrg->xsrt, output_run->file_idx);
1196 /* Close and remove any remaining input runs. */
1197 for (i = 0; i < run_cnt; i++)
1199 close_temp_file (mrg->xsrt, runs[i].file_idx, runs[i].file);
1200 remove_temp_file (mrg->xsrt, runs[i].file_idx);
1206 /* Reads as many cases as possible into RUN's buffer.
1207 Reads nonzero unless a disk error occurs. */
1209 fill_run_buffer (struct merge_state *mrg, struct run *run)
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)
1215 if (!read_temp_file (mrg->xsrt, run->file_idx, run->file,
1216 (*run->buffer_tail)->data,
1217 dict_get_case_size (default_dict)))
1220 run->unread_case_cnt--;
1228 sort_sink_destroy (struct case_sink *sink UNUSED)
1233 static struct case_source *
1234 sort_sink_make_source (struct case_sink *sink)
1236 struct initial_run_state *irs = sink->aux;
1238 return create_case_source (&sort_source_class, irs->xsrt->scp);
1241 const struct case_sink_class sort_sink_class =
1247 sort_sink_make_source,
1250 /* Reads all the records from the source stream and passes them
1253 sort_source_read (struct case_source *source,
1254 write_case_func *write_case, write_case_data wc_data)
1256 struct sort_cases_pgm *scp = source->aux;
1258 read_sort_output (scp, write_case, wc_data);
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);
1268 /* Reads all the records from the output stream and passes them to the
1269 function provided, which must have an interface identical to
1272 read_sort_output (struct sort_cases_pgm *scp,
1273 write_case_func *write_case, write_case_data wc_data)
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);
1282 /* No results. Probably an external sort that failed. */
1287 read_internal_sort_output (struct internal_sort *isrt,
1288 write_case_func *write_case,
1289 write_case_data wc_data)
1291 struct ccase *save_temp_case = temp_case;
1292 struct case_list **p;
1294 for (p = isrt->results; *p; p++)
1296 temp_case = &(*p)->c;
1297 write_case (wc_data);
1299 free (isrt->results);
1301 temp_case = save_temp_case;
1305 read_external_sort_output (struct external_sort *xsrt,
1306 write_case_func *write_case,
1307 write_case_data wc_data)
1313 assert (xsrt->run_cnt == 1);
1314 file_idx = xsrt->initial_runs[0].file_idx;
1316 file = open_temp_file (xsrt, file_idx, "rb");
1323 for (i = 0; i < vfm_source_info.ncases; i++)
1325 if (!read_temp_file (xsrt, file_idx, file,
1326 temp_case, vfm_source_info.case_size))
1332 if (!write_case (wc_data))
1338 sort_source_destroy (struct case_source *source)
1340 struct sort_cases_pgm *scp = source->aux;
1342 destroy_sort_cases_pgm (scp);
1345 const struct case_source_class sort_source_class =
1349 sort_source_destroy,