diff options
Diffstat (limited to 'text_cmds/sort/file.c')
| -rw-r--r-- | text_cmds/sort/file.c | 1684 |
1 files changed, 1684 insertions, 0 deletions
diff --git a/text_cmds/sort/file.c b/text_cmds/sort/file.c new file mode 100644 index 0000000..346819b --- /dev/null +++ b/text_cmds/sort/file.c @@ -0,0 +1,1684 @@ +/*- + * Copyright (C) 2009 Gabor Kovesdan <gabor@FreeBSD.org> + * Copyright (C) 2012 Oleg Moskalenko <mom040267@gmail.com> + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS + * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) + * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY + * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF + * SUCH DAMAGE. + */ + +#include <sys/cdefs.h> +__FBSDID("$FreeBSD$"); + +#include <sys/mman.h> +#include <sys/stat.h> +#include <sys/types.h> +#include <sys/queue.h> + +#include <err.h> +#include <fcntl.h> +#if defined(SORT_THREADS) +#include <pthread.h> +#endif +#ifndef __APPLE__ +#include <semaphore.h> +#else +#include <mach/mach_init.h> +#include <mach/mach_error.h> +#include <mach/semaphore.h> +#include <mach/task.h> +#endif +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> +#include <wchar.h> +#include <wctype.h> + +#include "coll.h" +#include "file.h" +#include "radixsort.h" + +unsigned long long free_memory = 1000000; +unsigned long long available_free_memory = 1000000; + +bool use_mmap; + +const char *tmpdir = "/var/tmp"; +const char *compress_program; + +size_t max_open_files = 16; + +/* + * How much space we read from file at once + */ +#define READ_CHUNK (4096) + +/* + * File reader structure + */ +struct file_reader +{ + struct reader_buffer rb; + FILE *file; + char *fname; + unsigned char *buffer; + unsigned char *mmapaddr; + unsigned char *mmapptr; + size_t bsz; + size_t cbsz; + size_t mmapsize; + size_t strbeg; + int fd; + char elsymb; +}; + +/* + * Structure to be used in file merge process. + */ +struct file_header +{ + struct file_reader *fr; + struct sort_list_item *si; /* current top line */ + size_t file_pos; +}; + +/* + * List elements of "cleanable" files list. + */ +struct CLEANABLE_FILE +{ + char *fn; + LIST_ENTRY(CLEANABLE_FILE) files; +}; + +/* + * List header of "cleanable" files list. + */ +static LIST_HEAD(CLEANABLE_FILES,CLEANABLE_FILE) tmp_files; + +/* + * Semaphore to protect the tmp file list. + * We use semaphore here because it is signal-safe, according to POSIX. + * And semaphore does not require pthread library. + */ +#ifndef __APPLE__ +static sem_t tmp_files_sem; +#else +static semaphore_t tmp_files_sem; +#endif + +static void mt_sort(struct sort_list *list, + int (*sort_func)(void *, size_t, size_t, + int (*)(const void *, const void *)), const char* fn); + +/* + * Init tmp files list + */ +void +init_tmp_files(void) +{ + + LIST_INIT(&tmp_files); +#ifndef __APPLE__ + sem_init(&tmp_files_sem, 0, 1); +#else + { + mach_port_t self = mach_task_self(); + kern_return_t ret = semaphore_create(self, &tmp_files_sem, SYNC_POLICY_FIFO, 1); + if (ret != KERN_SUCCESS) { + err(2,NULL); + } + } +#endif +} + +/* + * Save name of a tmp file for signal cleanup + */ +void +tmp_file_atexit(const char *tmp_file) +{ + + if (tmp_file) { +#ifndef __APPLE__ + sem_wait(&tmp_files_sem); +#else + semaphore_wait(tmp_files_sem); +#endif + struct CLEANABLE_FILE *item = + sort_malloc(sizeof(struct CLEANABLE_FILE)); + item->fn = sort_strdup(tmp_file); + LIST_INSERT_HEAD(&tmp_files, item, files); +#ifndef __APPLE__ + sem_post(&tmp_files_sem); +#else + semaphore_signal(tmp_files_sem); +#endif + } +} + +/* + * Clear tmp files + */ +void +clear_tmp_files(void) +{ + struct CLEANABLE_FILE *item; + +#ifndef __APPLE__ + sem_wait(&tmp_files_sem); +#else + semaphore_wait(tmp_files_sem); +#endif + LIST_FOREACH(item,&tmp_files,files) { + if ((item) && (item->fn)) + unlink(item->fn); + } +#ifndef __APPLE__ + sem_post(&tmp_files_sem); +#else + semaphore_signal(tmp_files_sem); +#endif +} + +/* + * Check whether a file is a temporary file + */ +static bool +file_is_tmp(const char* fn) +{ + struct CLEANABLE_FILE *item; + bool ret = false; + + if (fn) { +#ifndef __APPLE__ + sem_wait(&tmp_files_sem); +#else + semaphore_wait(tmp_files_sem); +#endif + LIST_FOREACH(item,&tmp_files,files) { + if ((item) && (item->fn)) + if (strcmp(item->fn, fn) == 0) { + ret = true; + break; + } + } +#ifndef __APPLE__ + sem_post(&tmp_files_sem); +#else + semaphore_signal(tmp_files_sem); +#endif + } + + return (ret); +} + +/* + * Generate new temporary file name + */ +char * +new_tmp_file_name(void) +{ + static size_t tfcounter = 0; + static const char *fn = ".bsdsort."; + char *ret; + size_t sz; + + sz = strlen(tmpdir) + 1 + strlen(fn) + 32 + 1; + ret = sort_malloc(sz); + + sprintf(ret, "%s/%s%d.%lu", tmpdir, fn, (int) getpid(), (unsigned long)(tfcounter++)); + tmp_file_atexit(ret); + return (ret); +} + +/* + * Initialize file list + */ +void +file_list_init(struct file_list *fl, bool tmp) +{ + + if (fl) { + fl->count = 0; + fl->sz = 0; + fl->fns = NULL; + fl->tmp = tmp; + } +} + +/* + * Add a file name to the list + */ +void +file_list_add(struct file_list *fl, char *fn, bool allocate) +{ + + if (fl && fn) { + if (fl->count >= fl->sz || (fl->fns == NULL)) { + fl->sz = (fl->sz) * 2 + 1; + fl->fns = sort_realloc(fl->fns, fl->sz * + sizeof(char *)); + } + fl->fns[fl->count] = allocate ? sort_strdup(fn) : fn; + fl->count += 1; + } +} + +/* + * Populate file list from array of file names + */ +void +file_list_populate(struct file_list *fl, int argc, char **argv, bool allocate) +{ + + if (fl && argv) { + int i; + + for (i = 0; i < argc; i++) + file_list_add(fl, argv[i], allocate); + } +} + +/* + * Clean file list data and delete the files, + * if this is a list of temporary files + */ +void +file_list_clean(struct file_list *fl) +{ + + if (fl) { + if (fl->fns) { + size_t i; + + for (i = 0; i < fl->count; i++) { + if (fl->fns[i]) { + if (fl->tmp) + unlink(fl->fns[i]); + sort_free(fl->fns[i]); + fl->fns[i] = 0; + } + } + sort_free(fl->fns); + fl->fns = NULL; + } + fl->sz = 0; + fl->count = 0; + fl->tmp = false; + } +} + +/* + * Init sort list + */ +void +sort_list_init(struct sort_list *l) +{ + + if (l) { + l->count = 0; + l->size = 0; + l->memsize = sizeof(struct sort_list); + l->list = NULL; + } +} + +/* + * Add string to sort list + */ +void +sort_list_add(struct sort_list *l, struct bwstring *str) +{ + + if (l && str) { + size_t indx = l->count; + + if ((l->list == NULL) || (indx >= l->size)) { + size_t newsize = (l->size + 1) + 1024; + + l->list = sort_realloc(l->list, + sizeof(struct sort_list_item*) * newsize); + l->memsize += (newsize - l->size) * + sizeof(struct sort_list_item*); + l->size = newsize; + } + l->list[indx] = sort_list_item_alloc(); + sort_list_item_set(l->list[indx], str); + l->memsize += sort_list_item_size(l->list[indx]); + l->count += 1; + } +} + +/* + * Clean sort list data + */ +void +sort_list_clean(struct sort_list *l) +{ + + if (l) { + if (l->list) { + size_t i; + + for (i = 0; i < l->count; i++) { + struct sort_list_item *item; + + item = l->list[i]; + + if (item) { + sort_list_item_clean(item); + sort_free(item); + l->list[i] = NULL; + } + } + sort_free(l->list); + l->list = NULL; + } + l->count = 0; + l->size = 0; + l->memsize = sizeof(struct sort_list); + } +} + +/* + * Write sort list to file + */ +void +sort_list_dump(struct sort_list *l, const char *fn) +{ + + if (l && fn) { + FILE *f; + + f = openfile(fn, "w"); + if (f == NULL) + err(2, NULL); + + if (l->list) { + size_t i; + if (!(sort_opts_vals.uflag)) { + for (i = 0; i < l->count; ++i) + bwsfwrite(l->list[i]->str, f, + sort_opts_vals.zflag); + } else { + struct sort_list_item *last_printed_item = NULL; + struct sort_list_item *item; + for (i = 0; i < l->count; ++i) { + item = l->list[i]; + if ((last_printed_item == NULL) || + list_coll(&last_printed_item, &item)) { + bwsfwrite(item->str, f, sort_opts_vals.zflag); + last_printed_item = item; + } + } + } + } + + closefile(f, fn); + } +} + +/* + * Checks if the given file is sorted. Stops at the first disorder, + * prints the disordered line and returns 1. + */ +int +check(const char *fn) +{ + struct bwstring *s1, *s2, *s1disorder, *s2disorder; + struct file_reader *fr; + struct keys_array *ka1, *ka2; + int res; + size_t pos, posdisorder; + + s1 = s2 = s1disorder = s2disorder = NULL; + ka1 = ka2 = NULL; + + fr = file_reader_init(fn); + + res = 0; + pos = 1; + posdisorder = 1; + + if (fr == NULL) { + err(2, NULL); +#ifndef __APPLE__ + goto end; +#endif + } + + s1 = file_reader_readline(fr); + if (s1 == NULL) + goto end; + + ka1 = keys_array_alloc(); + preproc(s1, ka1); + + s2 = file_reader_readline(fr); + if (s2 == NULL) + goto end; + + ka2 = keys_array_alloc(); + preproc(s2, ka2); + + for (;;) { + + if (debug_sort) { + bwsprintf(stdout, s2, "s1=<", ">"); + bwsprintf(stdout, s1, "s2=<", ">"); + } + int cmp = key_coll(ka2, ka1, 0); + if (debug_sort) + printf("; cmp1=%d", cmp); + + if (!cmp && sort_opts_vals.complex_sort && + !(sort_opts_vals.uflag) && !(sort_opts_vals.sflag)) { + cmp = top_level_str_coll(s2, s1); + if (debug_sort) + printf("; cmp2=%d", cmp); + } + if (debug_sort) + printf("\n"); + + if ((sort_opts_vals.uflag && (cmp <= 0)) || (cmp < 0)) { + if (!(sort_opts_vals.csilentflag)) { + s2disorder = bwsdup(s2); + posdisorder = pos; + if (debug_sort) + s1disorder = bwsdup(s1); + } + res = 1; + goto end; + } + + pos++; + + clean_keys_array(s1, ka1); + sort_free(ka1); + ka1 = ka2; + ka2 = NULL; + + bwsfree(s1); + s1 = s2; + + s2 = file_reader_readline(fr); + if (s2 == NULL) + goto end; + + ka2 = keys_array_alloc(); + preproc(s2, ka2); + } + +end: + if (ka1) { + clean_keys_array(s1, ka1); + sort_free(ka1); + } + + if (s1) + bwsfree(s1); + + if (ka2) { + clean_keys_array(s2, ka2); + sort_free(ka2); + } + + if (s2) + bwsfree(s2); + + if ((fn == NULL) || (*fn == 0) || (strcmp(fn, "-") == 0)) { + for (;;) { + s2 = file_reader_readline(fr); + if (s2 == NULL) + break; + bwsfree(s2); + } + } + + file_reader_free(fr); + + if (s2disorder) { + bws_disorder_warnx(s2disorder, fn, posdisorder); + if (s1disorder) { + bws_disorder_warnx(s1disorder, fn, posdisorder); + if (s1disorder != s2disorder) + bwsfree(s1disorder); + } + bwsfree(s2disorder); + s1disorder = NULL; + s2disorder = NULL; + } + + if (res) + exit(res); + + return (0); +} + +/* + * Opens a file. If the given filename is "-", stdout will be + * opened. + */ +FILE * +openfile(const char *fn, const char *mode) +{ + FILE *file; + + if (strcmp(fn, "-") == 0) { + return ((mode && mode[0] == 'r') ? stdin : stdout); + } else { + mode_t orig_file_mask = 0; + int is_tmp = file_is_tmp(fn); + + if (is_tmp && (mode[0] == 'w')) + orig_file_mask = umask(S_IWGRP | S_IWOTH | + S_IRGRP | S_IROTH); + + if (is_tmp && (compress_program != NULL)) { + char *cmd; + size_t cmdsz; + + cmdsz = strlen(fn) + 128; + cmd = sort_malloc(cmdsz); + + fflush(stdout); + + if (mode[0] == 'r') + snprintf(cmd, cmdsz - 1, "cat %s | %s -d", + fn, compress_program); + else if (mode[0] == 'w') + snprintf(cmd, cmdsz - 1, "%s > %s", + compress_program, fn); + else + err(2, "%s", getstr(7)); + + if ((file = popen(cmd, mode)) == NULL) + err(2, NULL); + + sort_free(cmd); + + } else + if ((file = fopen(fn, mode)) == NULL) + err(2, NULL); + + if (is_tmp && (mode[0] == 'w')) + umask(orig_file_mask); + } + + return (file); +} + +/* + * Close file + */ +void +closefile(FILE *f, const char *fn) +{ + if (f == NULL) { + ; + } else if (f == stdin) { + ; + } else if (f == stdout) { + fflush(f); + } else { + if (file_is_tmp(fn) && compress_program != NULL) { + if(pclose(f)<0) + err(2,NULL); + } else + fclose(f); + } +} + +/* + * Reads a file into the internal buffer. + */ +struct file_reader * +file_reader_init(const char *fsrc) +{ + struct file_reader *ret; + + if (fsrc == NULL) + fsrc = "-"; + + ret = sort_malloc(sizeof(struct file_reader)); + memset(ret, 0, sizeof(struct file_reader)); + + ret->elsymb = '\n'; + if (sort_opts_vals.zflag) + ret->elsymb = 0; + + ret->fname = sort_strdup(fsrc); + + if (strcmp(fsrc, "-") && (compress_program == NULL) && use_mmap) { + + do { + struct stat stat_buf; + void *addr; + size_t sz = 0; + int fd, flags; + +#if defined(__APPLE__) + flags = 0; +#else + flags = MAP_NOCORE | MAP_NOSYNC; +#endif + + fd = open(fsrc, O_RDONLY); + if (fd < 0) + err(2, NULL); + + if (fstat(fd, &stat_buf) < 0) { + close(fd); + break; + } + + sz = stat_buf.st_size; + +#if defined(MAP_PREFAULT_READ) + flags |= MAP_PREFAULT_READ; +#endif + + addr = mmap(NULL, sz, PROT_READ, flags, fd, 0); + if (addr == MAP_FAILED) { + close(fd); + break; + } + + ret->fd = fd; + ret->mmapaddr = addr; + ret->mmapsize = sz; + ret->mmapptr = ret->mmapaddr; + + } while (0); + } + + if (ret->mmapaddr == NULL) { + ret->file = openfile(fsrc, "r"); + if (ret->file == NULL) + err(2, NULL); + + if (strcmp(fsrc, "-")) { + ret->cbsz = READ_CHUNK; + ret->buffer = sort_malloc(ret->cbsz); + ret->bsz = 0; + ret->strbeg = 0; + + ret->bsz = fread(ret->buffer, 1, ret->cbsz, ret->file); + if (ret->bsz == 0) { + if (ferror(ret->file)) + err(2, NULL); + } + } + } + + return (ret); +} + +struct bwstring * +file_reader_readline(struct file_reader *fr) +{ + struct bwstring *ret = NULL; + + if (fr->mmapaddr) { + unsigned char *mmapend; + + mmapend = fr->mmapaddr + fr->mmapsize; + if (fr->mmapptr >= mmapend) + return (NULL); + else { + unsigned char *strend; + size_t sz; + + sz = mmapend - fr->mmapptr; + strend = memchr(fr->mmapptr, fr->elsymb, sz); + + if (strend == NULL) { + ret = bwscsbdup(fr->mmapptr, sz); + fr->mmapptr = mmapend; + } else { + ret = bwscsbdup(fr->mmapptr, strend - + fr->mmapptr); + fr->mmapptr = strend + 1; + } + } + + } else if (fr->file != stdin) { + unsigned char *strend; + size_t bsz1, remsz, search_start; + + search_start = 0; + remsz = 0; + strend = NULL; + + if (fr->bsz > fr->strbeg) + remsz = fr->bsz - fr->strbeg; + + /* line read cycle */ + for (;;) { + if (remsz > search_start) + strend = memchr(fr->buffer + fr->strbeg + + search_start, fr->elsymb, remsz - + search_start); + else + strend = NULL; + + if (strend) + break; + if (feof(fr->file)) + break; + + if (fr->bsz != fr->cbsz) + /* NOTREACHED */ + err(2, "File read software error 1"); + + if (remsz > (READ_CHUNK >> 1)) { + search_start = fr->cbsz - fr->strbeg; + fr->cbsz += READ_CHUNK; + fr->buffer = sort_realloc(fr->buffer, + fr->cbsz); + bsz1 = fread(fr->buffer + fr->bsz, 1, + READ_CHUNK, fr->file); + if (bsz1 == 0) { + if (ferror(fr->file)) + err(2, NULL); + break; + } + fr->bsz += bsz1; + remsz += bsz1; + } else { + if (remsz > 0 && fr->strbeg>0) + bcopy(fr->buffer + fr->strbeg, + fr->buffer, remsz); + + fr->strbeg = 0; + search_start = remsz; + bsz1 = fread(fr->buffer + remsz, 1, + fr->cbsz - remsz, fr->file); + if (bsz1 == 0) { + if (ferror(fr->file)) + err(2, NULL); + break; + } + fr->bsz = remsz + bsz1; + remsz = fr->bsz; + } + } + + if (strend == NULL) + strend = fr->buffer + fr->bsz; + + if ((fr->buffer + fr->strbeg <= strend) && + (fr->strbeg < fr->bsz) && (remsz>0)) + ret = bwscsbdup(fr->buffer + fr->strbeg, strend - + fr->buffer - fr->strbeg); + + fr->strbeg = (strend - fr->buffer) + 1; + + } else { + size_t len = 0; + + ret = bwsfgetln(fr->file, &len, sort_opts_vals.zflag, + &(fr->rb)); + } + + return (ret); +} + +static void +file_reader_clean(struct file_reader *fr) +{ + + if (fr) { + if (fr->mmapaddr) + munmap(fr->mmapaddr, fr->mmapsize); + + if (fr->fd) + close(fr->fd); + + if (fr->buffer) + sort_free(fr->buffer); + + if (fr->file) + if (fr->file != stdin) + closefile(fr->file, fr->fname); + + if(fr->fname) + sort_free(fr->fname); + + memset(fr, 0, sizeof(struct file_reader)); + } +} + +void +file_reader_free(struct file_reader *fr) +{ + + if (fr) { + file_reader_clean(fr); + sort_free(fr); + } +} + +int +procfile(const char *fsrc, struct sort_list *list, struct file_list *fl) +{ + struct file_reader *fr; + + fr = file_reader_init(fsrc); + if (fr == NULL) + err(2, NULL); + + /* file browse cycle */ + for (;;) { + struct bwstring *bws; + + bws = file_reader_readline(fr); + + if (bws == NULL) + break; + + sort_list_add(list, bws); + + if (list->memsize >= available_free_memory) { + char *fn; + + fn = new_tmp_file_name(); + sort_list_to_file(list, fn); + file_list_add(fl, fn, false); + sort_list_clean(list); + } + } + + file_reader_free(fr); + + return (0); +} + +/* + * Compare file headers. Files with EOF always go to the end of the list. + */ +static int +file_header_cmp(struct file_header *f1, struct file_header *f2) +{ + + if (f1 == f2) + return (0); + else { + if (f1->fr == NULL) { + return ((f2->fr == NULL) ? 0 : +1); + } else if (f2->fr == NULL) + return (-1); + else { + int ret; + + ret = list_coll(&(f1->si), &(f2->si)); + if (!ret) + return ((f1->file_pos < f2->file_pos) ? -1 : +1); + return (ret); + } + } +} + +/* + * Allocate and init file header structure + */ +static void +file_header_init(struct file_header **fh, const char *fn, size_t file_pos) +{ + + if (fh && fn) { + struct bwstring *line; + + *fh = sort_malloc(sizeof(struct file_header)); + (*fh)->file_pos = file_pos; + (*fh)->fr = file_reader_init(fn); + if ((*fh)->fr == NULL) { + perror(fn); + err(2, "%s", getstr(8)); + } + line = file_reader_readline((*fh)->fr); + if (line == NULL) { + file_reader_free((*fh)->fr); + (*fh)->fr = NULL; + (*fh)->si = NULL; + } else { + (*fh)->si = sort_list_item_alloc(); + sort_list_item_set((*fh)->si, line); + } + } +} + +/* + * Close file + */ +static void +file_header_close(struct file_header **fh) +{ + + if (fh && *fh) { + if ((*fh)->fr) { + file_reader_free((*fh)->fr); + (*fh)->fr = NULL; + } + if ((*fh)->si) { + sort_list_item_clean((*fh)->si); + sort_free((*fh)->si); + (*fh)->si = NULL; + } + sort_free(*fh); + *fh = NULL; + } +} + +/* + * Swap two array elements + */ +static void +file_header_swap(struct file_header **fh, size_t i1, size_t i2) +{ + struct file_header *tmp; + + tmp = fh[i1]; + fh[i1] = fh[i2]; + fh[i2] = tmp; +} + +/* heap algorithm ==>> */ + +/* + * See heap sort algorithm + * "Raises" last element to its right place + */ +static void +file_header_heap_swim(struct file_header **fh, size_t indx) +{ + + if (indx > 0) { + size_t parent_index; + + parent_index = (indx - 1) >> 1; + + if (file_header_cmp(fh[indx], fh[parent_index]) < 0) { + /* swap child and parent and continue */ + file_header_swap(fh, indx, parent_index); + file_header_heap_swim(fh, parent_index); + } + } +} + +/* + * Sink the top element to its correct position + */ +static void +file_header_heap_sink(struct file_header **fh, size_t indx, size_t size) +{ + size_t left_child_index; + size_t right_child_index; + + left_child_index = indx + indx + 1; + right_child_index = left_child_index + 1; + + if (left_child_index < size) { + size_t min_child_index; + + min_child_index = left_child_index; + + if ((right_child_index < size) && + (file_header_cmp(fh[left_child_index], + fh[right_child_index]) > 0)) + min_child_index = right_child_index; + if (file_header_cmp(fh[indx], fh[min_child_index]) > 0) { + file_header_swap(fh, indx, min_child_index); + file_header_heap_sink(fh, min_child_index, size); + } + } +} + +/* <<== heap algorithm */ + +/* + * Adds element to the "left" end + */ +static void +file_header_list_rearrange_from_header(struct file_header **fh, size_t size) +{ + + file_header_heap_sink(fh, 0, size); +} + +/* + * Adds element to the "right" end + */ +static void +file_header_list_push(struct file_header *f, struct file_header **fh, size_t size) +{ + + fh[size++] = f; + file_header_heap_swim(fh, size - 1); +} + +struct last_printed +{ + struct bwstring *str; +}; + +/* + * Prints the current line of the file + */ +static void +file_header_print(struct file_header *fh, FILE *f_out, struct last_printed *lp) +{ + + if (fh && fh->fr && f_out && fh->si && fh->si->str) { + if (sort_opts_vals.uflag) { + if ((lp->str == NULL) || (str_list_coll(lp->str, &(fh->si)))) { + bwsfwrite(fh->si->str, f_out, sort_opts_vals.zflag); + if (lp->str) + bwsfree(lp->str); + lp->str = bwsdup(fh->si->str); + } + } else + bwsfwrite(fh->si->str, f_out, sort_opts_vals.zflag); + } +} + +/* + * Read next line + */ +static void +file_header_read_next(struct file_header *fh) +{ + + if (fh && fh->fr) { + struct bwstring *tmp; + + tmp = file_reader_readline(fh->fr); + if (tmp == NULL) { + file_reader_free(fh->fr); + fh->fr = NULL; + if (fh->si) { + sort_list_item_clean(fh->si); + sort_free(fh->si); + fh->si = NULL; + } + } else { + if (fh->si == NULL) + fh->si = sort_list_item_alloc(); + sort_list_item_set(fh->si, tmp); + } + } +} + +/* + * Merge array of "files headers" + */ +static void +file_headers_merge(size_t fnum, struct file_header **fh, FILE *f_out) +{ + struct last_printed lp; + size_t i; + + memset(&lp, 0, sizeof(lp)); + + /* + * construct the initial sort structure + */ + for (i = 0; i < fnum; i++) + file_header_list_push(fh[i], fh, i); + + while (fh[0]->fr) { /* unfinished files are always in front */ + /* output the smallest line: */ + file_header_print(fh[0], f_out, &lp); + /* read a new line, if possible: */ + file_header_read_next(fh[0]); + /* re-arrange the list: */ + file_header_list_rearrange_from_header(fh, fnum); + } + + if (lp.str) + bwsfree(lp.str); +} + +/* + * Merges the given files into the output file, which can be + * stdout. + */ +static void +merge_files_array(size_t argc, char **argv, const char *fn_out) +{ + + if (argv && fn_out) { + struct file_header **fh; + FILE *f_out; + size_t i; + + f_out = openfile(fn_out, "w"); + + if (f_out == NULL) + err(2, NULL); + + fh = sort_malloc((argc + 1) * sizeof(struct file_header *)); + + for (i = 0; i < argc; i++) + file_header_init(fh + i, argv[i], (size_t) i); + + file_headers_merge(argc, fh, f_out); + + for (i = 0; i < argc; i++) + file_header_close(fh + i); + + sort_free(fh); + + closefile(f_out, fn_out); + } +} + +/* + * Shrinks the file list until its size smaller than max number of opened files + */ +static int +shrink_file_list(struct file_list *fl) +{ + + if ((fl == NULL) || (size_t) (fl->count) < max_open_files) + return (0); + else { + struct file_list new_fl; + size_t indx = 0; + + file_list_init(&new_fl, true); + while (indx < fl->count) { + char *fnew; + size_t num; + + num = fl->count - indx; + fnew = new_tmp_file_name(); + + if ((size_t) num >= max_open_files) + num = max_open_files - 1; + merge_files_array(num, fl->fns + indx, fnew); + if (fl->tmp) { + size_t i; + + for (i = 0; i < num; i++) + unlink(fl->fns[indx + i]); + } + file_list_add(&new_fl, fnew, false); + indx += num; + } + fl->tmp = false; /* already taken care of */ + file_list_clean(fl); + + fl->count = new_fl.count; + fl->fns = new_fl.fns; + fl->sz = new_fl.sz; + fl->tmp = new_fl.tmp; + + return (1); + } +} + +/* + * Merge list of files + */ +void +merge_files(struct file_list *fl, const char *fn_out) +{ + + if (fl && fn_out) { + while (shrink_file_list(fl)); + + merge_files_array(fl->count, fl->fns, fn_out); + } +} + +static const char * +get_sort_method_name(int sm) +{ + + if (sm == SORT_MERGESORT) + return "mergesort"; + else if (sort_opts_vals.sort_method == SORT_RADIXSORT) + return "radixsort"; + else if (sort_opts_vals.sort_method == SORT_HEAPSORT) + return "heapsort"; + else + return "quicksort"; +} + +/* + * Wrapper for qsort + */ +static int sort_qsort(void *list, size_t count, size_t elem_size, + int (*cmp_func)(const void *, const void *)) +{ + + qsort(list, count, elem_size, cmp_func); + return (0); +} + +/* + * Sort list of lines and writes it to the file + */ +void +sort_list_to_file(struct sort_list *list, const char *outfile) +{ + struct sort_mods *sm = &(keys[0].sm); + + if (!(sm->Mflag) && !(sm->Rflag) && !(sm->Vflag) && !(sm->Vflag) && + !(sm->gflag) && !(sm->hflag) && !(sm->nflag)) { + if ((sort_opts_vals.sort_method == SORT_DEFAULT) && byte_sort) + sort_opts_vals.sort_method = SORT_RADIXSORT; + + } else if (sort_opts_vals.sort_method == SORT_RADIXSORT) + err(2, "%s", getstr(9)); + + /* + * to handle stable sort and the unique cases in the + * right order, we need stable basic algorithm + */ + if (sort_opts_vals.sflag) { + switch (sort_opts_vals.sort_method){ + case SORT_MERGESORT: + break; + case SORT_RADIXSORT: + break; + case SORT_DEFAULT: + sort_opts_vals.sort_method = SORT_MERGESORT; + break; + default: + errx(2, "%s", getstr(10)); + } + } + + if (sort_opts_vals.sort_method == SORT_DEFAULT) + sort_opts_vals.sort_method = DEFAULT_SORT_ALGORITHM; + + if (debug_sort) + printf("sort_method=%s\n", + get_sort_method_name(sort_opts_vals.sort_method)); + + switch (sort_opts_vals.sort_method){ + case SORT_RADIXSORT: + rxsort(list->list, list->count); + sort_list_dump(list, outfile); + break; + case SORT_MERGESORT: + mt_sort(list, mergesort, outfile); + break; + case SORT_HEAPSORT: + mt_sort(list, heapsort, outfile); + break; + case SORT_QSORT: + mt_sort(list, sort_qsort, outfile); + break; + default: + mt_sort(list, DEFAULT_SORT_FUNC, outfile); + break; + } +} + +/******************* MT SORT ************************/ + +#if defined(SORT_THREADS) +/* semaphore to count threads */ +#ifndef __APPLE__ +static sem_t mtsem; +#else +static semaphore_t mtsem; +#endif + +/* current system sort function */ +static int (*g_sort_func)(void *, size_t, size_t, + int(*)(const void *, const void *)); + +/* + * Sort cycle thread (in multi-threaded mode) + */ +static void* +mt_sort_thread(void* arg) +{ + struct sort_list *list = arg; + + g_sort_func(list->list, list->count, sizeof(struct sort_list_item *), + (int(*)(const void *, const void *)) list_coll); + +#ifndef __APPLE__ + sem_post(&mtsem); +#else + semaphore_signal(mtsem); +#endif + + return (arg); +} + +/* + * Compare sub-lists. Empty sub-lists always go to the end of the list. + */ +static int +sub_list_cmp(struct sort_list *l1, struct sort_list *l2) +{ + + if (l1 == l2) + return (0); + else { + if (l1->count == 0) { + return ((l2->count == 0) ? 0 : +1); + } else if (l2->count == 0) { + return (-1); + } else { + int ret; + + ret = list_coll(&(l1->list[0]), &(l2->list[0])); + if (!ret) + return ((l1->sub_list_pos < l2->sub_list_pos) ? + -1 : +1); + return (ret); + } + } +} + +/* + * Swap two array elements + */ +static void +sub_list_swap(struct sort_list **sl, size_t i1, size_t i2) +{ + struct sort_list *tmp; + + tmp = sl[i1]; + sl[i1] = sl[i2]; + sl[i2] = tmp; +} + +/* heap algorithm ==>> */ + +/* + * See heap sort algorithm + * "Raises" last element to its right place + */ +static void +sub_list_swim(struct sort_list **sl, size_t indx) +{ + + if (indx > 0) { + size_t parent_index; + + parent_index = (indx - 1) >> 1; + + if (sub_list_cmp(sl[indx], sl[parent_index]) < 0) { + /* swap child and parent and continue */ + sub_list_swap(sl, indx, parent_index); + sub_list_swim(sl, parent_index); + } + } +} + +/* + * Sink the top element to its correct position + */ +static void +sub_list_sink(struct sort_list **sl, size_t indx, size_t size) +{ + size_t left_child_index; + size_t right_child_index; + + left_child_index = indx + indx + 1; + right_child_index = left_child_index + 1; + + if (left_child_index < size) { + size_t min_child_index; + + min_child_index = left_child_index; + + if ((right_child_index < size) && + (sub_list_cmp(sl[left_child_index], + sl[right_child_index]) > 0)) + min_child_index = right_child_index; + if (sub_list_cmp(sl[indx], sl[min_child_index]) > 0) { + sub_list_swap(sl, indx, min_child_index); + sub_list_sink(sl, min_child_index, size); + } + } +} + +/* <<== heap algorithm */ + +/* + * Adds element to the "right" end + */ +static void +sub_list_push(struct sort_list *s, struct sort_list **sl, size_t size) +{ + + sl[size++] = s; + sub_list_swim(sl, size - 1); +} + +struct last_printed_item +{ + struct sort_list_item *item; +}; + +/* + * Prints the current line of the file + */ +static void +sub_list_header_print(struct sort_list *sl, FILE *f_out, + struct last_printed_item *lp) +{ + + if (sl && sl->count && f_out && sl->list[0]->str) { + if (sort_opts_vals.uflag) { + if ((lp->item == NULL) || (list_coll(&(lp->item), + &(sl->list[0])))) { + bwsfwrite(sl->list[0]->str, f_out, + sort_opts_vals.zflag); + lp->item = sl->list[0]; + } + } else + bwsfwrite(sl->list[0]->str, f_out, + sort_opts_vals.zflag); + } +} + +/* + * Read next line + */ +static void +sub_list_next(struct sort_list *sl) +{ + + if (sl && sl->count) { + sl->list += 1; + sl->count -= 1; + } +} + +/* + * Merge sub-lists to a file + */ +static void +merge_sub_lists(struct sort_list **sl, size_t n, FILE* f_out) +{ + struct last_printed_item lp; + size_t i; + + memset(&lp,0,sizeof(lp)); + + /* construct the initial list: */ + for (i = 0; i < n; i++) + sub_list_push(sl[i], sl, i); + + while (sl[0]->count) { /* unfinished lists are always in front */ + /* output the smallest line: */ + sub_list_header_print(sl[0], f_out, &lp); + /* move to a new line, if possible: */ + sub_list_next(sl[0]); + /* re-arrange the list: */ + sub_list_sink(sl, 0, n); + } +} + +/* + * Merge sub-lists to a file + */ +static void +merge_list_parts(struct sort_list **parts, size_t n, const char *fn) +{ + FILE* f_out; + + f_out = openfile(fn,"w"); + + merge_sub_lists(parts, n, f_out); + + closefile(f_out, fn); +} + +#endif /* defined(SORT_THREADS) */ +/* + * Multi-threaded sort algorithm "driver" + */ +static void +mt_sort(struct sort_list *list, + int(*sort_func)(void *, size_t, size_t, int(*)(const void *, const void *)), + const char* fn) +{ +#if defined(SORT_THREADS) + if (nthreads < 2 || list->count < MT_SORT_THRESHOLD) { + size_t nthreads_save = nthreads; + nthreads = 1; +#endif + /* if single thread or small data, do simple sort */ + sort_func(list->list, list->count, + sizeof(struct sort_list_item *), + (int(*)(const void *, const void *)) list_coll); + sort_list_dump(list, fn); +#if defined(SORT_THREADS) + nthreads = nthreads_save; + } else { + /* multi-threaded sort */ + struct sort_list **parts; + size_t avgsize, cstart, i; + + /* array of sub-lists */ + parts = sort_malloc(sizeof(struct sort_list*) * nthreads); + cstart = 0; + avgsize = list->count / nthreads; + + /* set global system sort function */ + g_sort_func = sort_func; + + /* set sublists */ + for (i = 0; i < nthreads; ++i) { + size_t sz = 0; + + parts[i] = sort_malloc(sizeof(struct sort_list)); + parts[i]->list = list->list + cstart; + parts[i]->memsize = 0; + parts[i]->sub_list_pos = i; + + sz = (i == nthreads - 1) ? list->count - cstart : + avgsize; + + parts[i]->count = sz; + + parts[i]->size = parts[i]->count; + + cstart += sz; + } + + /* init threads counting semaphore */ +#ifndef __APPLE__ + sem_init(&mtsem, 0, 0); +#else + { + mach_port_t self = mach_task_self(); + kern_return_t ret = semaphore_create(self, &mtsem, SYNC_POLICY_FIFO, 0); + if (ret != KERN_SUCCESS) { + err(2,NULL); + } + } +#endif + + /* start threads */ + for (i = 0; i < nthreads; ++i) { + pthread_t pth; + pthread_attr_t attr; + + pthread_attr_init(&attr); +#ifndef __APPLE__ + pthread_attr_setdetachstate(&attr, PTHREAD_DETACHED); +#else + pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED); +#endif + + for (;;) { + int res = pthread_create(&pth, &attr, + mt_sort_thread, parts[i]); + + if (res >= 0) + break; + if (errno == EAGAIN) { +#ifndef __APPLE__ + pthread_yield(); +#else + sched_yield(); +#endif + continue; + } + err(2, NULL); + } + + pthread_attr_destroy(&attr); + } + + /* wait for threads completion */ + for (i = 0; i < nthreads; ++i) { +#ifndef __APPLE__ + sem_wait(&mtsem); +#else + semaphore_wait(mtsem); +#endif + } + /* destroy the semaphore - we do not need it anymore */ +#ifndef __APPLE__ + sem_destroy(&mtsem); +#else + { + mach_port_t self = mach_task_self(); + semaphore_destroy(self,mtsem); + } +#endif + + /* merge sorted sub-lists to the file */ + merge_list_parts(parts, nthreads, fn); + + /* free sub-lists data */ + for (i = 0; i < nthreads; ++i) { + sort_free(parts[i]); + } + sort_free(parts); + } +#endif /* defined(SORT_THREADS) */ +} |