sort/common/internal.c

	internal.c revision 7c478bd95313f5f23a4c958a745db2134aa03244
/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only
 * (the "License").  You may not use this file except in compliance
 * with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */
/*
 * Copyright 1998-2003 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

#pragma ident   "%Z%%M% %I% %E% SMI"

#include "internal.h"

#define INSERTION_THRESHOLD 12

static void
swap_range(int i, int j, int n, line_rec_t **I)
{
    while (n-- > 0) {
        line_rec_t *t;

        t = I[i];
        I[i++] = I[j];
        I[j++] = t;
    }
}

/*
 * offset_is_algorithm() implements a simple insertion sort on line records that
 * allows comparison from an offset into the l_collate field (since the subfiles
 * should have already been sorted to that depth).
 */
static void
offset_is_algorithm(line_rec_t **X, ssize_t n,
    int (*collate_fcn)(line_rec_t *, line_rec_t *, ssize_t, flag_t),
    ssize_t depth, flag_t coll_flags)
{
    ssize_t i;

    __S(stats_incr_subfiles());

    /*
     * Place lowest element in X[0].
     */
    for (i = 0; i < n; i++) {
        if (collate_fcn(X[0], X[i], depth, coll_flags) > 0) {
            swap((void **)&X[0], (void **)&X[i]);
            ASSERT(collate_fcn(X[0], X[i], depth, coll_flags) <= 0);
        }
    }

    /*
     * Insertion sort.
     */
    for (i = 2; i < n; i++) {
        ssize_t j = i;
        line_rec_t *t = X[i];
        while (collate_fcn(t, X[j - 1], depth, coll_flags) < 0) {
            X[j] = X[j - 1];
            j--;
            ASSERT(j > 0);
        }
        X[j] = t;
    }
}

/*
 * tqs_algorithm() is called from rqs_algorithm() when a subpartition is
 * encountered whose line records share indistinguishable l_collate fields.  It
 * implements a semi-iterative ternary quicksort.
 */
static void
tqs_algorithm(line_rec_t **X, ssize_t n,
    int (*collate_fcn)(line_rec_t *, line_rec_t *, ssize_t, flag_t),
    flag_t coll_flags)
{
    ssize_t l;          /* boundary of left partition */
    ssize_t le;         /* boundary of left equal partition */
    ssize_t r;          /* boundary of right partition */
    ssize_t re;         /* boundary of right equal partition */

    ssize_t p, q;           /* scratch for indices, comparisons */

    coll_flags |= COLL_DATA_ONLY;

tqs_start:

    /*
     * completion criteria
     */
    if (n <= 1)
        return;

    if (n <= INSERTION_THRESHOLD) {
        offset_is_algorithm(X, n, collate_fcn, 0, coll_flags);
        return;
    }

    /*
     * selection of partition element
     */
    le = rand() % n;
    swap((void **)&X[0], (void **)&X[le]);

    le = l = 1;
    r = re = n - 1;

    for (;;) {
        while (l <= r &&
            (p = collate_fcn(X[l], X[0], 0, coll_flags)) <= 0) {
            if (p == 0)
                swap((void **)&X[le++], (void **)&X[l]);
            l++;
        }

        while (l <= r &&
            (p = collate_fcn(X[r], X[0], 0, coll_flags)) >= 0) {
            if (p == 0)
                swap((void **)&X[r], (void **)&X[re--]);
            r--;
        }

        if (l > r)
            break;

        swap((void **)&X[l++], (void **)&X[r--]);
    }

    /*
     * swap equal partitions into middle
     */
    p = MIN(le, l - le);
    swap_range(0, l - p, p, X);
    p = MIN(re - r, n - re - 1);
    swap_range(l, n - p, p, X);

    /*
     * Iterate with larger subpartition, recurse into smaller.
     */
    p = l - le;
    q = re - r;

    if (p > q) {
        tqs_algorithm(&X[n - q], q, collate_fcn, coll_flags);

        n = p;
    } else {
        tqs_algorithm(X, p, collate_fcn, coll_flags);

        X = &X[n - q];
        n = q;
    }

    goto tqs_start;
    /*NOTREACHED*/
}

/*
 * The following semi-iterative radix quicksort is derived from that presented
 * in
 *  J. Bentley and R. Sedgewick, Fast Algorithms for Sorting and Searching
 *  Strings, in Eighth Annual ACM-SIAM Symposium on Discrete Algorithms,
 *  1997 (SODA 1997),
 * and
 *  R. Sedgewick, Algorithms in C, 3rd ed., vol. 1, Addison-Wesley, 1998.
 */

static void
rqs_algorithm(line_rec_t **X, ssize_t n, ssize_t depth,
    int (*collate_fcn)(line_rec_t *, line_rec_t *, ssize_t, flag_t),
    flag_t coll_flags)
{
    uchar_t v;          /* partition radix value */

    ssize_t l;          /* boundary of left partition */
    ssize_t le;         /* boundary of left equal partition */
    ssize_t r;          /* boundary of right partition */
    ssize_t re;         /* boundary of right equal partition */

    ssize_t p;          /* scratch for indices, comparisons */
    line_rec_t *t;          /* scratch for swaps */

rqs_start:

    /*
     * completion criteria
     */
    if (n <= 1)
        return;

    if (n <= INSERTION_THRESHOLD) {
        offset_is_algorithm(X, n, collate_fcn, depth, coll_flags);
        return;
    }

    /*
     * selection of partition element
     */
    le = rand() % n;
    swap((void **)&X[0], (void **)&X[le]);
    v = X[0]->l_collate.usp[depth];

    le = l = 1;
    r = re = n - 1;

    for (;;) {
        while (l <= r &&
            (p = *(X[l]->l_collate.usp + depth) - v) <= 0) {
            if (p == 0) {
                t = X[le];
                X[le] = X[l];
                X[l] = t;
                le++;
            }
            (l)++;
        }

        while (l <= r &&
            (p = *(X[r]->l_collate.usp + depth) - v) >= 0) {
            if (p == 0) {
                t = X[r];
                X[r] = X[re];
                X[re] = t;
                (re)--;
            }
            (r)--;
        }

        if (l > r)
            break;

        t = X[l];
        X[l] = X[r];
        X[r] = t;
        (l)++;
        (r)--;
    }

    /*
     * swap equal partitions into middle
     */
    p = MIN(le, l - le);
    swap_range(0, l - p, p, X);
    p = MIN(re - r, n - re - 1);
    swap_range(l, n - p, p, X);

    /*
     * recurse into subpartitions as necessary
     */
    p = re - r;
    if (p > 0)
        rqs_algorithm(&X[n - p], p, depth, collate_fcn, coll_flags);

    p = l - le;
    if (p > 0)
        rqs_algorithm(X, p, depth, collate_fcn, coll_flags);

    if (le + n - re - 1 <= 1)
        return;

    /*
     * - 1 so that we don't count the final null.
     */
    if (X[p]->l_collate_length - 1 > depth) {
        /*
         * Equivalent recursion: rqs_algorithm(&X[p], le + n - re - 1,
         * depth + 1, collate_fcn, coll_only);
         */
        X = &X[p];
        n = le + n - re - 1;
        depth++;
        goto rqs_start;
    }

    if (!(coll_flags & COLL_UNIQUE)) {
        __S(stats_incr_tqs_calls());
        tqs_algorithm(&X[p], le + n - re - 1, collate_fcn, coll_flags);
    }
}

static void
radix_quicksort(stream_t *C, flag_t coll_flags)
{
    ASSERT((C->s_status & STREAM_SOURCE_MASK) == STREAM_ARRAY);

    if (C->s_element_size == sizeof (char))
        rqs_algorithm(C->s_type.LA.s_array, C->s_type.LA.s_array_size,
            0, collated, coll_flags);
    else
        rqs_algorithm(C->s_type.LA.s_array, C->s_type.LA.s_array_size,
            0, collated_wide, coll_flags);
}

void
internal_sort(sort_t *S)
{
    size_t input_mem, sort_mem;
    size_t prev_sort_mem = 0;
    void *prev_sort_buf = NULL;

    int numerator, denominator;
    int memory_left;
    int currently_primed;
    flag_t coll_flags;

    stream_t *sort_stream = NULL;
    stream_t *cur_stream;

    set_memory_ratio(S, &numerator, &denominator);
    set_cleanup_chain(&S->m_temporary_streams);

    if (S->m_field_options & FIELD_REVERSE_COMPARISONS)
        coll_flags = COLL_REVERSE;
    else
        coll_flags = 0;

    /*
     * For the entire line special case, we can speed comparisons by
     * recognizing that the collation vector contains all the information
     * required to order the line against other lines of the file.
     * COLL_UNIQUE provides such an exit; if we use the standard put-line
     * operation for the output stream, we achieve the desired fast path.
     */
    if (S->m_entire_line)
        coll_flags |= COLL_UNIQUE;

    hold_file_descriptor();

    cur_stream = S->m_input_streams;
    while (cur_stream != NULL) {
        if (!(cur_stream->s_status & STREAM_OPEN)) {
            if (stream_open_for_read(S, cur_stream) == -1)
                die(EMSG_DESCRIPTORS);
        }

        if (cur_stream->s_status & STREAM_MMAP) {
            input_mem = 0;
        } else {
            input_mem = (size_t)(((u_longlong_t)numerator *
                S->m_memory_available) / denominator);
            stream_set_size(cur_stream, input_mem);
        }

        sort_mem = S->m_memory_available - input_mem;

        memory_left = ST_MEM_AVAIL;

        currently_primed = SOP_PRIME(cur_stream);

        sort_stream = safe_realloc(sort_stream, sizeof (stream_t));
        stream_clear(sort_stream);
        sort_stream->s_buffer = prev_sort_buf;
        sort_stream->s_buffer_size = prev_sort_mem;
        stream_set(sort_stream, STREAM_OPEN | STREAM_ARRAY);
        sort_stream->s_element_size = S->m_single_byte_locale ?
            sizeof (char) : sizeof (wchar_t);
        stream_set_size(sort_stream, sort_mem);
        prev_sort_buf = sort_stream->s_buffer;
        prev_sort_mem = sort_stream->s_buffer_size;

        while (memory_left == ST_MEM_AVAIL) {
            if (currently_primed == PRIME_SUCCEEDED)
                memory_left = stream_insert(S, cur_stream,
                    sort_stream);

            if (SOP_EOS(cur_stream)) {
                if (cur_stream->s_consumer == NULL) {
                    (void) SOP_FREE(cur_stream);
                    (void) SOP_CLOSE(cur_stream);
                }

                cur_stream = cur_stream->s_next;

                if (cur_stream == NULL)
                    break;

                if (!(cur_stream->s_status & STREAM_OPEN) &&
                    (stream_open_for_read(S, cur_stream) == -1))
                    break;

                if (!(cur_stream->s_status & STREAM_MMAP)) {
                    input_mem = numerator *
                        S->m_memory_available / denominator;
                    stream_set_size(cur_stream,
                        input_mem);
                }
                currently_primed = SOP_PRIME(cur_stream);
            }
        }

        radix_quicksort(sort_stream, coll_flags);

#ifndef DEBUG_NO_CACHE_TEMP
        if ((cur_stream == NULL ||
            ((cur_stream->s_status & STREAM_OPEN) &&
            (SOP_EOS(cur_stream) && cur_stream->s_next == NULL)))) {
            (void) stream_push_to_temporary(&S->m_temporary_streams,
                sort_stream, ST_CACHE |
                (S->m_single_byte_locale ? 0 : ST_WIDE));
        } else {
#endif
            release_file_descriptor();
            (void) stream_push_to_temporary(&S->m_temporary_streams,
                sort_stream, ST_NOCACHE |
                (S->m_single_byte_locale ? 0 : ST_WIDE));

            hold_file_descriptor();
#ifdef DEBUG_NO_CACHE_TEMP
            if (cur_stream == NULL)
                break;
#endif
            stream_unset(cur_stream, STREAM_NOT_FREEABLE);
            stream_close_all_previous(cur_stream);
            cur_stream->s_consumer = NULL;
#ifndef DEBUG_NO_CACHE_TEMP
        }
#endif
        if (cur_stream == NULL)
            break;

        if (cur_stream->s_status & STREAM_OPEN &&
            SOP_EOS(cur_stream)) {
            if (cur_stream->s_consumer == NULL) {
                (void) SOP_FREE(cur_stream);
                (void) SOP_CLOSE(cur_stream);
            }

            cur_stream = cur_stream->s_next;
        }
    }

    release_file_descriptor();
}