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