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 *, int *idx_to_fv);
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;
80 msg (SE, _("SORT CASES may not be used after TEMPORARY. "
81 "Temporary transformations will be made permanent."));
85 success = sort_cases (scp, 0);
86 destroy_sort_cases_pgm (scp);
88 return lex_end_of_command ();
93 /* Parses a list of sort keys and returns a struct sort_cases_pgm
94 based on it. Returns a null pointer on error. */
95 struct sort_cases_pgm *
98 struct sort_cases_pgm *scp;
100 scp = xmalloc (sizeof *scp);
110 int prev_var_cnt = scp->var_cnt;
111 enum sort_direction direction = SRT_ASCEND;
114 if (!parse_variables (default_dict, &scp->vars, &scp->var_cnt,
115 PV_NO_DUPLICATE | PV_APPEND | PV_NO_SCRATCH))
118 /* Sort direction. */
121 if (lex_match_id ("D") || lex_match_id ("DOWN"))
122 direction = SRT_DESCEND;
123 else if (!lex_match_id ("A") && !lex_match_id ("UP"))
125 msg (SE, _("`A' or `D' expected inside parentheses."));
128 if (!lex_match (')'))
130 msg (SE, _("`)' expected."));
134 scp->dirs = xrealloc (scp->dirs, sizeof *scp->dirs * scp->var_cnt);
135 for (; prev_var_cnt < scp->var_cnt; prev_var_cnt++)
136 scp->dirs[prev_var_cnt] = direction;
138 while (token != '.' && token != '/');
143 destroy_sort_cases_pgm (scp);
147 /* Destroys a SORT CASES program. */
149 destroy_sort_cases_pgm (struct sort_cases_pgm *scp)
153 assert (scp->ref_cnt > 0);
154 if (--scp->ref_cnt > 0)
159 destroy_internal_sort (scp->isrt);
160 destroy_external_sort (scp->xsrt);
165 /* Sorts the active file based on the key variables specified in
166 global variables vars and var_cnt. The output is either to the
167 active file, if SEPARATE is zero, or to a separate file, if
168 SEPARATE is nonzero. In the latter case the output cases can be
169 read with a call to read_sort_output(). (In the former case the
170 output cases should be dealt with through the usual vfm interface.)
172 The caller is responsible for freeing vars[]. */
174 sort_cases (struct sort_cases_pgm *scp, int separate)
177 = sizeof (union value) * dict_get_compacted_value_cnt (default_dict);
179 /* Not sure this is necessary but it's good to be safe. */
180 if (separate && case_source_is_class (vfm_source, &sort_source_class))
181 procedure (NULL, NULL);
183 /* SORT CASES cancels PROCESS IF. */
184 expr_free (process_if_expr);
185 process_if_expr = NULL;
187 /* Try an internal sort first. */
188 scp->isrt = do_internal_sort (scp, separate);
189 if (scp->isrt != NULL)
192 /* Fall back to an external sort. */
193 if (vfm_source != NULL
194 && case_source_is_class (vfm_source, &storage_source_class))
195 storage_source_to_disk (vfm_source);
196 scp->xsrt = do_external_sort (scp, separate);
197 if (scp->xsrt != NULL)
200 destroy_sort_cases_pgm (scp);
204 /* Results of an internal sort. */
207 struct case_list **results;
210 /* If the data is in memory, do an internal sort. Return
212 static struct internal_sort *
213 do_internal_sort (struct sort_cases_pgm *scp, int separate)
215 struct internal_sort *isrt;
217 isrt = xmalloc (sizeof *isrt);
218 isrt->results = NULL;
220 if (case_source_is_class (vfm_source, &storage_source_class)
221 && !storage_source_on_disk (vfm_source))
223 struct case_list *case_list;
224 struct case_list **case_array;
228 case_cnt = vfm_source->class->count (vfm_source);
232 if (case_cnt > get_max_workspace() / sizeof *case_array)
235 case_list = storage_source_get_cases (vfm_source);
236 case_array = malloc (sizeof *case_array * (case_cnt + 1));
237 if (case_array == NULL)
240 for (i = 0; case_list != NULL; i++)
242 case_array[i] = case_list;
243 case_list = case_list->next;
245 assert (i == case_cnt);
246 case_array[case_cnt] = NULL;
248 sort (case_array, case_cnt, sizeof *case_array,
249 compare_case_lists, scp);
253 storage_source_set_cases (vfm_source, case_array[0]);
254 for (i = 1; i <= case_cnt; i++)
255 case_array[i - 1]->next = case_array[i];
259 isrt->results = case_array;
269 /* Destroys an internal sort result. */
271 destroy_internal_sort (struct internal_sort *isrt)
275 free (isrt->results);
280 /* Compares the VAR_CNT variables in VARS[] between the
281 `case_list's at A and B, and returns a strcmp()-type
284 compare_case_lists (const void *a_, const void *b_, void *scp_)
286 struct sort_cases_pgm *scp = scp_;
287 struct case_list *const *pa = a_;
288 struct case_list *const *pb = b_;
289 struct case_list *a = *pa;
290 struct case_list *b = *pb;
292 return compare_record (a->c.data, b->c.data, scp, NULL);
297 /* Maximum order of merge. If you increase this, then you should
298 use a heap for comparing cases during merge. */
299 #define MAX_MERGE_ORDER 7
301 /* Minimum total number of records for buffers. */
302 #define MIN_BUFFER_TOTAL_SIZE_RECS 64
304 /* Minimum single input buffer size, in bytes and records. */
305 #define MIN_BUFFER_SIZE_BYTES 4096
306 #define MIN_BUFFER_SIZE_RECS 16
308 #if MIN_BUFFER_SIZE_RECS * 2 + 16 > MIN_BUFFER_TOTAL_SIZE_RECS
309 #error MIN_BUFFER_SIZE_RECS and MIN_BUFFER_TOTAL_SIZE_RECS do not make sense.
312 /* An initial run and its length. */
315 int file_idx; /* File index. */
316 size_t case_cnt; /* Number of cases. */
319 /* Sorts initial runs A and B in decending order by length. */
321 compare_initial_runs (const void *a_, const void *b_, void *aux UNUSED)
323 const struct initial_run *a = a_;
324 const struct initial_run *b = b_;
326 return a->case_cnt > b->case_cnt ? -1 : a->case_cnt <b->case_cnt;
329 /* Results of an external sort. */
332 struct sort_cases_pgm *scp; /* SORT CASES info. */
333 struct initial_run *initial_runs; /* Array of initial runs. */
334 size_t run_cnt, run_cap; /* Number of runs, allocated capacity. */
335 char *temp_dir; /* Temporary file directory name. */
336 char *temp_name; /* Name of a temporary file. */
337 int next_file_idx; /* Lowest unused file index. */
340 /* Prototypes for helper functions. */
341 static void sort_sink_write (struct case_sink *, const struct ccase *);
342 static int write_initial_runs (struct external_sort *, int separate);
343 static int init_external_sort (struct external_sort *);
344 static int merge (struct external_sort *);
345 static void rmdir_temp_dir (struct external_sort *);
346 static void remove_temp_file (struct external_sort *xsrt, int file_idx);
348 /* Performs an external sort of the active file according to the
349 specification in SCP. Forms initial runs using a heap as a
350 reservoir. Determines the optimum merge pattern via Huffman's
351 method (see Knuth vol. 3, 2nd edition, p. 365-366), and merges
352 according to that pattern. */
353 static struct external_sort *
354 do_external_sort (struct sort_cases_pgm *scp, int separate)
356 struct external_sort *xsrt;
359 xsrt = xmalloc (sizeof *xsrt);
361 if (!init_external_sort (xsrt))
363 if (!write_initial_runs (xsrt, separate))
373 /* Don't destroy anything because we'll need it for reading
379 destroy_external_sort (xsrt);
386 destroy_external_sort (struct external_sort *xsrt)
392 for (i = 0; i < xsrt->run_cnt; i++)
393 remove_temp_file (xsrt, xsrt->initial_runs[i].file_idx);
394 rmdir_temp_dir (xsrt);
395 free (xsrt->temp_dir);
396 free (xsrt->temp_name);
397 free (xsrt->initial_runs);
403 /* Creates and returns the name of a temporary directory. */
407 const char *parent_dir;
410 if (getenv ("TMPDIR") != NULL)
411 parent_dir = getenv ("TMPDIR");
413 parent_dir = P_tmpdir;
415 temp_dir = xmalloc (strlen (parent_dir) + 32);
416 sprintf (temp_dir, "%s%cpsppXXXXXX", parent_dir, DIR_SEPARATOR);
417 if (mkdtemp (temp_dir) == NULL)
419 msg (SE, _("%s: Creating temporary directory: %s."),
420 temp_dir, strerror (errno));
427 #else /* !HAVE_MKDTEMP */
428 /* Creates directory DIR. */
430 do_mkdir (const char *dir)
433 return mkdir (dir, S_IRWXU);
439 /* Creates and returns the name of a temporary directory. */
445 for (i = 0; i < 100; i++)
447 char temp_dir[L_tmpnam + 1];
448 if (tmpnam (temp_dir) == NULL)
450 msg (SE, _("Generating temporary directory name failed: %s."),
454 else if (do_mkdir (temp_dir) == 0)
455 return xstrdup (temp_dir);
458 msg (SE, _("Creating temporary directory failed: %s."), strerror (errno));
461 #endif /* !HAVE_MKDTEMP */
463 /* Sets up to open temporary files. */
465 init_external_sort (struct external_sort *xsrt)
468 xsrt->temp_dir = NULL;
469 xsrt->next_file_idx = 0;
474 xsrt->initial_runs = xmalloc (sizeof *xsrt->initial_runs * xsrt->run_cap);
476 /* Temporary directory. */
477 xsrt->temp_dir = make_temp_dir ();
478 xsrt->temp_name = NULL;
479 if (xsrt->temp_dir == NULL)
481 xsrt->temp_name = xmalloc (strlen (xsrt->temp_dir) + 64);
486 /* Returns nonzero if we should return an error even though the
487 operation succeeded. Useful for testing. */
489 simulate_error (void)
491 static int op_err_cnt = -1;
494 if (op_err_cnt == -1 || op_cnt++ < op_err_cnt)
503 /* Removes the directory created for temporary files, if one
506 rmdir_temp_dir (struct external_sort *xsrt)
508 if (xsrt->temp_dir != NULL && rmdir (xsrt->temp_dir) == -1)
510 msg (SE, _("%s: Error removing directory for temporary files: %s."),
511 xsrt->temp_dir, strerror (errno));
512 xsrt->temp_dir = NULL;
516 /* Returns the name of temporary file number FILE_IDX for XSRT.
517 The name is written into a static buffer, so be careful. */
519 get_temp_file_name (struct external_sort *xsrt, int file_idx)
521 assert (xsrt->temp_dir != NULL);
522 sprintf (xsrt->temp_name, "%s%c%04d",
523 xsrt->temp_dir, DIR_SEPARATOR, file_idx);
524 return xsrt->temp_name;
527 /* Opens temporary file numbered FILE_IDX for XSRT with mode MODE
528 and returns the FILE *. */
530 open_temp_file (struct external_sort *xsrt, int file_idx, const char *mode)
535 temp_file = get_temp_file_name (xsrt, file_idx);
537 file = fopen (temp_file, mode);
538 if (simulate_error () || file == NULL)
539 msg (SE, _("%s: Error opening temporary file for %s: %s."),
540 temp_file, mode[0] == 'r' ? "reading" : "writing",
546 /* Closes FILE, which is the temporary file numbered FILE_IDX
547 under XSRT. Returns nonzero only if successful. */
549 close_temp_file (struct external_sort *xsrt, int file_idx, FILE *file)
553 char *temp_file = get_temp_file_name (xsrt, file_idx);
554 if (simulate_error () || fclose (file) == EOF)
556 msg (SE, _("%s: Error closing temporary file: %s."),
557 temp_file, strerror (errno));
564 /* Delete temporary file numbered FILE_IDX for XSRT. */
566 remove_temp_file (struct external_sort *xsrt, int file_idx)
570 char *temp_file = get_temp_file_name (xsrt, file_idx);
571 if (simulate_error () || remove (temp_file) != 0)
572 msg (SE, _("%s: Error removing temporary file: %s."),
573 temp_file, strerror (errno));
577 /* Writes SIZE bytes from buffer DATA into FILE, which is
578 temporary file numbered FILE_IDX from XSRT. */
580 write_temp_file (struct external_sort *xsrt, int file_idx,
581 FILE *file, const void *data, size_t size)
583 if (!simulate_error () && fwrite (data, size, 1, file) == 1)
587 char *temp_file = get_temp_file_name (xsrt, file_idx);
588 msg (SE, _("%s: Error writing temporary file: %s."),
589 temp_file, strerror (errno));
594 /* Reads SIZE bytes into buffer DATA into FILE, which is
595 temporary file numbered FILE_IDX from XSRT. */
597 read_temp_file (struct external_sort *xsrt, int file_idx,
598 FILE *file, void *data, size_t size)
600 if (!simulate_error () && fread (data, size, 1, file) == 1)
604 char *temp_file = get_temp_file_name (xsrt, file_idx);
606 msg (SE, _("%s: Error reading temporary file: %s."),
607 temp_file, strerror (errno));
609 msg (SE, _("%s: Unexpected end of temporary file."),
615 /* Replacement selection. */
617 /* Pairs a record with a run number. */
620 int run; /* Run number of case. */
621 struct case_list *record; /* Case data. */
624 /* Represents a set of initial runs during an external sort. */
625 struct initial_run_state
627 struct external_sort *xsrt;
629 int *idx_to_fv; /* Translation table copied from sink. */
632 struct record_run *records; /* Records arranged as a heap. */
633 size_t record_cnt; /* Current number of records. */
634 size_t record_cap; /* Capacity for records. */
635 struct case_list *free_list;/* Cases not in heap. */
637 /* Run currently being output. */
638 int file_idx; /* Temporary file number. */
639 size_t case_cnt; /* Number of cases so far. */
640 FILE *output_file; /* Output file. */
641 struct case_list *last_output;/* Record last output. */
643 int okay; /* Zero if an error has been encountered. */
646 static const struct case_sink_class sort_sink_class;
648 static void destroy_initial_run_state (struct initial_run_state *irs);
649 static int allocate_cases (struct initial_run_state *);
650 static struct case_list *grab_case (struct initial_run_state *);
651 static void release_case (struct initial_run_state *, struct case_list *);
652 static void output_record (struct initial_run_state *irs);
653 static void start_run (struct initial_run_state *irs);
654 static void end_run (struct initial_run_state *irs);
655 static int compare_record_run (const struct record_run *,
656 const struct record_run *,
657 struct initial_run_state *);
658 static int compare_record_run_minheap (const void *, const void *, void *);
660 /* Writes initial runs for XSRT, sending them to a separate file
661 if SEPARATE is nonzero. */
663 write_initial_runs (struct external_sort *xsrt, int separate)
665 struct initial_run_state *irs;
668 /* Allocate memory for cases. */
669 irs = xmalloc (sizeof *irs);
672 irs->record_cnt = irs->record_cap = 0;
673 irs->free_list = NULL;
674 irs->output_file = NULL;
675 irs->last_output = NULL;
679 if (!allocate_cases (irs))
682 /* Create case sink. */
685 if (vfm_sink != NULL && vfm_sink->class->destroy != NULL)
686 vfm_sink->class->destroy (vfm_sink);
687 vfm_sink = create_case_sink (&sort_sink_class, default_dict, irs);
688 xsrt->scp->ref_cnt++;
691 /* Create initial runs. */
693 procedure (NULL, NULL);
694 irs->idx_to_fv = NULL;
695 while (irs->record_cnt > 0 && irs->okay)
702 destroy_initial_run_state (irs);
707 /* Add a single case to an initial run. */
709 sort_sink_write (struct case_sink *sink, const struct ccase *c)
711 struct initial_run_state *irs = sink->aux;
712 struct record_run *new_record_run;
717 irs->idx_to_fv = sink->idx_to_fv;
719 /* Compose record_run for this run and add to heap. */
720 assert (irs->record_cnt < irs->record_cap);
721 new_record_run = irs->records + irs->record_cnt++;
722 new_record_run->record = grab_case (irs);
723 memcpy (new_record_run->record->c.data, c->data, irs->xsrt->scp->case_size);
724 new_record_run->run = irs->file_idx;
725 if (irs->last_output != NULL
726 && compare_record (c->data, irs->last_output->c.data,
727 irs->xsrt->scp, sink->idx_to_fv) < 0)
728 new_record_run->run = irs->xsrt->next_file_idx;
729 push_heap (irs->records, irs->record_cnt, sizeof *irs->records,
730 compare_record_run_minheap, irs);
732 /* Output a record if the reservoir is full. */
733 if (irs->record_cnt == irs->record_cap && irs->okay)
737 /* Destroys the initial run state represented by IRS. */
739 destroy_initial_run_state (struct initial_run_state *irs)
741 struct case_list *iter, *next;
747 /* Release cases to free list. */
748 for (i = 0; i < irs->record_cnt; i++)
749 release_case (irs, irs->records[i].record);
750 if (irs->last_output != NULL)
751 release_case (irs, irs->last_output);
753 /* Free cases in free list. */
754 for (iter = irs->free_list; iter != NULL; iter = next)
761 if (irs->output_file != NULL)
762 close_temp_file (irs->xsrt, irs->file_idx, irs->output_file);
767 /* Allocates room for lots of cases as a buffer. */
769 allocate_cases (struct initial_run_state *irs)
771 int approx_case_cost; /* Approximate memory cost of one case in bytes. */
772 int max_cases; /* Maximum number of cases to allocate. */
775 /* Allocate as many cases as we can within the workspace
777 approx_case_cost = (sizeof *irs->records
778 + sizeof *irs->free_list
779 + irs->xsrt->scp->case_size
780 + 4 * sizeof (void *));
781 max_cases = get_max_workspace() / approx_case_cost;
782 irs->records = malloc (sizeof *irs->records * max_cases);
783 for (i = 0; i < max_cases; i++)
786 c = malloc (sizeof *c
787 + irs->xsrt->scp->case_size
788 - sizeof (union value));
794 release_case (irs, c);
797 /* irs->records gets all but one of the allocated cases.
798 The extra is used for last_output. */
799 irs->record_cap = max_cases - 1;
801 /* Fail if we didn't allocate an acceptable number of cases. */
802 if (irs->records == NULL || max_cases < MIN_BUFFER_TOTAL_SIZE_RECS)
804 msg (SE, _("Out of memory. Could not allocate room for minimum of %d "
805 "cases of %d bytes each. (PSPP workspace is currently "
806 "restricted to a maximum of %d KB.)"),
807 MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, get_max_workspace() / 1024);
813 /* Compares the VAR_CNT variables in VARS[] between the `value's at
814 A and B, and returns a strcmp()-type result. */
816 compare_record (const union value *a, const union value *b,
817 const struct sort_cases_pgm *scp,
825 for (i = 0; i < scp->var_cnt; i++)
827 struct variable *v = scp->vars[i];
831 if (idx_to_fv != NULL)
832 fv = idx_to_fv[v->index];
836 if (v->type == NUMERIC)
841 result = af < bf ? -1 : af > bf;
844 result = memcmp (a[fv].s, b[fv].s, v->width);
848 if (scp->dirs[i] == SRT_DESCEND)
857 /* Compares record-run tuples A and B on run number first, then
858 on the current record according to SCP. */
860 compare_record_run (const struct record_run *a,
861 const struct record_run *b,
862 struct initial_run_state *irs)
864 if (a->run != b->run)
865 return a->run > b->run ? 1 : -1;
867 return compare_record (a->record->c.data, b->record->c.data,
868 irs->xsrt->scp, irs->idx_to_fv);
871 /* Compares record-run tuples A and B on run number first, then
872 on the current record according to SCP, but in descending
875 compare_record_run_minheap (const void *a, const void *b, void *irs)
877 return -compare_record_run (a, b, irs);
880 /* Begins a new initial run, specifically its output file. */
882 start_run (struct initial_run_state *irs)
884 irs->file_idx = irs->xsrt->next_file_idx++;
886 irs->output_file = open_temp_file (irs->xsrt, irs->file_idx, "wb");
887 if (irs->output_file == NULL)
889 if (irs->last_output != NULL)
891 release_case (irs, irs->last_output);
892 irs->last_output = NULL;
896 /* Ends the current initial run. */
898 end_run (struct initial_run_state *irs)
900 struct external_sort *xsrt = irs->xsrt;
902 /* Record initial run. */
903 if (xsrt->run_cnt >= xsrt->run_cap)
907 = xrealloc (xsrt->initial_runs,
908 sizeof *xsrt->initial_runs * xsrt->run_cap);
910 xsrt->initial_runs[xsrt->run_cnt].file_idx = irs->file_idx;
911 xsrt->initial_runs[xsrt->run_cnt].case_cnt = irs->case_cnt;
914 /* Close file handle. */
915 if (irs->output_file != NULL
916 && !close_temp_file (irs->xsrt, irs->file_idx, irs->output_file))
918 irs->output_file = NULL;
921 /* Writes a record to the current initial run. */
923 output_record (struct initial_run_state *irs)
925 struct record_run *record_run;
927 /* Extract minimum case from heap. */
928 assert (irs->record_cnt > 0);
929 pop_heap (irs->records, irs->record_cnt--, sizeof *irs->records,
930 compare_record_run_minheap, irs);
931 record_run = irs->records + irs->record_cnt;
933 /* Bail if an error has occurred. */
937 /* Start new run if necessary. */
938 assert (record_run->run == irs->file_idx
939 || record_run->run == irs->xsrt->next_file_idx);
940 if (record_run->run != irs->file_idx)
945 assert (record_run->run == irs->file_idx);
949 if (irs->output_file != NULL
950 && !write_temp_file (irs->xsrt, irs->file_idx, irs->output_file,
951 &record_run->record->c, irs->xsrt->scp->case_size))
954 /* This record becomes last_output. */
955 if (irs->last_output != NULL)
956 release_case (irs, irs->last_output);
957 irs->last_output = record_run->record;
960 /* Gets a case from the free list in IRS. It is an error to call
961 this function if the free list is empty. */
962 static struct case_list *
963 grab_case (struct initial_run_state *irs)
967 assert (irs != NULL);
968 assert (irs->free_list != NULL);
971 irs->free_list = c->next;
975 /* Returns C to the free list in IRS. */
977 release_case (struct initial_run_state *irs, struct case_list *c)
979 assert (irs != NULL);
982 c->next = irs->free_list;
988 /* State of merging initial runs. */
991 struct external_sort *xsrt; /* External sort state. */
992 struct ccase **cases; /* Buffers. */
993 size_t case_cnt; /* Number of buffers. */
997 static int merge_once (struct merge_state *,
998 const struct initial_run[], size_t,
999 struct initial_run *);
1000 static int fill_run_buffer (struct merge_state *, struct run *);
1001 static int mod (int, int);
1003 /* Performs a series of P-way merges of initial runs
1006 merge (struct external_sort *xsrt)
1008 struct merge_state mrg; /* State of merge. */
1009 size_t approx_case_cost; /* Approximate memory cost of one case. */
1010 int max_order; /* Maximum order of merge. */
1011 size_t dummy_run_cnt; /* Number of dummy runs to insert. */
1017 /* Allocate as many cases as possible into cases. */
1018 approx_case_cost = (sizeof *mrg.cases
1019 + xsrt->scp->case_size + 4 * sizeof (void *));
1020 mrg.case_cnt = get_max_workspace() / approx_case_cost;
1021 mrg.cases = malloc (sizeof *mrg.cases * mrg.case_cnt);
1022 if (mrg.cases == NULL)
1024 for (i = 0; i < mrg.case_cnt; i++)
1026 mrg.cases[i] = malloc (xsrt->scp->case_size);
1027 if (mrg.cases[i] == NULL)
1033 if (mrg.case_cnt < MIN_BUFFER_TOTAL_SIZE_RECS)
1035 msg (SE, _("Out of memory. Could not allocate room for minimum of %d "
1036 "cases of %d bytes each. (PSPP workspace is currently "
1037 "restricted to a maximum of %d KB.)"),
1038 MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, get_max_workspace() / 1024);
1042 /* Determine maximum order of merge. */
1043 max_order = MAX_MERGE_ORDER;
1044 if (mrg.case_cnt / max_order < MIN_BUFFER_SIZE_RECS)
1045 max_order = mrg.case_cnt / MIN_BUFFER_SIZE_RECS;
1046 else if (mrg.case_cnt / max_order * xsrt->scp->case_size
1047 < MIN_BUFFER_SIZE_BYTES)
1048 max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES / xsrt->scp->case_size);
1051 if (max_order > xsrt->run_cnt)
1052 max_order = xsrt->run_cnt;
1054 /* Repeatedly merge the P shortest existing runs until only one run
1056 make_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
1057 compare_initial_runs, NULL);
1058 dummy_run_cnt = mod (1 - (int) xsrt->run_cnt, max_order - 1);
1059 assert (max_order == 1
1060 || (xsrt->run_cnt + dummy_run_cnt) % (max_order - 1) == 1);
1061 while (xsrt->run_cnt > 1)
1063 struct initial_run output_run;
1067 /* Choose order of merge (max_order after first merge). */
1068 order = max_order - dummy_run_cnt;
1071 /* Choose runs to merge. */
1072 assert (xsrt->run_cnt >= order);
1073 for (i = 0; i < order; i++)
1074 pop_heap (xsrt->initial_runs, xsrt->run_cnt--,
1075 sizeof *xsrt->initial_runs,
1076 compare_initial_runs, NULL);
1079 if (!merge_once (&mrg, xsrt->initial_runs + xsrt->run_cnt, order,
1083 /* Add output run to heap. */
1084 xsrt->initial_runs[xsrt->run_cnt++] = output_run;
1085 push_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
1086 compare_initial_runs, NULL);
1089 /* Exactly one run is left, which contains the entire sorted
1090 file. We could use it to find a total case count. */
1091 assert (xsrt->run_cnt == 1);
1096 for (i = 0; i < mrg.case_cnt; i++)
1097 free (mrg.cases[i]);
1103 /* Modulo function as defined by Knuth. */
1111 else if (x > 0 && y > 0)
1113 else if (x < 0 && y > 0)
1114 return y - (-x) % y;
1119 /* A run of data for use in merging. */
1122 FILE *file; /* File that contains run. */
1123 int file_idx; /* Index of file that contains run. */
1124 struct ccase **buffer; /* Case buffer. */
1125 struct ccase **buffer_head; /* First unconsumed case in buffer. */
1126 struct ccase **buffer_tail; /* One past last unconsumed case in buffer. */
1127 size_t buffer_cap; /* Number of cases buffer can hold. */
1128 size_t unread_case_cnt; /* Number of cases not yet read. */
1131 /* Merges the RUN_CNT initial runs specified in INPUT_RUNS into a
1132 new run. Returns nonzero only if successful. Adds an entry
1133 to MRG->xsrt->runs for the output file if and only if the
1134 output file is actually created. Always deletes all the input
1137 merge_once (struct merge_state *mrg,
1138 const struct initial_run input_runs[],
1140 struct initial_run *output_run)
1142 struct run runs[MAX_MERGE_ORDER];
1143 FILE *output_file = NULL;
1147 /* Initialize runs[]. */
1148 for (i = 0; i < run_cnt; i++)
1150 runs[i].file = NULL;
1151 runs[i].file_idx = input_runs[i].file_idx;
1152 runs[i].buffer = mrg->cases + mrg->case_cnt / run_cnt * i;
1153 runs[i].buffer_head = runs[i].buffer;
1154 runs[i].buffer_tail = runs[i].buffer;
1155 runs[i].buffer_cap = mrg->case_cnt / run_cnt;
1156 runs[i].unread_case_cnt = input_runs[i].case_cnt;
1159 /* Open input files. */
1160 for (i = 0; i < run_cnt; i++)
1162 runs[i].file = open_temp_file (mrg->xsrt, runs[i].file_idx, "rb");
1163 if (runs[i].file == NULL)
1167 /* Create output file and count cases to be output. */
1168 output_run->file_idx = mrg->xsrt->next_file_idx++;
1169 output_run->case_cnt = 0;
1170 for (i = 0; i < run_cnt; i++)
1171 output_run->case_cnt += input_runs[i].case_cnt;
1172 output_file = open_temp_file (mrg->xsrt, output_run->file_idx, "wb");
1173 if (output_file == NULL)
1176 /* Prime buffers. */
1177 for (i = 0; i < run_cnt; i++)
1178 if (!fill_run_buffer (mrg, runs + i))
1184 struct run *min_run;
1188 for (i = 1; i < run_cnt; i++)
1189 if (compare_record ((*runs[i].buffer_head)->data,
1190 (*min_run->buffer_head)->data,
1191 mrg->xsrt->scp, NULL) < 0)
1194 /* Write minimum to output file. */
1195 if (!write_temp_file (mrg->xsrt, min_run->file_idx, output_file,
1196 (*min_run->buffer_head)->data,
1197 mrg->xsrt->scp->case_size))
1200 /* Remove case from buffer. */
1201 if (++min_run->buffer_head >= min_run->buffer_tail)
1203 /* Buffer is empty. Fill from file. */
1204 if (!fill_run_buffer (mrg, min_run))
1207 /* If buffer is still empty, delete its run. */
1208 if (min_run->buffer_head >= min_run->buffer_tail)
1210 close_temp_file (mrg->xsrt, min_run->file_idx, min_run->file);
1211 remove_temp_file (mrg->xsrt, min_run->file_idx);
1212 *min_run = runs[--run_cnt];
1214 /* We could donate the now-unused buffer space to
1220 /* Close output file. */
1221 close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1226 /* Close and remove output file. */
1227 if (output_file != NULL)
1229 close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1230 remove_temp_file (mrg->xsrt, output_run->file_idx);
1233 /* Close and remove any remaining input runs. */
1234 for (i = 0; i < run_cnt; i++)
1236 close_temp_file (mrg->xsrt, runs[i].file_idx, runs[i].file);
1237 remove_temp_file (mrg->xsrt, runs[i].file_idx);
1243 /* Reads as many cases as possible into RUN's buffer.
1244 Reads nonzero unless a disk error occurs. */
1246 fill_run_buffer (struct merge_state *mrg, struct run *run)
1248 run->buffer_head = run->buffer_tail = run->buffer;
1249 while (run->unread_case_cnt > 0
1250 && run->buffer_tail < run->buffer + run->buffer_cap)
1252 if (!read_temp_file (mrg->xsrt, run->file_idx, run->file,
1253 (*run->buffer_tail)->data,
1254 mrg->xsrt->scp->case_size))
1257 run->unread_case_cnt--;
1264 static struct case_source *
1265 sort_sink_make_source (struct case_sink *sink)
1267 struct initial_run_state *irs = sink->aux;
1269 return create_case_source (&sort_source_class, default_dict,
1273 static const struct case_sink_class sort_sink_class =
1279 sort_sink_make_source,
1282 struct sort_source_aux
1284 struct sort_cases_pgm *scp;
1286 write_case_func *write_case;
1287 write_case_data wc_data;
1290 /* Passes C to the write_case function. */
1292 sort_source_read_helper (const struct ccase *src, void *aux_)
1294 struct sort_source_aux *aux = aux_;
1296 memcpy (aux->dst, src, aux->scp->case_size);
1297 return aux->write_case (aux->wc_data);
1300 /* Reads all the records from the source stream and passes them
1303 sort_source_read (struct case_source *source,
1305 write_case_func *write_case, write_case_data wc_data)
1307 struct sort_cases_pgm *scp = source->aux;
1308 struct sort_source_aux aux;
1312 aux.write_case = write_case;
1313 aux.wc_data = wc_data;
1315 read_sort_output (scp, sort_source_read_helper, &aux);
1318 static void read_internal_sort_output (struct internal_sort *isrt,
1319 read_sort_output_func *, void *aux);
1320 static void read_external_sort_output (struct external_sort *xsrt,
1321 read_sort_output_func *, void *aux);
1323 /* Reads all the records from the output stream and passes them to the
1324 function provided, which must have an interface identical to
1327 read_sort_output (struct sort_cases_pgm *scp,
1328 read_sort_output_func *output_func, void *aux)
1330 assert ((scp->isrt != NULL) + (scp->xsrt != NULL) <= 1);
1331 if (scp->isrt != NULL)
1332 read_internal_sort_output (scp->isrt, output_func, aux);
1333 else if (scp->xsrt != NULL)
1334 read_external_sort_output (scp->xsrt, output_func, aux);
1337 /* No results. Probably an external sort that failed. */
1342 read_internal_sort_output (struct internal_sort *isrt,
1343 read_sort_output_func *output_func,
1346 struct case_list **p;
1348 for (p = isrt->results; *p; p++)
1349 if (!output_func (&(*p)->c, aux))
1351 free (isrt->results);
1355 read_external_sort_output (struct external_sort *xsrt,
1356 read_sort_output_func *output_func, void *aux)
1363 assert (xsrt->run_cnt == 1);
1364 file_idx = xsrt->initial_runs[0].file_idx;
1366 file = open_temp_file (xsrt, file_idx, "rb");
1373 c = xmalloc (xsrt->scp->case_size);
1374 for (i = 0; i < xsrt->initial_runs[0].case_cnt; i++)
1376 if (!read_temp_file (xsrt, file_idx, file, c, xsrt->scp->case_size))
1382 if (!output_func (c, aux))
1389 sort_source_destroy (struct case_source *source)
1391 struct sort_cases_pgm *scp = source->aux;
1393 destroy_sort_cases_pgm (scp);
1396 const struct case_source_class sort_source_class =
1401 sort_source_destroy,