#pragma prototyped
static const char usage[] =
"[-n?\n@(#)$Id: sort (AT&T Research) 2010-08-11 $\n]"
USAGE_LICENSE
"[+NAME?sort - sort and/or merge files]"
"[+DESCRIPTION?\bsort\b sorts lines of all the \afiles\a together and "
"writes the result on the standard output. The file name \b-\b means the "
"standard input. If no files are named, the standard input is sorted.]"
"[+?The default sort key is an entire line. Default ordering is "
"lexicographic by bytes in machine collating sequence. The ordering is "
"affected globally by the locale and/or the following options, one or "
"more of which may appear. See \brecsort\b(3) for details.]"
"[+?For backwards compatibility the \b-o\b option is allowed in any file "
"operand position when neither the \b-c\b nor the \b--\b options are "
"specified.]"
"[k:key?Restrict the sort key to a string beginning at \apos1\a and "
"ending at \apos2\a. \apos1\a and \apos2\a each have the form \am.n\a, "
"counting from 1, optionally followed by one or more of the flags "
"\bCMbdfginprZ\b; \bm\b counts fields from the beginning of the line and "
"\bn\b counts characters from the beginning of the field. If any flags "
"are present they override all the global ordering options for this key. "
"If \a.n\a is missing from \apos1\a, it is taken to be 1; if missing "
"from \apos2\a, it is taken to be the end of the field. If \apos2\a is "
"missing, it is taken to be end of line. The second form specifies a "
"fixed record length \areclen\a, and the last form specifies a fixed "
"field at byte position \aposition\a (counting from 1) of \alength\a "
"bytes. The obsolescent \breclen:fieldlen:offset\b (byte offset from 0) "
"is also accepted.]:[pos1[,pos2]]|.reclen|.position.length]]]]]"
"[K:oldkey?Specified in pairs: \b-K\b \apos1\a \b-K\b \apos2\a, where "
"positions count from 0.]# [pos]"
"[R:record|recfmt?Sets the record format to \aformat\a; newlines will be "
"treated as normal characters. The formats are:]:[format]"
"{"
"[+d[\aterminator\a]]?Variable length with record \aterminator\a "
"character, \b\\n\b by default.]"
"[+[f]]\areclen\a?Fixed record length \areclen\a.]"
"[+v[op...]]?Variable length. \bh4o0z2bi\b (4 byte IBM V format "
"descriptor) if \aop\a are omitted. \aop\a may be a combination "
"of:]"
"{"
"[+h\an\a?Header size is \an\a bytes (default 4).]"
"[+o\an\a?Size offset in header is \an\a bytes (default "
"0).]"
"[+z\an\a?Size length is \an\a bytes (default "
"min(\bh\b-\bo\b,2)).]"
"[+b?Size is big-endian (default).]"
"[+l?Size is little-endian (default \bb\b).]"
"[+i?Record length includes header (default).]"
"[+n?Record length does not include header (default "
"\bi\b).]"
"}"
"[+%?If the record format is not otherwise specified, and the "
"any input file name, from left to right, ends with "
"\b%\b\aformat\a or \b%\b\aformat\a\b.\b* then the record format "
"is set to \aformat\a. In addition, the \b-o\b path, if "
"specified and if it does not contain \b%\b and if it names a "
"regular file, is renamed to contain the input \b%\b\aformat\a.]"
"[+-?The first block of the first input file is sampled to check "
"for \bv\b variable length and \bf\b fixed length format "
"records. Not all formats are detected. \bsort\b exits with an "
"error diagnostic if the record format cannot be determined from "
"the sample.]"
"}"
"[b:ignorespace|ignore-leading-blanks?Ignore leading white space (spaces "
"and tabs) in field comparisons.]"
"[d:dictionary?`Phone directory' order: only letters, digits and white "
"space are significant in string comparisons.]"
"[E:codeset|convert?The field data codeset is \acodeset\a or the field "
"data must be converted from the \afrom\a codeset to the \ato\a codeset. "
"The codesets are:]:[codeset|from::to]"
"{\fcodesets\f}"
"[f:fold|ignorecase?Fold lower case letters onto upper case.]"
"[h:scaled|human-readable?Compare numbers scaled with IEEE 1541-2002 "
"suffixes.]"
"[i:ignorecontrol?Ignore characters outside the ASCII range 040-0176 in "
"string comparisons.]"
"[J:shuffle|jumble?Do a random shuffle of the sort keys. \aseed\a "
"specifies a pseudo random number generator seed. A \aseed\a of 0 "
"generates a seed based on time and pid.]#[seed]"
"[n:numeric?An initial numeric string, consisting of optional white "
"space, optional sign, and a nonempty string of digits with optional "
"decimal point, is sorted by value.]"
"[g:floating?Numeric, like \b-n\b, with \be\b-style exponents allowed.]"
"[p:bcd|packed-decimal?Compare packed decimal (bcd) numbers with "
"trailing sign.]"
"[M:months?Compare as month names. The first three characters after "
"optional white space are folded to lower case and compared. Invalid "
"fields compare low to \bjan\b.]"
"[r:reverse|invert?Reverse the sense of comparisons.]"
"[t:tabs?`Tab character' separating fields is \achar\a.]:[tab-char]"
"[c:check?Check that the single input file is sorted according to the "
"ordering rules; give no output on the standard output. If the input "
"is out of sort then write one diagnostic line on the standard error "
"and exit with code \b1\b.]"
"[C:silent-check?Like \b--check\b except no diagnostic is written.]"
"[j:processes|nproc|jobs?Use up to \ajobs\a separate processes to sort "
"the input. The current implementation still uses one process for the "
"final merge phase; improvements are planned.]#[processes]"
"[m:merge?Merge; the input files are already sorted.]"
"[u:unique?Unique. Keep only the first of multiple records that compare "
"equal on all keys. Implies \b-s\b.]"
"[s:stable?Stable sort. When all keys compare equal, preserve input "
"order. The default is \b--nostable\b (\aunstable\a sort): when all "
"keys compare equal, break the tie by using the entire record, ignoring "
"all but the \b-r\b option.]"
"[o:output?Place output in the designated \afile\a instead of on the "
"standard output. This file may be the same as one of the inputs. The "
"\afile\a \b-\b names the standard output. The option may appear among "
"the file arguments, except after \b--\b.]:[output]"
"[l:library?Load the external sort discipline \alibrary\a with optional "
"comma separated \aname=value\a arguments. Libraries are loaded, in left "
"to right order, after the sort method has been "
"initialized.]:[library[,name=value...]]]"
"[T:tempdir?Put temporary files in \atempdir\a.]:[tempdir:=/usr/tmp]"
"[L:list?List the available sort methods. See the \b-x\b option.]"
"[x:method?Specify the sort method to apply:]:[method:=rasp]"
"{\fmethods\f} [v:verbose?Trace the sort progress on the standard "
"error.]"
"[P:plugins?List plugin information for each \afile\a operand in "
"--\astyle\a on the standard error. If no \afile\a operands are "
"specified then the first instance of each \bsort\b plugin installed on "
"\b$PATH\b or a sibling dir on \b$PATH\b is listed. The special "
"\astyle\a \blist\b lists a line on the standard output for each plugin "
"with the name, a tab character, and plugin specific command line "
"options parameterized by \b${style}\b (suitable for \beval\b'ing in "
"\bsh\b(1).)]:[style:=list|man|html|nroff|usage]"
"[Z:zd|zoned-decimal?Compare zoned decimal (ZD) numbers with embedded "
"trailing sign.]"
"[z:size|zip?Suggest using the specified number of bytes of internal "
"store to tune performance. Power of 2 and power of 10 size suffixes are "
"accepted. Type is a single character and may be one of:]:[type[size]]]"
"{"
"[+a?Buffer alignment.]"
"[+b?Input reserve buffer size.]"
"[+c?Input chunk size; sort chunks of this size and disable "
"merge.]"
"[+i?Input buffer size.]"
"[+m?Maximum number of intermediate merge files.]"
"[+p?Input sort size; sort chunks of this size before merge.]"
"[+o?Output buffer size.]"
"[+r?Maximum record size.]"
"[+I?Decompress the input if it is compressed.]"
"[+O?\bgzip\b(1) compress the output.]"
"}"
"[X:test?Enables implementation defined test code. Some or all of these "
"may be disabled.]:[test]"
"{"
"[+dump?List detailed information on the option settings.]"
"[+io?List io file paths.]"
"[+keys?List the canonical key for each record.]"
"[+read?Force input file read by disabling memory mapping.]"
"[+show?Show setup information and exit before sorting.]"
"[+test?Immediatly exit with status 0; used to verify this "
"implementation]"
"}"
"[D:debug?Sets the debug trace level. Higher levels produce more "
"output.]# [level]"
"[S|y?Equivalent to \b-zp\b\asize\a; if \asize\a has no suffix then \bki\b "
"is assumed.]:[size]"
"\n"
"\n[ file ... ]\n"
"\n"
"[+?+\apos1\a -\apos2\a is the classical alternative to \b-k\b, with "
"counting from 0 instead of 1, and pos2 designating next-after-last "
"instead of last character of the key. A missing character count in "
"\apos2\a means 0, which in turn excludes any \b-t\b tab character from "
"the end of the key. Thus +1 -1.3 is the same as \b-k\b 2,2.3 and +1r -3 "
"is the same as \b-k\b 2r,3.]"
"[+?Under option \b-t\b\ax\a fields are strings separated by \ax\a; "
"otherwise fields are non-empty strings separated by white space. White "
"space before a field is part of the field, except under option \b-b\b. "
"A \bb\b flag may be attached independently to \apos1\a and \apos2\a.]"
"[+?When there are multiple sort keys, later keys are compared only "
"after all earlier keys compare equal. Except under option \b-s\b, lines "
"with all keys equal are ordered with all bytes significant. \b-S\b "
"turns off \b-s\b, the last occurrence, left-to-right, takes affect.]"
"[+?Sorting is done by a method determined by the \b-x\b option. \b-L\b "
"lists the available methods. rasp (radix+splay-tree) is the default and "
"current all-around best.]"
"[+?Single-letter options may be combined into a single string, such as "
"\b-cnrt:\b. The option combination \b-di\b and the combination of "
"\b-n\b with any of \b-diM\b are improper. Posix argument conventions "
"are supported.]"
"[+?Options \b-b\b, \b-c\b, \b-d\b, \b-f\b, \b-i\b, \b-k\b, \b-m\b, "
"\b-n\b, \b-o\b, \b-r\b, \b-t\b, and \b-u\b are in the Posix and/or "
"X/Open standards.]"
"[+DIAGNOSTICS?\asort\a comments and exits with non-zero status for "
"various trouble conditions and for disorder discovered under option "
"\b-c\b.]"
"[+SEE ALSO?\bcomm\b(1), \bjoin\b(1), \buniq\b(1), \brecsort\b(3)]"
"[+CAVEATS?The never-documented default \apos1\a=0 for cases such as "
"\bsort -1\b has been abolished. An input file overwritten by \b-o\b is "
"not replaced until the entire output file is generated in the same "
"directory as the input, at which point the input is renamed.]"
;
#include <sfio_t.h>
#include <ast.h>
#include <error.h>
#include <ctype.h>
#include <fs3d.h>
#include <ls.h>
#include <option.h>
#include <recsort.h>
#include <recfmt.h>
#include <sfdcgzip.h>
#include <vmalloc.h>
#include <wait.h>
#include <iconv.h>
#include <dlldefs.h>
#define INMIN (1024)
#define INBRK (64*INMIN)
#define INMAX (1024*INBRK)
#define INREC (16*INMIN)
#define TEST_dump 0x80000000
#define TEST_io 0x40000000
#define TEST_keys 0x20000000
#define TEST_read 0x10000000
#define TEST_show 0x08000000
#define TEST_reserve 0x04000000
#define pathstdin(s) (!(s)||streq(s,"-")||streq(s,"/dev/stdin")||streq(s,"/dev/fd/0"))
#define pathstdout(s) (!(s)||streq(s,"-")||streq(s,"/dev/stdout")||streq(s,"/dev/fd/1"))
typedef struct Part_s
{
Sfdisc_t disc;
off_t offset;
off_t size;
off_t remain;
} Part_t;
typedef struct Job_s
{
off_t offset;
off_t size;
size_t chunk;
int intermediates;
} Job_t;
typedef struct Sort_s
{
Rskeydisc_t disc;
Rs_t* rec;
Rskey_t* key;
Rsdefkey_f defkeyf;
Sfio_t* tp;
Sfio_t* op;
Job_t* jobs;
char* overwrite;
char* buf;
Sfio_t* opened;
size_t cur;
size_t hit;
size_t end;
size_t bufsize;
off_t total;
unsigned long test;
int child;
int chunk;
int hadstdin;
int map;
int mfiles;
int nfiles;
int xfiles;
int preserve;
int single;
int verbose;
int zip;
Sfio_t* files[OPEN_MAX > 68 ? 64 : (OPEN_MAX-4)];
} Sort_t;
static int
optinfo(Opt_t* op, Sfio_t* sp, const char* s, Optdisc_t* dp)
{
register iconv_list_t* ic;
register int n;
if (streq(s, "codesets"))
{
n = 0;
for (ic = iconv_list(NiL); ic; ic = iconv_list(ic))
if (ic->ccode >= 0)
n += sfprintf(sp, "[%c:%s?%s]", ic->match[ic->match[0] == '('], ic->name, ic->desc);
return n;
}
else if (streq(s, "methods"))
return rskeylist(NiL, sp, 1);
return 0;
}
static int
verify(Rs_t* rs, int op, Void_t* data, Void_t* arg, Rsdisc_t* disc)
{
if (op == RS_VERIFY)
error(3, "disorder at record %lld", (Sflong_t)((Rsobj_t*)data)->order);
return 0;
}
static int
verify_silent(Rs_t* rs, int op, Void_t* data, Void_t* arg, Rsdisc_t* disc)
{
if (op == RS_VERIFY)
exit(1);
return 0;
}
static Sfio_t*
fileopen(register Sort_t* sp, const char* path)
{
Sfio_t* fp;
if (fp = sp->opened)
sp->opened = 0;
else
{
if (pathstdin(path))
{
if (sp->hadstdin)
error(3, "%s: can only read once", path);
sp->hadstdin = 1;
fp = sfstdin;
}
else if (!(fp = sfopen(NiL, path, "r")))
error(ERROR_SYSTEM|3, "%s: cannot open", path);
if (rsfileread(sp->rec, fp, path))
{
if (fp != sfstdin)
sfclose(fp);
return 0;
}
sfset(fp, SF_SHARE, 0);
if (sp->zip & SF_READ)
sfdcgzip(fp, 0);
}
return fp;
}
static void
noexit(int code)
{
}
static void
showlib(Sort_t* sp, Rskey_t* kp, const char* name, const char* style)
{
char* args;
char buf[128];
if (style)
sfsprintf(args = buf, sizeof(buf), "%s,%s", name, style);
else
args = (char*)name;
if (!rslib(sp->rec, kp, args, RS_IGNORE) && !style)
sfprintf(sfstdout, "%s\t--library=%s,${style}\n", name, name);
}
static int
showplugins(Sort_t* sp, Rskey_t* kp, const char* style)
{
Dllscan_t* dls;
Dllent_t* dle;
char* name;
void (*oexit)(int);
if (streq(style, "list"))
style = 0;
else
{
oexit = error_info.exit;
error_info.exit = noexit;
}
if (*kp->input)
while (name = *kp->input++)
showlib(sp, kp, name, style);
else if (dls = dllsopen("sort", NiL, NiL))
{
while (dle = dllsread(dls))
showlib(sp, kp, dle->name, style);
dllsclose(dls);
}
if (style)
error_info.exit = oexit;
return 0;
}
struct Lib_s; typedef struct Lib_s Lib_t;
struct Lib_s
{
Lib_t* next;
char* name;
};
static int
parse(register Sort_t* sp, char** argv)
{
register Rskey_t* key = sp->key;
register int n;
register char* s;
char* e;
char* p;
char** a;
char** v;
size_t z;
int i;
int map;
char* plugins = 0;
Lib_t* firstlib = 0;
Lib_t* lastlib = 0;
Lib_t* lib;
Recfmt_t r;
int obsolescent = 1;
char opt[64];
Optdisc_t optdisc;
struct stat st;
optinit(&optdisc, optinfo);
for (;;)
{
switch (optget(argv, usage))
{
case 0:
break;
case 'c':
case 'C':
obsolescent = 0;
key->meth = Rsverify;
key->disc->events = RS_VERIFY;
key->disc->eventf = opt_info.option[1] == 'C' ? verify_silent : verify;
continue;
case 'E':
case 'J':
sfsprintf(opt, sizeof(opt), "%c%s", opt_info.option[1], opt_info.arg);
if (rskeyopt(key, opt, 1))
return 0;
continue;
case 'j':
key->nproc = opt_info.num;
continue;
case 'k':
if (rskey(key, opt_info.arg, 0))
return -1;
continue;
case 'l':
if (!(lib = newof(0, Lib_t, 1, 0)))
error(ERROR_SYSTEM|3, "out of space");
lib->name = opt_info.arg;
if (lastlib)
lastlib = lastlib->next = lib;
else
firstlib = lastlib = lib;
continue;
case 'm':
key->merge = !!opt_info.num;
continue;
case 'o':
key->output = opt_info.arg;
continue;
case 's':
if (opt_info.num)
key->type &= ~RS_DATA;
else
key->type |= RS_DATA;
continue;
case 't':
if (key->tab[0])
error(1, "%s: %s conflicts with %s", opt_info.option, *opt_info.arg, key->tab);
if ((n = mbsize(opt_info.arg)) < 1)
{
error(1, "%s: %s: invalid tab character", opt_info.option, opt_info.arg);
n = 0;
}
if (*(opt_info.arg + n) || n >= sizeof(key->tab))
error(1, "%s: %s: single character expected", opt_info.option, opt_info.arg);
memcpy(key->tab, opt_info.arg, n);
key->tab[n] = 0;
continue;
case 'u':
key->type &= ~RS_DATA;
key->type |= RS_UNIQ;
continue;
case 'v':
key->verbose = !!opt_info.num;
sp->verbose = key->verbose || (key->test & TEST_show);
continue;
case 'x':
if (!(key->meth = rskeymeth(key, opt_info.arg)))
error(2, "%s: unknown method", opt_info.arg);
continue;
case 'z':
if (isalpha(n = *(s = opt_info.arg)))
s++;
else
n = 'r';
size:
z = strton(s, &e, NiL, 1);
if (*e == '%')
{
error(2, "%s %c%s: %% not supported -- do you really want that much memory?", opt_info.option, n, s);
return -1;
}
if (*e || z < ((n == 'm' || n == 'o' || n == 'r' || isupper(n)) ? 0 : 512))
{
error(2, "%s %c%s: invalid size", opt_info.option, n, s);
return -1;
}
switch (n)
{
case 'a':
key->alignsize = z;
break;
case 'b':
sp->bufsize = z;
break;
case 'c':
sp->chunk = 1;
key->alignsize = key->insize = z;
break;
case 'i':
key->insize = z;
break;
case 'm':
if (z <= 0 || z > elementsof(sp->files))
z = elementsof(sp->files);
sp->xfiles = z;
break;
case 'p':
key->procsize = z;
break;
case 'o':
key->outsize = z;
break;
case 'r':
key->recsize = z;
break;
case 'I':
sp->zip |= SF_READ;
break;
case 'O':
sp->zip |= SF_WRITE;
break;
}
continue;
case 'D':
error_info.trace = -opt_info.num;
continue;
case 'K':
if (opt_info.offset)
{
opt_info.offset = 0;
opt_info.index++;
}
if (rskey(key, opt_info.arg, *opt_info.option))
return -1;
continue;
case 'L':
rskeylist(key, sfstdout, 0);
exit(0);
case 'P':
plugins = opt_info.arg;
continue;
case 'R':
key->disc->data = recstr(opt_info.arg, &e);
if (*e)
{
error(2, "%s: invalid record format", opt_info.arg);
return -1;
}
continue;
case 'S':
case 'y':
n = 'p';
s = opt_info.arg;
if (*s && *(e = s + strlen(s) - 1) != '%' && !isalpha(*e))
{
sfsprintf(opt, sizeof(opt), "%ski", s);
s = opt;
}
goto size;
case 'T':
pathtemp(NiL, 0, opt_info.arg, "/TMPPATH", NiL);
continue;
case 'X':
s = opt_info.arg;
opt_info.num = strton(s, &e, NiL, 1);
if (*e)
{
if (streq(s, "dump"))
opt_info.num = TEST_dump;
else if (streq(s, "io"))
opt_info.num = TEST_io;
else if (streq(s, "keys"))
opt_info.num = TEST_keys;
else if (streq(s, "read"))
opt_info.num = TEST_read;
else if (streq(s, "reserve"))
opt_info.num = TEST_reserve;
else if (streq(s, "show"))
opt_info.num = TEST_show;
else if (streq(s, "test"))
{
sfprintf(sfstdout, "ok\n");
exit(0);
}
else
error(1, "%s: unknown test", s);
}
if (*opt_info.option == '+')
key->test &= ~opt_info.num;
else
key->test |= opt_info.num;
sp->test = key->test;
sp->verbose = key->verbose || (key->test & TEST_show);
continue;
case '?':
error(ERROR_USAGE|4, "%s", opt_info.arg);
return 1;
case ':':
error(2, "%s", opt_info.arg);
return -1;
default:
opt[0] = opt_info.option[1];
opt[1] = 0;
if (rskeyopt(key, opt, 1))
return 0;
continue;
}
break;
}
argv += opt_info.index;
if (obsolescent && (opt_info.index <= 1 || !streq(*(argv - 1), "--")))
{
a = v = argv;
while (s = *a++)
{
if (*s == '-' && *(s + 1) == 'o')
{
if (!*(s += 2) && !(s = *a++))
{
error(2, "-o: output argument expected");
break;
}
key->output = s;
}
else
*v++ = s;
}
*v = 0;
}
key->input = argv;
while (lib = firstlib)
{
if (rslib(sp->rec, key, lib->name, 0))
return 1;
firstlib = firstlib->next;
free(lib);
}
if (plugins)
exit(showplugins(sp, key, plugins));
if (map = RECTYPE(key->disc->data) == REC_method && REC_M_INDEX(key->disc->data) == REC_M_path)
for (n = 0, i = -1; p = key->input[n]; n++)
if (s = strrchr(p, '%'))
{
r = recstr(s + 1, &e);
if (!*e || *e == '.' && e > (s + 1))
{
if (r != key->disc->data && i >= 0 && (RECTYPE(r) != REC_variable || RECTYPE(key->disc->data) != REC_variable || REC_V_ATTRIBUTES(r) != REC_V_ATTRIBUTES(key->disc->data)))
{
error(2, "%s: format %s incompatible with %s format %s", p, fmtrec(r, 0), key->input[i], fmtrec(key->disc->data, 0));
return 1;
}
if (RECTYPE(r) != REC_variable || RECTYPE(key->disc->data) != REC_variable || REC_V_SIZE(key->disc->data) < REC_V_SIZE(r))
key->disc->data = r;
i = n;
}
}
if (RECTYPE(key->disc->data) == REC_method && ((n = REC_M_INDEX(key->disc->data)) == REC_M_path || n == REC_M_data))
{
if ((sp->opened = fileopen(sp, key->input[0])) && (s = sfreserve(sp->opened, SF_UNBOUND, SF_LOCKR)))
{
struct stat st;
z = sfvalue(sp->opened);
if (fstat(sffileno(sp->opened), &st) || st.st_size < z)
st.st_size = 0;
key->disc->data = recfmt(s, z, st.st_size);
sfread(sp->opened, s, 0);
}
else
{
z = sp->opened ? sfvalue(sp->opened) : -1;
key->disc->data = REC_N_TYPE();
}
if (z && key->disc->data == REC_N_TYPE())
error(3, "%s: record format cannot be determined from data sample", key->input[0]);
}
if (RECTYPE(key->disc->data) == REC_fixed)
key->fixed = REC_F_SIZE(key->disc->data);
if (map && key->output && key->disc->data != REC_N_TYPE() && (stat(key->output, &st) || S_ISREG(st.st_mode)))
{
if (p = strrchr(key->output, '/'))
s = p + 1;
else
s = key->output;
if (!strchr(s, '%'))
{
p = key->output;
if (!(e = strrchr(s, '.')))
e = s + strlen(s);
if (RECTYPE(key->disc->data) == REC_variable && !REC_V_SIZE(key->disc->data))
key->disc->data |= ((1<<15)-1);
if (!(s = strdup(sfprints("%-*.*s%%%s%s", e - p, e - p, p, fmtrec(key->disc->data, 1), e))))
error(ERROR_SYSTEM|3, "out of space");
key->output = s;
if (sp->verbose)
error(0, "%s rename output %s => %s", error_info.id, p, s);
}
}
if (sp->verbose)
error(0, "%s %s record format", error_info.id, fmtrec(key->disc->data, 0));
return error_info.errors != 0;
}
static ssize_t
dumpkey(Rs_t* rs, unsigned char* dat, size_t datlen, unsigned char* key, size_t keylen, Rsdisc_t* disc)
{
Sort_t* sp = (Sort_t*)RSKEYDISC(disc);
ssize_t n;
int i;
char buf[2];
if ((n = (*sp->defkeyf)(rs, dat, datlen, key, keylen, disc)) > 0)
{
buf[1] = 0;
for (i = 0; i < n; i++)
{
buf[0] = key[i];
sfputr(sp->tp, fmtesc(buf), -1);
}
sfprintf(sfstderr, "key: %s\n", sfstruse(sp->tp));
}
return n;
}
static int
init(register Sort_t* sp, Rskeydisc_t* dp, char** argv)
{
register Rskey_t* key;
register char* s;
register char** p;
char* t;
int n;
unsigned long x;
unsigned long z;
size_t fixed;
struct stat is;
struct stat os;
memset(sp, 0, sizeof(*sp));
sp->xfiles = elementsof(sp->files);
sfset(sfstdout, SF_SHARE, 0);
sfset(sfstderr, SF_SHARE, 0);
Vmdcsbrk->round = INBRK;
dp->version = RSKEY_VERSION;
dp->flags = 0;
dp->errorf = errorf;
if (!(sp->key = key = rskeyopen(dp, NiL)) || !(sp->rec = rsnew(key->disc)))
return -1;
z = key->insize = 2 * INMAX;
#if 0
if (conformance(0, 0))
#endif
key->type |= RS_DATA;
if ((n = strtol(astconf("PAGESIZE", NiL, NiL), &t, 0)) > 0 && !*t)
key->alignsize = n;
if ((n = parse(sp, argv)) || rskeyinit(key))
{
rskeyclose(key);
if (n < 0)
error(ERROR_USAGE|4, "%s", optusage(NiL));
return -1;
}
if ((x = key->insize) != z)
z = 0;
if (x > INMAX)
x = INMAX;
else if (x < INMIN && !sp->chunk)
x = INMIN;
if (sp->single = !key->input[1])
{
if (!(sp->opened = fileopen(sp, key->input[0])))
error(ERROR_SYSTEM|3, "%s: cannot open", key->input[0]);
if (fstat(sffileno(sp->opened), &is))
error(ERROR_SYSTEM|3, "%s: cannot stat", key->input[0]);
if (!S_ISREG(is.st_mode) || sp->opened->disc)
{
sp->total = 0;
sp->test |= TEST_read;
}
else if (x > (sp->total = is.st_size))
x = sp->total;
}
else
sp->test |= TEST_read;
if (sp->zip & SF_READ)
sp->test |= TEST_read;
fixed = key->fixed;
if ((sp->test & TEST_reserve) || !(sp->test & TEST_read))
{
sp->map = 1;
if (z)
x = key->procsize;
if (fixed)
x += fixed - x % fixed;
}
else
{
if (z)
x = key->procsize;
for (;;)
{
if (fixed)
x += fixed - x % fixed;
if (sp->buf = (char*)vmalign(Vmheap, x, key->alignsize))
break;
if ((x >>= 1) < INMIN)
error(ERROR_SYSTEM|3, "out of space");
}
sp->hit = x - key->alignsize;
}
if (sp->test & TEST_keys)
{
if (!key->disc->defkeyf)
error(2, "no key function to intercept");
else if (!(sp->tp = sfstropen()))
error(ERROR_SYSTEM|3, "out of space");
else
{
sp->defkeyf = key->disc->defkeyf;
key->disc->defkeyf = dumpkey;
}
}
if (key->nproc > 1)
{
off_t offset;
off_t total;
off_t size;
size_t chunk;
int i;
Job_t* jp;
if (!sp->map || pathstdin(key->input[0]))
{
uno:
key->nproc = 1;
}
else if ((n = (sp->total + key->procsize - 1) / (key->procsize)) <= 1)
goto uno;
else
{
if (n < key->nproc)
key->nproc = n;
else
n = key->nproc;
if (!(sp->jobs = vmnewof(Vmheap, 0, Job_t, n, 0)))
goto uno;
size = (sp->total + n - 1) / n;
if (fixed)
{
if (size % fixed)
size += fixed - size % fixed;
i = (size + x - 1) / x;
if (i * n > sp->xfiles)
{
error(1, "multi-process multi-stage not implemented; falling back to one processor");
goto uno;
}
chunk = size / i;
if (chunk % fixed)
chunk += fixed - chunk % fixed;
size = chunk * i;
offset = 0;
total = sp->total;
for (jp = sp->jobs; jp < sp->jobs + n; jp++)
{
jp->offset = offset;
if (size > total)
size = total;
total -= (jp->size = size);
jp->chunk = chunk;
jp->intermediates = i;
offset += size;
}
if (key->procsize > chunk)
key->procsize = chunk;
else
{
size = key->procsize;
i = (chunk + size - 1) / size;
size = chunk / i;
if (size % fixed)
size += fixed - size % fixed;
key->procsize = size;
}
}
else
{
register char* s;
register char* t;
register char* b;
off_t ideal;
off_t scan;
char* file;
Sfio_t* ip;
i = (size + x - 1) / x;
if (i * n > sp->xfiles)
{
error(1, "multi-process multi-stage not implemented; falling back to one processor");
goto uno;
}
chunk = (size + i - 1) / i;
size = ideal = chunk * i;
if ((scan = INREC) >= ideal)
scan = (ideal / 32) * 4;
offset = 0;
total = sp->total;
file = key->input[0];
if (!(ip = fileopen(sp, file)))
error(ERROR_SYSTEM|3, "%s: cannot read", file);
for (jp = sp->jobs; jp < sp->jobs + n; jp++)
{
jp->offset = offset;
if (((size = ideal) + scan) >= total)
size = total;
else
{
size -= scan / 2;
if (sfseek(ip, offset + size, SEEK_SET) != (offset + size))
error(ERROR_SYSTEM|3, "%s: record boundary seek error at offset %lld", file, (Sflong_t)offset + size);
if (!(b = (char*)sfreserve(ip, scan, 0)))
error(ERROR_SYSTEM|3, "%s: record boundary read error at offset %lld", file, (Sflong_t)offset + size);
s = t = b + scan / 2 - 1;
while (*s++ != '\n')
{
if (t < b)
{
bigger:
if (((size += scan) + offset) >= (total - scan))
error(3, "%s: monster record at offset %lld", (Sflong_t)offset);
if (sfseek(ip, offset + size, SEEK_SET) != (offset + size))
error(ERROR_SYSTEM|3, "%s: record boundary input seek error at %lld", file, (Sflong_t)offset + size);
if (!(b = (char*)sfreserve(ip, scan, 0)))
error(ERROR_SYSTEM|3, "%s: record boundary read error at %lld", file, (Sflong_t)offset + size);
t = (s = b) + scan;
do
{
if (s >= t)
goto bigger;
} while (*s++ != '\n');
break;
}
if (*t-- == '\n')
{
s = t + 2;
break;
}
}
size += (s - b);
}
total -= (jp->size = size);
jp->chunk = (size + i - 1) / i;
if (jp->chunk > chunk)
chunk = jp->chunk;
jp->intermediates = i;
offset += size;
}
if (rsfileclose(sp->rec, ip))
return -1;
sfclose(ip);
key->procsize = (key->procsize > chunk) ? chunk : chunk / ((chunk + key->procsize - 1) / key->procsize);
}
}
}
key->insize = sp->end = x;
n = stat("/dev/null", &is);
if (pathstdout(key->output))
{
key->output = "/dev/stdout";
sp->op = sfstdout;
if (!n && !fstat(sffileno(sp->op), &os) && os.st_dev == is.st_dev && os.st_ino == is.st_ino)
key->type |= RS_IGNORE;
}
else if (key->input)
{
if (!stat(key->output, &os))
{
if (!n && os.st_dev == is.st_dev && os.st_ino == is.st_ino)
key->type |= RS_IGNORE;
else if (eaccess(key->output, W_OK))
error(ERROR_SYSTEM|3, "%s: cannot write", key->output);
else if (!fs3d(FS3D_TEST) || !iview(&os))
{
p = key->input;
while (s = *p++)
if (!pathstdin(s))
{
if (stat(s, &is))
error(ERROR_SYSTEM|2, "%s: not found", s);
else if (os.st_dev == is.st_dev && os.st_ino == is.st_ino)
{
if (t = strrchr(key->output, '/'))
{
s = key->output;
*t = 0;
}
else
s = ".";
if (sp->overwrite = pathtemp(NiL, 0, s, error_info.id, &n))
sp->op = sfnew(sfstdout, NiL, SF_UNBOUND, n, SF_WRITE);
if (t) *t = '/';
if (!sp->op || fstat(n, &is))
error(ERROR_SYSTEM|3, "%s: cannot create overwrite file %s", key->output, sp->overwrite);
if (os.st_uid != is.st_uid || os.st_gid != is.st_gid)
sp->preserve = 1;
break;
}
}
}
}
if (!sp->overwrite && !(sp->op = sfopen(sfstdout, key->output, "w")))
error(ERROR_SYSTEM|3, "%s: cannot write", key->output);
}
if (rsfilewrite(sp->rec, sp->op, key->output))
return -1;
if (key->outsize > 0)
sfsetbuf(sp->op, NiL, key->outsize);
if (sp->zip & SF_WRITE)
sfdcgzip(sp->op, 0);
if (rsinit(sp->rec, key->meth, key->procsize, key->type, key))
{
error(ERROR_SYSTEM|2, "sort library initialization error");
rskeyclose(key);
return -1;
}
if (sp->rec->meth->type == RS_MTCOPY)
sp->chunk = 1;
if (sp->test & TEST_io)
{
for (n = 0; s = key->input[n]; n++)
error(0, "%s input[%d]\t\"%s\"", error_info.id, n, s);
if (s = key->output)
error(0, "%s output\t\"%s\"", error_info.id, s);
}
return 0;
}
static void
clear(register Sort_t* sp, Sfio_t* fp)
{
register int i;
for (i = fp ? sp->mfiles : 0; i < sp->nfiles; i++)
{
rstempclose(sp->rec, sp->files[i]);
sp->files[i] = 0;
}
if (fp)
{
sp->files[sp->mfiles++] = fp;
sp->nfiles = sp->mfiles;
if (sp->mfiles >= (sp->xfiles - 1))
sp->mfiles = 0;
}
else
sp->nfiles = sp->mfiles = 0;
}
static ssize_t
flush(register Sort_t* sp, register size_t r)
{
register Sfio_t* fp;
register size_t n;
register size_t m;
register size_t b;
if (sp->chunk)
{
if (rswrite(sp->rec, sp->op, RS_OTEXT))
{
if (!error_info.errors)
error(ERROR_SYSTEM|2, "%s: write error", sp->key->output);
return -1;
}
}
else if (sp->rec->meth->type != RS_MTVERIFY)
{
if (!(fp = sp->files[sp->nfiles]))
{
if (sp->child || !(fp = rstempwrite(sp->rec, (Sfio_t*)0)))
error(ERROR_SYSTEM|3, "cannot create intermediate sort file %d", sp->nfiles);
sp->files[sp->nfiles] = fp;
}
sp->nfiles++;
if (sp->verbose)
error(0, "%s write intermediate", error_info.id);
if (rswrite(sp->rec, fp, 0))
{
error(ERROR_SYSTEM|2, "intermediate sort file write error");
return -1;
}
if (rstempread(sp->rec, fp))
{
error(ERROR_SYSTEM|2, "intermediate sort file rewind error");
return -1;
}
if (sp->nfiles >= sp->xfiles)
{
if (sp->child || !(fp = rstempwrite(sp->rec, (Sfio_t*)0)))
error(ERROR_SYSTEM|3, "cannot create intermediate merge file");
if (sp->verbose)
error(0, "%s merge multi-stage intermediate", error_info.id);
if (rsmerge(sp->rec, fp, sp->files + sp->mfiles, sp->nfiles - sp->mfiles, 0))
{
error(ERROR_SYSTEM|2, "intermediate merge file write error");
return -1;
}
if (rstempread(sp->rec, fp))
{
error(ERROR_SYSTEM|2, "intermediate merge file rewind error");
return -1;
}
clear(sp, fp);
}
}
if (!sp->map && (m = sp->cur - r))
{
n = roundof(m, sp->key->alignsize) - m;
if (n < r)
{
m = n;
while (r < sp->cur)
sp->buf[n++] = sp->buf[r++];
sp->cur = n;
}
else
{
b = r;
r += m;
n += m;
sp->cur = n;
while (r > b)
sp->buf[--n] = sp->buf[--r];
m = n;
}
}
else
m = sp->cur = 0;
return m;
}
static int
input(register Sort_t* sp, Sfio_t* ip, const char* name)
{
register ssize_t n;
register ssize_t p;
register ssize_t m;
register ssize_t r;
size_t z;
char* b;
int c;
char del[2];
error_info.file = ip == sfstdin ? (char*)0 : (char*)name;
m = -1;
z = 0;
if (sp->bufsize)
sfsetbuf(ip, NiL, sp->bufsize);
else if (sfsize(ip) > SF_BUFSIZE)
{
if (sp->map)
sfsetbuf(ip, NiL, z = sp->end);
else
sfsetbuf(ip, NiL, 0);
}
r = sp->cur = roundof(sp->cur, sp->key->alignsize);
p = 0;
for (;;)
{
if (sp->cur > sp->hit)
{
if (sp->single && !sp->nfiles && sp->total == (sp->map ? 0 : p))
break;
if ((r = flush(sp, r)) < 0)
return -1;
}
if (!sp->map)
n = sfeof(ip) ? 0 : sfread(ip, sp->buf + sp->cur, sp->end - sp->cur);
else
{
sp->buf = (char*)sfreserve(ip, m, SF_LOCKR);
n = sfvalue(ip);
if (!sp->buf)
{
if (m < 0 && n < -m && z == sp->end)
{
sfsetbuf(ip, NiL, z = 2 * sp->end);
sp->buf = (char*)sfreserve(ip, m, SF_LOCKR);
n = sfvalue(ip);
if (sp->verbose && n)
error(0, "%s buffer boundary expand to %I*d", error_info.id, sizeof(n), n);
}
if (!sp->buf && n > 0 && !(sp->buf = sfreserve(ip, n, SF_LOCKR)))
n = -1;
}
}
if (n <= 0)
{
if (n < 0)
error(ERROR_SYSTEM|3, "read error");
if (sp->cur <= r)
break;
if (sp->key->fixed)
{
error(1, "incomplete record length=%lld", (Sflong_t)(sp->cur - r));
break;
}
if (RECTYPE(sp->key->disc->data) == REC_delimited && sp->buf[sp->cur - 1] != (c = REC_D_DELIMITER(sp->key->disc->data)))
{
sp->buf[sp->cur++] = c;
if (c == '\n')
error(1, "newline appended");
else
{
del[0] = c;
del[1] = 0;
error(1, "%s appended", fmtquote(del, "'", NiL, 1, 0));
}
}
}
sp->cur += n;
process:
if (sp->verbose && !sp->child)
error(ERROR_PROMPT, "%s process %lld ->", error_info.id, (Sflong_t)(sp->cur - r));
if ((p = rsprocess(sp->rec, sp->buf + r, sp->cur - r)) < 0)
error(ERROR_SYSTEM|3, "sort error");
if (sp->verbose)
{
if (sp->child)
error(0, "%s process %lld -> %lld", error_info.id, (Sflong_t)(sp->cur - r), (Sflong_t)p);
else
error(0, " %lld", (Sflong_t)p);
}
if (sp->map)
{
if (sp->map > 2)
break;
sfread(ip, sp->buf, p);
if (p)
{
m = -(n - p + 1);
if (((sp->total -= p) / 3) < (sp->end / 2) && sp->total > sp->end)
{
if ((r = flush(sp, r)) < 0)
return -1;
sfsetbuf(ip, NiL, sp->total);
}
}
else if (sp->map == 1)
{
sp->map++;
m = -(n + 1);
}
else if (n > sp->end)
{
error(2, "monster record", n, sp->end, sp->cur, p);
break;
}
else if (sp->key->fixed)
{
error(1, "incomplete record length=%ld", n - p);
break;
}
else
{
sp->cur = n - p;
if (!(b = vmnewof(Vmheap, 0, char, sp->cur, 1)))
error(ERROR_SYSTEM|3, "out of space");
memcpy(b, sp->buf + p, sp->cur);
if (RECTYPE(sp->key->disc->data) == REC_delimited && b[sp->cur - 1] != (c = REC_D_DELIMITER(sp->key->disc->data)))
{
b[sp->cur++] = '\n';
if (c == '\n')
error(1, "newline appended");
else
{
del[0] = c;
del[1] = 0;
error(1, "%s appended", fmtquote(del, "'", "'", 1, 0));
}
}
sp->buf = b;
sp->map++;
goto process;
}
}
else
r += p;
}
error_info.file = 0;
return 0;
}
static ssize_t
partread(Sfio_t* fp, Void_t* buf, size_t size, Sfdisc_t* dp)
{
register Part_t* pp = (Part_t*)dp;
if (pp->remain <= 0)
return 0;
if (size > pp->remain)
size = pp->remain;
pp->remain -= size;
return sfrd(fp, buf, size, dp);
}
static Sfoff_t
partseek(Sfio_t* fp, Sfoff_t lloffset, int op, Sfdisc_t* dp)
{
register Part_t* pp = (Part_t*)dp;
off_t offset = lloffset;
switch (op)
{
case SEEK_SET:
offset += pp->offset;
break;
case SEEK_CUR:
offset += pp->offset;
break;
case SEEK_END:
offset = pp->offset + pp->size - offset;
op = SEEK_SET;
break;
}
if ((offset = sfsk(fp, offset, op, dp)) >= 0)
{
offset -= pp->offset;
pp->remain = pp->size - offset;
}
return offset;
}
static void
jobs(register Sort_t* sp)
{
register Job_t* jp;
register Job_t* xp;
register int i;
register int j;
register int f;
int status;
char* file;
Sfio_t* ip;
Part_t part;
char id[32];
sp->single = 0;
if (sp->verbose)
error(0, "%s %d processes %lld total", error_info.id, sp->key->nproc, (Sflong_t)sp->total);
xp = sp->jobs + sp->key->nproc;
if (sp->test & TEST_show)
{
for (jp = sp->jobs; jp < xp; jp++)
error(0, "%s#%d pos %12lld : len %10lld : buf %10lld : num %2d", error_info.id, jp - sp->jobs + 1, (Sflong_t)jp->offset, (Sflong_t)jp->size, (Sflong_t)jp->chunk, jp->intermediates);
exit(0);
}
f = 0;
for (jp = sp->jobs; jp < xp; jp++)
for (i = 0; i < jp->intermediates; i++)
if (!(sp->files[f++] = rstempwrite(sp->rec, (Sfio_t*)0)))
error(ERROR_SYSTEM|3, "cannot create intermediate file %d", i);
part.disc.readf = partread;
part.disc.writef = 0;
part.disc.seekf = partseek;
part.disc.exceptf = 0;
part.disc.disc = 0;
file = sp->key->input[0];
j = 0;
for (jp = sp->jobs; jp < xp; jp++)
{
ip = fileopen(sp, file);
switch (fork())
{
case -1:
error(ERROR_SYSTEM|3, "not enough child processes");
case 0:
sp->child = 1;
sfsprintf(id, sizeof(id), "%s#%d", error_info.id, jp - sp->jobs + 1);
error_info.id = id;
sp->end = jp->chunk;
part.offset = jp->offset;
sp->total = part.size = part.remain = jp->size;
sfdisc(ip, &part.disc);
for (i = 0; i < jp->intermediates; i++)
sp->files[i] = sp->files[j++];
while (i < f)
sp->files[i++] = 0;
if (sp->verbose)
error(0, "%s pos %12lld : len %10lld : buf %10lld : num %2d", error_info.id, (Sflong_t)jp->offset, (Sflong_t)jp->size, (Sflong_t)jp->chunk, jp->intermediates);
exit(input(sp, ip, file) < 0);
}
if (rsfileclose(sp->rec, ip))
exit(1);
sfclose(ip);
j += jp->intermediates;
}
sp->nfiles = f;
i = 0;
j = sp->key->nproc;
while (j > 0)
{
if (wait(&status) != -1)
{
if (status)
i++;
j--;
}
else if (errno != EINTR)
{
error(ERROR_SYSTEM|3, "%d process%s did not complete", j, j == 1 ? "" : "es");
break;
}
}
if (i)
error(3, "%d child process%s failed", i, i == 1 ? "" : "es");
}
static void
done(register Sort_t* sp)
{
while (rsdisc(sp->rec, NiL, RS_POP));
if ((sfsync(sp->op) || sp->op != sfstdout && rsfileclose(sp->rec, sp->op)) && !error_info.errors)
error(ERROR_SYSTEM|2, "%s: write error", sp->key->output);
if (rsclose(sp->rec))
error(2, "sort error");
if (sp->map > 2)
vmfree(Vmheap, sp->buf);
if (sp->overwrite)
{
if (error_info.errors)
remove(sp->overwrite);
else if (sp->preserve)
{
Sfio_t* ip;
Sfio_t* op;
if (ip = sfopen(NiL, sp->overwrite, "r"))
{
if (op = sfopen(NiL, sp->key->output, "w"))
{
if ((sfmove(ip, op, SF_UNBOUND, -1) < 0 || sfclose(op) || !sfeof(ip)) && !error_info.errors)
error(ERROR_SYSTEM|2, "%s: write error", sp->key->output);
sfclose(op);
}
else
error(ERROR_SYSTEM|2, "%s: cannot write", sp->key->output);
sfclose(ip);
remove(sp->overwrite);
}
else
error(ERROR_SYSTEM|2, "%s: cannot read", sp->overwrite);
sp->preserve = 0;
}
else if (remove(sp->key->output) || rename(sp->overwrite, sp->key->output))
error(ERROR_SYSTEM|2, "%s: cannot overwrite", sp->key->output);
free(sp->overwrite);
sp->overwrite = 0;
}
if (rskeyclose(sp->key))
error(2, "sort key error");
}
int
main(int argc, char** argv)
{
register char* s;
register Sfio_t* fp;
char** merge;
Sort_t sort;
error_info.id = "sort";
if (init(&sort, &sort.disc, argv))
exit(1);
if (sort.test & TEST_dump)
{
sfprintf(sfstderr, "main\n\tintermediates=%d\n", sort.xfiles);
rskeydump(sort.key, sfstderr);
}
if (sort.key->type & RS_CAT)
{
while (s = *sort.key->input++)
{
fp = fileopen(&sort, s);
if (sfmove(fp, sfstdout, SF_UNBOUND, -1) < 0 || !sfeof(fp))
error(ERROR_SYSTEM|2, "%s: read error", s);
if (sferror(sfstdout) || rsfileclose(sort.rec, fp))
break;
}
}
else
{
merge = sort.key->merge && sort.key->input[0] && sort.key->input[1] ? sort.key->input : (char**)0;
fp = 0;
if (sort.jobs)
jobs(&sort);
else if (sort.test & TEST_show)
exit(0);
else
while (s = *sort.key->input++)
{
fp = fileopen(&sort, s);
if (merge)
{
if (sort.nfiles >= sort.xfiles)
{
clear(&sort, NiL);
sort.key->input = merge;
merge = 0;
if (fp != sfstdin)
sfclose(fp);
fp = 0;
continue;
}
sort.files[sort.nfiles++] = fp;
}
else if (input(&sort, fp, s) < 0)
break;
else if (fp != sfstdin && !sort.map)
{
sfclose(fp);
fp = 0;
}
}
if (sort.nfiles)
{
if (sort.cur && flush(&sort, sort.cur) < 0)
return 1;
if (sort.verbose)
error(0, "%s merge text", error_info.id);
if (rsmerge(sort.rec, sort.op, sort.files, sort.nfiles, merge ? RS_TEXT : RS_OTEXT))
error(ERROR_SYSTEM|3, "merge error");
clear(&sort, NiL);
}
else
{
if (sort.verbose)
error(0, "%s write text", error_info.id);
if (rswrite(sort.rec, sort.op, RS_OTEXT) && !error_info.errors)
error(ERROR_SYSTEM|3, "%s: write error", sort.key->output);
if (fp && fp != sfstdin)
sfclose(fp);
}
}
done(&sort);
exit(error_info.errors != 0);
}
