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);
144 /* Destroys a SORT CASES program. */
146 destroy_sort_cases_pgm (struct sort_cases_pgm *scp)
150 assert (scp->ref_cnt > 0);
151 if (--scp->ref_cnt > 0)
156 destroy_internal_sort (scp->isrt);
157 destroy_external_sort (scp->xsrt);
162 /* Sorts the active file based on the key variables specified in
163 global variables vars and var_cnt. The output is either to the
164 active file, if SEPARATE is zero, or to a separate file, if
165 SEPARATE is nonzero. In the latter case the output cases can be
166 read with a call to read_sort_output(). (In the former case the
167 output cases should be dealt with through the usual vfm interface.)
169 The caller is responsible for freeing vars[]. */
171 sort_cases (struct sort_cases_pgm *scp, int separate)
173 scp->case_size = sizeof (union value) * compaction_nval;
175 /* Not sure this is necessary but it's good to be safe. */
176 if (separate && case_source_is_class (vfm_source, &sort_source_class))
177 procedure (NULL, NULL, NULL, NULL);
179 /* SORT CASES cancels PROCESS IF. */
180 expr_free (process_if_expr);
181 process_if_expr = NULL;
183 /* Try an internal sort first. */
184 scp->isrt = do_internal_sort (scp, separate);
185 if (scp->isrt != NULL)
188 /* Fall back to an external sort. */
189 write_active_file_to_disk ();
190 scp->xsrt = do_external_sort (scp, separate);
191 if (scp->xsrt != NULL)
194 destroy_sort_cases_pgm (scp);
198 /* Results of an internal sort. */
201 struct case_list **results;
204 /* If a reasonable situation is set up, do an internal sort of the
205 data. Return success. */
206 static struct internal_sort *
207 do_internal_sort (struct sort_cases_pgm *scp, int separate)
209 struct internal_sort *isrt;
211 isrt = xmalloc (sizeof *isrt);
212 isrt->results = NULL;
214 if (!case_source_is_class (vfm_source, &disk_source_class))
216 if (!case_source_is_class (vfm_source, &memory_source_class))
217 procedure (NULL, NULL, NULL, NULL);
219 if (case_source_is_class (vfm_source, &memory_source_class))
221 struct case_list *case_list;
222 struct case_list **case_array;
226 case_cnt = vfm_source->class->count (vfm_source);
230 if (case_cnt > set_max_workspace / sizeof *case_array)
233 case_list = memory_source_get_cases (vfm_source);
234 case_array = malloc (sizeof *case_array * (case_cnt + 1));
235 if (case_array == NULL)
238 for (i = 0; case_list != NULL; i++)
240 case_array[i] = case_list;
241 case_list = case_list->next;
243 assert (i == case_cnt);
244 case_array[case_cnt] = NULL;
246 sort (case_array, case_cnt, sizeof *case_array,
247 compare_case_lists, scp);
251 memory_source_set_cases (vfm_source, case_array[0]);
252 for (i = 1; i <= case_cnt; i++)
253 case_array[i - 1]->next = case_array[i];
257 isrt->results = case_array;
268 /* Destroys an internal sort result. */
270 destroy_internal_sort (struct internal_sort *isrt)
274 free (isrt->results);
279 /* Compares the VAR_CNT variables in VARS[] between the
280 `case_list's at A and B, and returns a strcmp()-type
283 compare_case_lists (const void *a_, const void *b_, void *scp_)
285 struct sort_cases_pgm *scp = scp_;
286 struct case_list *const *pa = a_;
287 struct case_list *const *pb = b_;
288 struct case_list *a = *pa;
289 struct case_list *b = *pb;
291 return compare_record (a->c.data, b->c.data, scp);
296 /* Maximum order of merge. If you increase this, then you should
297 use a heap for comparing cases during merge. */
298 #define MAX_MERGE_ORDER 7
300 /* Minimum total number of records for buffers. */
301 #define MIN_BUFFER_TOTAL_SIZE_RECS 64
303 /* Minimum single input buffer size, in bytes and records. */
304 #define MIN_BUFFER_SIZE_BYTES 4096
305 #define MIN_BUFFER_SIZE_RECS 16
307 #if MIN_BUFFER_SIZE_RECS * 2 + 16 > MIN_BUFFER_TOTAL_SIZE_RECS
308 #error MIN_BUFFER_SIZE_RECS and MIN_BUFFER_TOTAL_SIZE_RECS do not make sense.
311 /* An initial run and its length. */
314 int file_idx; /* File index. */
315 size_t case_cnt; /* Number of cases. */
318 /* Sorts initial runs A and B in decending order by length. */
320 compare_initial_runs (const void *a_, const void *b_, void *aux UNUSED)
322 const struct initial_run *a = a_;
323 const struct initial_run *b = b_;
325 return a->case_cnt > b->case_cnt ? -1 : a->case_cnt <b->case_cnt;
328 /* Results of an external sort. */
331 struct sort_cases_pgm *scp; /* SORT CASES info. */
332 struct initial_run *initial_runs; /* Array of initial runs. */
333 size_t run_cnt, run_cap; /* Number of runs, allocated capacity. */
334 char *temp_dir; /* Temporary file directory name. */
335 char *temp_name; /* Name of a temporary file. */
336 int next_file_idx; /* Lowest unused file index. */
339 /* Prototypes for helper functions. */
340 static void sort_sink_write (struct case_sink *, const struct ccase *);
341 static int write_initial_runs (struct external_sort *, int separate);
342 static int init_external_sort (struct external_sort *);
343 static int merge (struct external_sort *);
344 static void rmdir_temp_dir (struct external_sort *);
345 static void remove_temp_file (struct external_sort *xsrt, int file_idx);
347 /* Performs an external sort of the active file according to the
348 specification in SCP. Forms initial runs using a heap as a
349 reservoir. Determines the optimum merge pattern via Huffman's
350 method (see Knuth vol. 3, 2nd edition, p. 365-366), and merges
351 according to that pattern. */
352 static struct external_sort *
353 do_external_sort (struct sort_cases_pgm *scp, int separate)
355 struct external_sort *xsrt;
358 xsrt = xmalloc (sizeof *xsrt);
360 if (!init_external_sort (xsrt))
362 if (!write_initial_runs (xsrt, separate))
372 /* Don't destroy anything because we'll need it for reading
378 destroy_external_sort (xsrt);
385 destroy_external_sort (struct external_sort *xsrt)
391 for (i = 0; i < xsrt->run_cnt; i++)
392 remove_temp_file (xsrt, xsrt->initial_runs[i].file_idx);
393 rmdir_temp_dir (xsrt);
394 free (xsrt->temp_dir);
395 free (xsrt->temp_name);
396 free (xsrt->initial_runs);
402 /* Creates and returns the name of a temporary directory. */
406 const char *parent_dir;
409 if (getenv ("TMPDIR") != NULL)
410 parent_dir = getenv ("TMPDIR");
412 parent_dir = P_tmpdir;
414 temp_dir = xmalloc (strlen (parent_dir) + 32);
415 sprintf (temp_dir, "%s%cpsppXXXXXX", parent_dir, DIR_SEPARATOR);
416 if (mkdtemp (temp_dir) == NULL)
418 msg (SE, _("%s: Creating temporary directory: %s."),
419 temp_dir, strerror (errno));
426 #else /* !HAVE_MKDTEMP */
427 /* Creates directory DIR. */
429 do_mkdir (const char *dir)
432 return mkdir (dir, S_IRWXU);
438 /* Creates and returns the name of a temporary directory. */
444 for (i = 0; i < 100; i++)
446 char temp_dir[L_tmpnam + 1];
447 if (tmpnam (temp_dir) == NULL)
449 msg (SE, _("Generating temporary directory name failed: %s."),
453 else if (do_mkdir (temp_dir) == 0)
454 return xstrdup (temp_dir);
457 msg (SE, _("Creating temporary directory failed: %s."), strerror (errno));
460 #endif /* !HAVE_MKDTEMP */
462 /* Sets up to open temporary files. */
464 init_external_sort (struct external_sort *xsrt)
467 xsrt->temp_dir = NULL;
468 xsrt->next_file_idx = 0;
473 xsrt->initial_runs = xmalloc (sizeof *xsrt->initial_runs * xsrt->run_cap);
475 /* Temporary directory. */
476 xsrt->temp_dir = make_temp_dir ();
477 xsrt->temp_name = NULL;
478 if (xsrt->temp_dir == NULL)
480 xsrt->temp_name = xmalloc (strlen (xsrt->temp_dir) + 64);
485 /* Returns nonzero if we should return an error even though the
486 operation succeeded. Useful for testing. */
488 simulate_error (void)
490 static int op_err_cnt = -1;
493 if (op_err_cnt == -1 || op_cnt++ < op_err_cnt)
502 /* Removes the directory created for temporary files, if one
505 rmdir_temp_dir (struct external_sort *xsrt)
507 if (xsrt->temp_dir != NULL && rmdir (xsrt->temp_dir) == -1)
509 msg (SE, _("%s: Error removing directory for temporary files: %s."),
510 xsrt->temp_dir, strerror (errno));
511 xsrt->temp_dir = NULL;
515 /* Returns the name of temporary file number FILE_IDX for XSRT.
516 The name is written into a static buffer, so be careful. */
518 get_temp_file_name (struct external_sort *xsrt, int file_idx)
520 assert (xsrt->temp_dir != NULL);
521 sprintf (xsrt->temp_name, "%s%c%04d",
522 xsrt->temp_dir, DIR_SEPARATOR, file_idx);
523 return xsrt->temp_name;
526 /* Opens temporary file numbered FILE_IDX for XSRT with mode MODE
527 and returns the FILE *. */
529 open_temp_file (struct external_sort *xsrt, int file_idx, const char *mode)
534 temp_file = get_temp_file_name (xsrt, file_idx);
536 file = fopen (temp_file, mode);
537 if (simulate_error () || file == NULL)
538 msg (SE, _("%s: Error opening temporary file for %s: %s."),
539 temp_file, mode[0] == 'r' ? "reading" : "writing",
545 /* Closes FILE, which is the temporary file numbered FILE_IDX
546 under XSRT. Returns nonzero only if successful. */
548 close_temp_file (struct external_sort *xsrt, int file_idx, FILE *file)
552 char *temp_file = get_temp_file_name (xsrt, file_idx);
553 if (simulate_error () || fclose (file) == EOF)
555 msg (SE, _("%s: Error closing temporary file: %s."),
556 temp_file, strerror (errno));
563 /* Delete temporary file numbered FILE_IDX for XSRT. */
565 remove_temp_file (struct external_sort *xsrt, int file_idx)
569 char *temp_file = get_temp_file_name (xsrt, file_idx);
570 if (simulate_error () || remove (temp_file) != 0)
571 msg (SE, _("%s: Error removing temporary file: %s."),
572 temp_file, strerror (errno));
576 /* Writes SIZE bytes from buffer DATA into FILE, which is
577 temporary file numbered FILE_IDX from XSRT. */
579 write_temp_file (struct external_sort *xsrt, int file_idx,
580 FILE *file, const void *data, size_t size)
582 if (!simulate_error () && fwrite (data, size, 1, file) == 1)
586 char *temp_file = get_temp_file_name (xsrt, file_idx);
587 msg (SE, _("%s: Error writing temporary file: %s."),
588 temp_file, strerror (errno));
593 /* Reads SIZE bytes into buffer DATA into FILE, which is
594 temporary file numbered FILE_IDX from XSRT. */
596 read_temp_file (struct external_sort *xsrt, int file_idx,
597 FILE *file, void *data, size_t size)
599 if (!simulate_error () && fread (data, size, 1, file) == 1)
603 char *temp_file = get_temp_file_name (xsrt, file_idx);
605 msg (SE, _("%s: Error reading temporary file: %s."),
606 temp_file, strerror (errno));
608 msg (SE, _("%s: Unexpected end of temporary file."),
614 /* Replacement selection. */
616 /* Pairs a record with a run number. */
619 int run; /* Run number of case. */
620 struct case_list *record; /* Case data. */
623 /* Represents a set of initial runs during an external sort. */
624 struct initial_run_state
626 struct external_sort *xsrt;
629 struct record_run *records; /* Records arranged as a heap. */
630 size_t record_cnt; /* Current number of records. */
631 size_t record_cap; /* Capacity for records. */
632 struct case_list *free_list;/* Cases not in heap. */
634 /* Run currently being output. */
635 int file_idx; /* Temporary file number. */
636 size_t case_cnt; /* Number of cases so far. */
637 size_t case_size; /* Number of bytes in a case. */
638 FILE *output_file; /* Output file. */
639 struct case_list *last_output;/* Record last output. */
641 int okay; /* Zero if an error has been encountered. */
644 static void destroy_initial_run_state (struct initial_run_state *irs);
645 static int allocate_cases (struct initial_run_state *);
646 static struct case_list *grab_case (struct initial_run_state *);
647 static void release_case (struct initial_run_state *, struct case_list *);
648 static void output_record (struct initial_run_state *irs);
649 static void start_run (struct initial_run_state *irs);
650 static void end_run (struct initial_run_state *irs);
651 static int compare_record_run (const struct record_run *,
652 const struct record_run *,
653 struct sort_cases_pgm *);
654 static int compare_record_run_minheap (const void *, const void *, void *);
656 /* Writes initial runs for XSRT, sending them to a separate file
657 if SEPARATE is nonzero. */
659 write_initial_runs (struct external_sort *xsrt, int separate)
661 struct initial_run_state *irs;
664 /* Allocate memory for cases. */
665 irs = xmalloc (sizeof *irs);
668 irs->record_cnt = irs->record_cap = 0;
669 irs->free_list = NULL;
670 irs->output_file = NULL;
671 irs->last_output = NULL;
674 irs->case_size = dict_get_case_size (default_dict);
676 if (!allocate_cases (irs))
679 /* Create case sink. */
682 if (vfm_sink != NULL && vfm_sink->class->destroy != NULL)
683 vfm_sink->class->destroy (vfm_sink);
684 vfm_sink = create_case_sink (&sort_sink_class, irs);
685 xsrt->scp->ref_cnt++;
688 /* Create initial runs. */
690 procedure (NULL, NULL, NULL, NULL);
691 while (irs->record_cnt > 0 && irs->okay)
698 destroy_initial_run_state (irs);
703 /* Add a single case to an initial run. */
705 sort_sink_write (struct case_sink *sink, const struct ccase *c)
707 struct initial_run_state *irs = sink->aux;
708 struct record_run *new_record_run;
713 /* Compose record_run for this run and add to heap. */
714 assert (irs->record_cnt < irs->record_cap);
715 new_record_run = irs->records + irs->record_cnt++;
716 new_record_run->record = grab_case (irs);
717 memcpy (new_record_run->record->c.data, c->data, irs->case_size);
718 new_record_run->run = irs->file_idx;
719 if (irs->last_output != NULL
720 && compare_record (c->data, irs->last_output->c.data,
722 new_record_run->run = irs->xsrt->next_file_idx;
723 push_heap (irs->records, irs->record_cnt, sizeof *irs->records,
724 compare_record_run_minheap, irs->xsrt->scp);
726 /* Output a record if the reservoir is full. */
727 if (irs->record_cnt == irs->record_cap && irs->okay)
731 /* Destroys the initial run state represented by IRS. */
733 destroy_initial_run_state (struct initial_run_state *irs)
735 struct case_list *iter, *next;
741 /* Release cases to free list. */
742 for (i = 0; i < irs->record_cnt; i++)
743 release_case (irs, irs->records[i].record);
744 if (irs->last_output != NULL)
745 release_case (irs, irs->last_output);
747 /* Free cases in free list. */
748 for (iter = irs->free_list; iter != NULL; iter = next)
757 if (irs->output_file != NULL)
758 close_temp_file (irs->xsrt, irs->file_idx, irs->output_file);
761 /* Allocates room for lots of cases as a buffer. */
763 allocate_cases (struct initial_run_state *irs)
765 size_t case_size; /* Size of one case, in bytes. */
766 int approx_case_cost; /* Approximate memory cost of one case in bytes. */
767 int max_cases; /* Maximum number of cases to allocate. */
770 /* Allocate as many cases as we can within the workspace
772 case_size = dict_get_case_size (default_dict);
773 approx_case_cost = (sizeof *irs->records
774 + sizeof *irs->free_list
776 + 4 * sizeof (void *));
777 max_cases = set_max_workspace / approx_case_cost;
778 irs->records = malloc (sizeof *irs->records * max_cases);
779 for (i = 0; i < max_cases; i++)
782 c = malloc (sizeof *c + case_size - sizeof (union value));
788 release_case (irs, c);
791 /* irs->records gets all but one of the allocated cases.
792 The extra is used for last_output. */
793 irs->record_cap = max_cases - 1;
795 /* Fail if we didn't allocate an acceptable number of cases. */
796 if (irs->records == NULL || max_cases < MIN_BUFFER_TOTAL_SIZE_RECS)
798 msg (SE, _("Out of memory. Could not allocate room for minimum of %d "
799 "cases of %d bytes each. (PSPP workspace is currently "
800 "restricted to a maximum of %d KB.)"),
801 MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, set_max_workspace / 1024);
807 /* Compares the VAR_CNT variables in VARS[] between the `value's at
808 A and B, and returns a strcmp()-type result. */
810 compare_record (const union value *a, const union value *b,
811 const struct sort_cases_pgm *scp)
818 for (i = 0; i < scp->var_cnt; i++)
820 struct variable *v = scp->vars[i];
824 if (v->type == NUMERIC)
829 result = af < bf ? -1 : af > bf;
832 result = memcmp (a[fv].s, b[fv].s, v->width);
836 if (scp->dirs[i] == SRT_DESCEND)
845 /* Compares record-run tuples A and B on run number first, then
846 on the current record according to SCP. */
848 compare_record_run (const struct record_run *a,
849 const struct record_run *b,
850 struct sort_cases_pgm *scp)
852 if (a->run != b->run)
853 return a->run > b->run ? 1 : -1;
855 return compare_record (a->record->c.data, b->record->c.data, scp);
858 /* Compares record-run tuples A and B on run number first, then
859 on the current record according to SCP, but in descending
862 compare_record_run_minheap (const void *a, const void *b, void *scp)
864 return -compare_record_run (a, b, scp);
867 /* Begins a new initial run, specifically its output file. */
869 start_run (struct initial_run_state *irs)
871 irs->file_idx = irs->xsrt->next_file_idx++;
873 irs->output_file = open_temp_file (irs->xsrt, irs->file_idx, "wb");
874 if (irs->output_file == NULL)
876 if (irs->last_output != NULL)
878 release_case (irs, irs->last_output);
879 irs->last_output = NULL;
883 /* Ends the current initial run. */
885 end_run (struct initial_run_state *irs)
887 struct external_sort *xsrt = irs->xsrt;
889 /* Record initial run. */
890 if (xsrt->run_cnt >= xsrt->run_cap)
894 = xrealloc (xsrt->initial_runs,
895 sizeof *xsrt->initial_runs * xsrt->run_cap);
897 xsrt->initial_runs[xsrt->run_cnt].file_idx = irs->file_idx;
898 xsrt->initial_runs[xsrt->run_cnt].case_cnt = irs->case_cnt;
901 /* Close file handle. */
902 if (irs->output_file != NULL
903 && !close_temp_file (irs->xsrt, irs->file_idx, irs->output_file))
905 irs->output_file = NULL;
908 /* Writes a record to the current initial run. */
910 output_record (struct initial_run_state *irs)
912 struct record_run *record_run;
913 struct ccase *out_case;
915 /* Extract minimum case from heap. */
916 assert (irs->record_cnt > 0);
917 pop_heap (irs->records, irs->record_cnt--, sizeof *irs->records,
918 compare_record_run_minheap, irs->xsrt->scp);
919 record_run = irs->records + irs->record_cnt;
921 /* Bail if an error has occurred. */
925 /* Obtain case data to write to disk. */
926 out_case = &record_run->record->c;
927 if (compaction_necessary)
929 compact_case (compaction_case, out_case);
930 out_case = compaction_case;
933 /* Start new run if necessary. */
934 assert (record_run->run == irs->file_idx
935 || record_run->run == irs->xsrt->next_file_idx);
936 if (record_run->run != irs->file_idx)
941 assert (record_run->run == irs->file_idx);
945 if (irs->output_file != NULL
946 && !write_temp_file (irs->xsrt, irs->file_idx, irs->output_file,
948 sizeof *out_case->data * compaction_nval))
951 /* This record becomes last_output. */
952 if (irs->last_output != NULL)
953 release_case (irs, irs->last_output);
954 irs->last_output = record_run->record;
957 /* Gets a case from the free list in IRS. It is an error to call
958 this function if the free list is empty. */
959 static struct case_list *
960 grab_case (struct initial_run_state *irs)
964 assert (irs != NULL);
965 assert (irs->free_list != NULL);
968 irs->free_list = c->next;
972 /* Returns C to the free list in IRS. */
974 release_case (struct initial_run_state *irs, struct case_list *c)
976 assert (irs != NULL);
979 c->next = irs->free_list;
985 /* State of merging initial runs. */
988 struct external_sort *xsrt; /* External sort state. */
989 struct ccase **cases; /* Buffers. */
990 size_t case_cnt; /* Number of buffers. */
994 static int merge_once (struct merge_state *,
995 const struct initial_run[], size_t,
996 struct initial_run *);
997 static int fill_run_buffer (struct merge_state *, struct run *);
998 static int mod (int, int);
1000 /* Performs a series of P-way merges of initial runs
1003 merge (struct external_sort *xsrt)
1005 struct merge_state mrg; /* State of merge. */
1006 size_t case_size; /* Size of one case, in bytes. */
1007 size_t approx_case_cost; /* Approximate memory cost of one case. */
1008 int max_order; /* Maximum order of merge. */
1009 size_t dummy_run_cnt; /* Number of dummy runs to insert. */
1015 /* Allocate as many cases as possible into cases. */
1016 case_size = dict_get_case_size (default_dict);
1017 approx_case_cost = sizeof *mrg.cases + case_size + 4 * sizeof (void *);
1018 mrg.case_cnt = set_max_workspace / approx_case_cost;
1019 mrg.cases = malloc (sizeof *mrg.cases * mrg.case_cnt);
1020 if (mrg.cases == NULL)
1022 for (i = 0; i < mrg.case_cnt; i++)
1024 mrg.cases[i] = malloc (case_size);
1025 if (mrg.cases[i] == NULL)
1031 if (mrg.case_cnt < MIN_BUFFER_TOTAL_SIZE_RECS)
1033 msg (SE, _("Out of memory. Could not allocate room for minimum of %d "
1034 "cases of %d bytes each. (PSPP workspace is currently "
1035 "restricted to a maximum of %d KB.)"),
1036 MIN_BUFFER_TOTAL_SIZE_RECS, approx_case_cost, set_max_workspace / 1024);
1040 /* Determine maximum order of merge. */
1041 max_order = MAX_MERGE_ORDER;
1042 if (mrg.case_cnt / max_order < MIN_BUFFER_SIZE_RECS)
1043 max_order = mrg.case_cnt / MIN_BUFFER_SIZE_RECS;
1044 else if (mrg.case_cnt / max_order * case_size < MIN_BUFFER_SIZE_BYTES)
1045 max_order = mrg.case_cnt / (MIN_BUFFER_SIZE_BYTES / case_size);
1048 if (max_order > xsrt->run_cnt)
1049 max_order = xsrt->run_cnt;
1051 /* Repeatedly merge the P shortest existing runs until only one run
1053 make_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
1054 compare_initial_runs, NULL);
1055 dummy_run_cnt = mod (1 - (int) xsrt->run_cnt, max_order - 1);
1056 assert (max_order == 1
1057 || (xsrt->run_cnt + dummy_run_cnt) % (max_order - 1) == 1);
1058 while (xsrt->run_cnt > 1)
1060 struct initial_run output_run;
1064 /* Choose order of merge (max_order after first merge). */
1065 order = max_order - dummy_run_cnt;
1068 /* Choose runs to merge. */
1069 assert (xsrt->run_cnt >= order);
1070 for (i = 0; i < order; i++)
1071 pop_heap (xsrt->initial_runs, xsrt->run_cnt--,
1072 sizeof *xsrt->initial_runs,
1073 compare_initial_runs, NULL);
1076 if (!merge_once (&mrg, xsrt->initial_runs + xsrt->run_cnt, order,
1080 /* Add output run to heap. */
1081 xsrt->initial_runs[xsrt->run_cnt++] = output_run;
1082 push_heap (xsrt->initial_runs, xsrt->run_cnt, sizeof *xsrt->initial_runs,
1083 compare_initial_runs, NULL);
1086 /* Exactly one run is left, which contains the entire sorted
1087 file. We could use it to find a total case count. */
1088 assert (xsrt->run_cnt == 1);
1093 for (i = 0; i < mrg.case_cnt; i++)
1094 free (mrg.cases[i]);
1100 /* Modulo function as defined by Knuth. */
1108 else if (x > 0 && y > 0)
1110 else if (x < 0 && y > 0)
1111 return y - (-x) % y;
1116 /* A run of data for use in merging. */
1119 FILE *file; /* File that contains run. */
1120 int file_idx; /* Index of file that contains run. */
1121 struct ccase **buffer; /* Case buffer. */
1122 struct ccase **buffer_head; /* First unconsumed case in buffer. */
1123 struct ccase **buffer_tail; /* One past last unconsumed case in buffer. */
1124 size_t buffer_cap; /* Number of cases buffer can hold. */
1125 size_t unread_case_cnt; /* Number of cases not yet read. */
1128 /* Merges the RUN_CNT initial runs specified in INPUT_RUNS into a
1129 new run. Returns nonzero only if successful. Adds an entry
1130 to MRG->xsrt->runs for the output file if and only if the
1131 output file is actually created. Always deletes all the input
1134 merge_once (struct merge_state *mrg,
1135 const struct initial_run input_runs[],
1137 struct initial_run *output_run)
1139 struct run runs[MAX_MERGE_ORDER];
1140 FILE *output_file = NULL;
1145 /* Initialize runs[]. */
1146 for (i = 0; i < run_cnt; i++)
1148 runs[i].file = NULL;
1149 runs[i].file_idx = input_runs[i].file_idx;
1150 runs[i].buffer = mrg->cases + mrg->case_cnt / run_cnt * i;
1151 runs[i].buffer_head = runs[i].buffer;
1152 runs[i].buffer_tail = runs[i].buffer;
1153 runs[i].buffer_cap = mrg->case_cnt / run_cnt;
1154 runs[i].unread_case_cnt = input_runs[i].case_cnt;
1157 /* Open input files. */
1158 for (i = 0; i < run_cnt; i++)
1160 runs[i].file = open_temp_file (mrg->xsrt, runs[i].file_idx, "rb");
1161 if (runs[i].file == NULL)
1165 /* Create output file and count cases to be output. */
1166 output_run->file_idx = mrg->xsrt->next_file_idx++;
1167 output_run->case_cnt = 0;
1168 for (i = 0; i < run_cnt; i++)
1169 output_run->case_cnt += input_runs[i].case_cnt;
1170 output_file = open_temp_file (mrg->xsrt, output_run->file_idx, "wb");
1171 if (output_file == NULL)
1174 /* Prime buffers. */
1175 for (i = 0; i < run_cnt; i++)
1176 if (!fill_run_buffer (mrg, runs + i))
1180 case_size = dict_get_case_size (default_dict);
1183 struct run *min_run;
1187 for (i = 1; i < run_cnt; i++)
1188 if (compare_record ((*runs[i].buffer_head)->data,
1189 (*min_run->buffer_head)->data,
1190 mrg->xsrt->scp) < 0)
1193 /* Write minimum to output file. */
1194 if (!write_temp_file (mrg->xsrt, min_run->file_idx, output_file,
1195 (*min_run->buffer_head)->data, case_size))
1198 /* Remove case from buffer. */
1199 if (++min_run->buffer_head >= min_run->buffer_tail)
1201 /* Buffer is empty. Fill from file. */
1202 if (!fill_run_buffer (mrg, min_run))
1205 /* If buffer is still empty, delete its run. */
1206 if (min_run->buffer_head >= min_run->buffer_tail)
1208 close_temp_file (mrg->xsrt, min_run->file_idx, min_run->file);
1209 remove_temp_file (mrg->xsrt, min_run->file_idx);
1210 *min_run = runs[--run_cnt];
1212 /* We could donate the now-unused buffer space to
1218 /* Close output file. */
1219 close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1224 /* Close and remove output file. */
1225 if (output_file != NULL)
1227 close_temp_file (mrg->xsrt, output_run->file_idx, output_file);
1228 remove_temp_file (mrg->xsrt, output_run->file_idx);
1231 /* Close and remove any remaining input runs. */
1232 for (i = 0; i < run_cnt; i++)
1234 close_temp_file (mrg->xsrt, runs[i].file_idx, runs[i].file);
1235 remove_temp_file (mrg->xsrt, runs[i].file_idx);
1241 /* Reads as many cases as possible into RUN's buffer.
1242 Reads nonzero unless a disk error occurs. */
1244 fill_run_buffer (struct merge_state *mrg, struct run *run)
1246 run->buffer_head = run->buffer_tail = run->buffer;
1247 while (run->unread_case_cnt > 0
1248 && run->buffer_tail < run->buffer + run->buffer_cap)
1250 if (!read_temp_file (mrg->xsrt, run->file_idx, run->file,
1251 (*run->buffer_tail)->data,
1252 dict_get_case_size (default_dict)))
1255 run->unread_case_cnt--;
1262 static struct case_source *
1263 sort_sink_make_source (struct case_sink *sink)
1265 struct initial_run_state *irs = sink->aux;
1267 return create_case_source (&sort_source_class, default_dict,
1271 const struct case_sink_class sort_sink_class =
1277 sort_sink_make_source,
1280 struct sort_source_aux
1282 struct sort_cases_pgm *scp;
1284 write_case_func *write_case;
1285 write_case_data wc_data;
1288 /* Passes C to the write_case function. */
1290 sort_source_read_helper (const struct ccase *src, void *aux_)
1292 struct sort_source_aux *aux = aux_;
1294 memcpy (aux->dst, src, aux->scp->case_size);
1295 return aux->write_case (aux->wc_data);
1298 /* Reads all the records from the source stream and passes them
1301 sort_source_read (struct case_source *source,
1303 write_case_func *write_case, write_case_data wc_data)
1305 struct sort_cases_pgm *scp = source->aux;
1306 struct sort_source_aux aux;
1310 aux.write_case = write_case;
1311 aux.wc_data = wc_data;
1313 read_sort_output (scp, sort_source_read_helper, &aux);
1316 static void read_internal_sort_output (struct internal_sort *isrt,
1317 read_sort_output_func *, void *aux);
1318 static void read_external_sort_output (struct external_sort *xsrt,
1319 read_sort_output_func *, void *aux);
1321 /* Reads all the records from the output stream and passes them to the
1322 function provided, which must have an interface identical to
1325 read_sort_output (struct sort_cases_pgm *scp,
1326 read_sort_output_func *output_func, void *aux)
1328 assert ((scp->isrt != NULL) + (scp->xsrt != NULL) <= 1);
1329 if (scp->isrt != NULL)
1330 read_internal_sort_output (scp->isrt, output_func, aux);
1331 else if (scp->xsrt != NULL)
1332 read_external_sort_output (scp->xsrt, output_func, aux);
1335 /* No results. Probably an external sort that failed. */
1340 read_internal_sort_output (struct internal_sort *isrt,
1341 read_sort_output_func *output_func,
1344 struct case_list **p;
1346 for (p = isrt->results; *p; p++)
1347 if (!output_func (&(*p)->c, aux))
1349 free (isrt->results);
1353 read_external_sort_output (struct external_sort *xsrt,
1354 read_sort_output_func *output_func, void *aux)
1361 assert (xsrt->run_cnt == 1);
1362 file_idx = xsrt->initial_runs[0].file_idx;
1364 file = open_temp_file (xsrt, file_idx, "rb");
1371 c = xmalloc (xsrt->scp->case_size);
1372 for (i = 0; i < xsrt->initial_runs[0].case_cnt; i++)
1374 if (!read_temp_file (xsrt, file_idx, file, c, xsrt->scp->case_size))
1380 if (!output_func (c, aux))
1387 sort_source_destroy (struct case_source *source)
1389 struct sort_cases_pgm *scp = source->aux;
1391 destroy_sort_cases_pgm (scp);
1394 const struct case_source_class sort_source_class =
1399 sort_source_destroy,