sump.c

/* sump.c - SUMP Pump(TM) SMP/CMP parallel data pump library.
 *          SUMP Pump is a trademark of Ordinal Technology Corp
 *
 * $Revision: 124 $
 *
 * Copyright (C) 2010 - 2011, Ordinal Technology Corp, http://www.ordinal.com
 * 
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of Version 2 of the GNU General Public
 * License as published by the Free Software Foundation.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 *
 * Linking SUMP Pump statically or dynamically with other modules is
 * making a combined work based on SUMP Pump.  Thus, the terms and
 * conditions of the GNU General Public License v.2 cover the whole
 * combination.
 *
 * In addition, as a special exception, the copyright holders of SUMP Pump
 * give you permission to combine SUMP Pump program with free software
 * programs or libraries that are released under the GNU LGPL and with
 * independent modules that communicate with SUMP Pump solely through
 * Ordinal Technology Corp's Nsort Subroutine Library interface as defined
 * in the Nsort User Guide, http://www.ordinal.com/NsortUserGuide.pdf.
 * You may copy and distribute such a system following the terms of the
 * GNU GPL for SUMP Pump and the licenses of the other code concerned,
 * provided that you include the source code of that other code when and
 * as the GNU GPL requires distribution of source code.
 *
 * Note that people who make modified versions of SUMP Pump are not
 * obligated to grant this special exception for their modified
 * versions; it is their choice whether to do so.  The GNU General
 * Public License gives permission to release a modified version without
 * this exception; this exception also makes it possible to release a
 * modified version which carries forward this exception.
 * 
 * For more information on SUMP Pump, see:
 *     http://www.ordinal.com/sump.html
 *     http://code.google.com/p/sump-pump/
 */

#  define AIO_CAPABLE

#if !defined(_WIN32)
# define _GNU_SOURCE
# include <pthread.h>
# include <unistd.h>
# include <dlfcn.h>
# include <sched.h>
# include <sys/mman.h>
# include <sys/types.h>
# include <sys/wait.h>
# include <sys/time.h>
# include <stdint.h>
# include <ctype.h>
# include <signal.h>

# if !defined(__CYGWIN32__)
#  include <aio.h>
#  include <sys/types.h>
#  include <sys/stat.h>
# endif

# define PTFlld	"lld"
# define PTFllu	"llu"
# define PTFllx	"llx"

# define ERRNO errno

#else  /* now defined(_WIN32) */

# define PTFlld	"I64d"
# define PTFllu	"I64u"
# define PTFllx	"I64x"

# define ERRNO GetLastError()

#endif  /* !defined(_WIN32) */

#include "sump.h"
#include "sumpversion.h"
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>

#include <string.h>
#include <errno.h>

#if defined(SUMP_PUMP_NO_SORT)
/* define some nsort typedefs to minimize the number of #if's in this file */
typedef unsigned nsort_t;       /* nsort context identifier */
typedef int nsort_msg_t;        /* return status & error message numbers */

#else
/* use Nsort include files */
# include "nsort.h"
# include "nsorterrno.h"

#endif

/* values for the flags sump pump structure member
 */
#define SP_UNICODE                      0x0001  /* not yet supported */
#define SP_UTF_8                        0x0002  /* input records are utf-8
                                                 * characters */
#define SP_ASCII   SP_UTF_8
#define SP_FIXED                        0x0003  /* input records are a fixed
                                                 * number of bytes */
#define SP_WHOLE_BUF                    0x0004  /* there are no input records,
                                                 * instead there is only an
                                                 * input buffer */
#define SP_REC_TYPE_MASK                0x0007
#define SP_SORT                         0x0008
#define SP_GROUP_BY                     0x0010
#define SP_EXEC                         0x0020

#define REC_TYPE(sp) ((sp)->flags & SP_REC_TYPE_MASK)

#define TRUE 1
#define FALSE 0

/* sort_state values */
#define SORT_INPUT      1
#define SORT_OUTPUT     2
#define SORT_DONE       3

#define ERROR_BUF_SIZE  500   /* size of error buffer */

#define DEFAULT_BUFFERED_TRANSFER_SIZE  (1024 * 1024)
#define DEFAULT_PIPE_TRANSFER_SIZE      8192


/* state structure for a sump pump instance */
struct sump
{
    unsigned            flags;         /* caller-defined bit flags,
                                        * see sump.h */
    sp_pump_t           pump_func;     /* caller-defined pump function that 
                                        * is executed in parallel by the
                                        * sump pump */
    void                *pump_arg;     /* caller-defined arg to pump func */
    unsigned            num_tasks;     /* number of pump tasks */
    unsigned            num_threads;   /* number of threads executing the
                                        * pump func */
    unsigned            num_in_bufs;   /* number of input buffers */
    unsigned            num_outputs;   /* number of sump pump output channels*/
    ssize_t             in_buf_size;   /* input buffer size in bytes */
    struct sump_out     *out;          /* array of output structures, one for
                                        * each output */
    void                *delimiter;    /* record delimiter, for text input */
    size_t              rec_size;      /* record size */
    pthread_mutex_t     sump_mtx;      /* mutex for sump pump infrastructure */
    pthread_mutex_t     sp_mtx;        /* mutex for pump funcs to use
                                        * via sp_mutex_lock() and
                                        * sp_mutex_unlock() calls */
    pthread_cond_t      in_buf_readable_cond; /* input buffer available for
                                               * reading by pump funcs */
    pthread_cond_t      in_buf_done_cond;  /* an input buffer has been
                                            * completely read by all potential
                                            * pump threads and can be reused */
    pthread_cond_t      task_avail_cond;   /* a task is available for the
                                            * taking by a sump pump thread */
    pthread_cond_t      task_drained_cond; /* a task has been completely 
                                            * executed and its output has
                                            * been drained (read) for its
                                            * output buffer(s) */
    pthread_cond_t      task_output_ready_cond; /* a task's output ready to
                                                 * be read */
    pthread_cond_t      task_output_empty_cond; /* a task's output buffer has
                                                 * been read and is empty */
    size_t              in_buf_current_bytes; /* bytes in current input buf */
    /* number of bytes at end of prev input buffer containing a partial rec */
    size_t              prev_in_buf_ending_rec_partial_bytes;
    pthread_t           *thread;        /* array of sump pump threads */
    uint64_t            cnt_in_buf_readable; /* number of input buffers that
                                              * have been filled with input
                                              * data and are available for
                                              * reading by sump pump threads
                                              * executing pump functions */
    uint64_t            cnt_in_buf_done; /* number of input buffers
                                          * that have been read by all
                                          * their readers */
    uint64_t            cnt_task_init;  /* number of tasks initialized and
                                         * available for the taking by any
                                         * sump pump thread */
    uint64_t            cnt_task_begun; /* number of tasks allocated/taken
                                         * and begun by sump pump threads */
    uint64_t            cnt_task_drained; /* number of tasks that have
                                           * been completed and had all
                                           * their output buffer(s)
                                           * completely read/drained. */
    uint64_t            cnt_task_done; /* number of done tasks whose
                                        * actual ending position has been
                                        * verified to be the same as their
                                        * expected ending */
    struct sp_task      *task;          /* array of sump pump tasks */
    struct in_buf       *in_buf;        /* array of sump pump input buffers */
    nsort_t             nsort_ctx;      /* used only if this is a sort */
    char                *error_buf;     /* buf to hold error msg */
    size_t              error_buf_size; /* buf to hold error msg */
    int                 error_code;     /* pump func generated error code */
    unsigned            sort_error;     /* sort error code */
    char                *sort_temp_buf; /* sort temporary buf */
    size_t              sort_temp_buf_size; /* size of sort temporary buf */
    size_t              sort_temp_buf_bytes; /* bytes of data in temp buf */
    char                input_eof;      /* sp_write_input() called with
                                         * size <= 0 */
    char                broken_input;   /* sp_write_input() called with
                                         * a negative size */
    char                sort_state;     /* only used for sorting */
    char                match_keys;     /* only used for sorting - indicates
                                         * -match has been specified */
    char                wait_done;      /* sp_wait already called for this sp*/
    char                in_file_alloc;  /* in_file string was malloc()'d and
                                         * should be free()'d */
    char                *in_file;       /* input file str or NULL if none */
    struct sp_file      *in_file_sp;    /* input file of sump pump */
    struct exec_state   *ex_state;      /* used when internal pump func invokes
                                         * an external executable program.
                                         * one state per sump pump thread */
    char                **exec_argv;    /* exec process command line */
};

/* struct for an output of a task */
struct task_out
{
    char        *buf;       /* output buffer where map task should
                             * write its results */
    size_t      size;       /* capacity of output buffer */
    size_t      bytes_copied; /* number of bytes written so far into
                               * output buffer */
    char        stalled;  /* the map thread handling this task is
                           * stalled waiting for the writer thread to
                           * empty its full buf */
};


#define INVALID_FD      (-1)
#define PIPE_BUF_SIZE   4096

/* macro to allow the stderr of external programs performing pump functions
 * to be the second output of the sump pump.  The initial implementation of
 * external programs for pump functions did gather stderr as the second
 * output; but it seemed problematic because when an eternal program failed
 * and wrote an error message to its standard error, the sump pump thread
 * that reads the pump func input and writes it to the standard input of
 * external program would notice that external program had terminated
 * abnormally and would set the error code for the sump pump; then the
 * thread reading the stderr of the external program would notice the sump
 * pump error code and exit before it would read the error message in the
 * stderr; thus the error message never made it through sump pump.
 *
#define SUMP_PIPE_STDERR
 */

/* per-pipe struct used when pump funcs call a separate program */
struct std_pipe
{
    struct exec_state   *ex;
    int                 perrno;
#if defined(win_nt)
    HANDLE              rd_h;   /* read handle */
    HANDLE              wr_h;   /* write handle */
#else
    int                 rd_fd;  /* read file descriptor */
    int                 wr_fd;  /* write file descriptor */
#endif
};

/* per-thread struct used when pump funcs call a separate program */
struct exec_state
{
    sp_task_t           t;
    struct std_pipe     in;
    char                in_buf[PIPE_BUF_SIZE];
    struct std_pipe     out;
    char                out_buf[PIPE_BUF_SIZE];
#if defined(SUMP_PIPE_STDERR)
    struct std_pipe     err;
    char                err_buf[PIPE_BUF_SIZE];
#endif
};

/* struct for a sump pump task */
struct sp_task
{
    struct sump *sp;            /* the "host" sp_t of this task */
    uint64_t    task_number;    /* task number */
    int         thread_index;   /* id of thread performing this task */
    char        *in_buf;        /* input buffer */
    size_t      in_buf_bytes;   /* number of bytes written into in_buf
                                 * by the reader thread.  these bytes
                                 * will be read out by map task */
    char        *rec_buf;       /* buf to hold rec returned by pf_get_rec() */
    size_t      rec_buf_size;   /* size of rec_buf */
    char        *curr_rec;      /* pointer to the current record */
    char        *temp_buf;       /* temp buf to help with printf */
    size_t      temp_buf_size;   /* size of the temp buf */
    char        *error_buf;      /* error message posted by sp_error() call */
    size_t      error_buf_size;  /* size of the error message buf */
    int         error_code;      /* pump func generated error code */
    int         sort_error;      /* nsort error code */
    uint64_t    curr_in_buf_index; /* current input buffer index */
    char        *begin_rec;      /* pointer to the beginning record */
    uint64_t    begin_in_buf_index;/* beginning input buffer index */
    /* the following 2 members are set by the thread calling sp_write_input()
     */
    uint64_t    expected_end_index; /* expected end in buf index */
    int         expected_end_offset; /* expected end in buf offset */
    
    char        first_group_rec; /* next record read will be the first
                                  * record for its record group */
    char        first_in_buf;     /* this task is still reading its
                                   * first input buffer */
    char        input_eof;       /* boolean: this task is done reading
                                  * its input */
    char        output_eof;      /* boolean: thread performing task is
                                  * done writing its output to its output
                                  * buffer, but we may still need to wait
                                  * until the output buffer has been read
                                  * before the task is done */
    int         outs_drained;   /* number of outputs for this task that
                                 * have been completely drained (read) */
    struct task_out *out;       /* array of task outputs */
};

/* struct for a sump pump input buffer */
typedef struct in_buf
{
    char        *in_buf;        /* input buffer */
    size_t      in_buf_bytes;   /* number of bytes written into in_buf
                                 * by the reader thread.  these bytes
                                 * will be read out by map task */
    size_t      in_buf_size;    /* size of the in_buf */
    size_t      alloc_size;     /* allocation size of the in_buf */
    unsigned    num_readers;    /* number of threads performing
                                 * tasks that read this buf */
    unsigned    num_readers_done;/* number of reader threads that are
                                  * done with this buffer */
} in_buf_t;

/* struct for a link (copy thread) between an output of one sump pump and
 * the input of another.
 */
struct sp_link
{
    struct sump         *out_sp;        /* sp_t we are reading from */
    unsigned            out_index;      /* output index of read sp_t */
    struct sump         *in_sp;         /* sp_t we are writing to */
    size_t              buf_size;       /* buf size */
    char                *buf;           /* temp buf for transfering data */
    pthread_t           thread;         /* thread executing link_main() */
    int                 error_code;     /* error code */
};

/* struct for a file reader or writer thread */
struct sp_file
{
    char        *fname;         /* file name */
    pthread_t   thread;         /* thread executing either file_reader() or
                                 * file_writer() */
    sp_t        sp;             /* sump pump this file */
    int         mode;           /* file access mode */
#if defined(win_nt)
    HANDLE      fd;             /* file handle */
#else
    int         fd;             /* file descriptor */
#endif
    char        wait_done;      /* sp_wait already called for this sp_file */
    int         out_index;      /* sump pump output index (if relevant) */
    int         aio_count;      /* the max and target number of async i/o's */
    size_t      transfer_size;  /* read or write request size */
    int         error_code;     /* error code */
    int         can_seek;       /* if true, then direct/async-capable file */
    int         is_std;         /* file is either stdin, stdout or stderr */
};

/* file access modes */
#define MODE_UNSPECIFIED    0   /* no access mode has been specified */
#define MODE_BUFFERED       1   /* use standard read() or write() calls */
#define MODE_DIRECT         2   /* direct and asynchronous r/w requests */


/* struct for a sump pump output */
struct sump_out
{
    size_t              buf_size;      /* size of each task's corresponding
                                        * output buffer for this output */
    int                 size_specified; /* boolean: non-zero if out buf size
                                         * has been set */
    double              buf_size_mult;/* factor increase over input buf size */
    size_t              partial_bytes_copied; /* the number of bytes copied
                                               * from the task output buffer
                                               * currently being read from */
    char                file_alloc;     /* file name was malloc()'d and
                                         * should be free()'d */
    char                *file;         /* output file str or NULL if none */
    struct sp_file      *file_sp;      /* output file for this sump pump out */
    uint64_t            cnt_task_drained; /* number of tasks that have
                                           * been completed and their
                                           * output buffer for this
                                           * particular output has been
                                           * completely read */
};


#if defined(AIO_CAPABLE)
/* sump aio struct */ 
struct sump_aio
{
    uint64_t            buf_index;      /* sump pump buffer index */
    size_t              buf_offset;     /* beginning io offset within buffer */
    char                last_buf_io;    /* boolean indicating last io for buf*/
    int64_t             file_offset;    /* file offset */
    size_t              nbytes;         /* request size */
    struct aiocb        aio;
};
#endif


/* global sump pump mutex */
static pthread_mutex_t  Global_lock = PTHREAD_MUTEX_INITIALIZER;

static sp_t             Global_external_sp;
static int              Global_external_count;

/* file descriptor for /dev/zero */
static int Zero_fd;

/* default size for sp_write_input() and sp_read_output() transfers for
 * regression testing of those interfaces.
 */
static size_t Default_rw_test_size;

/* default file access mode */
static int Default_file_mode = MODE_DIRECT;


/* die - quit program due to a fatal sump pump infrastructure error.
 */
static void die(char *fmt, ...)
{
    va_list     ap;

    fprintf(stderr, "sump pump fatal error: ");
    va_start(ap, fmt);
    vfprintf(stderr, fmt, ap);
    va_end(ap);
    exit(1);
}


#define PAGE_SIZE       page_size()

/* page_size - internal routine to get the system page size.
 */
static int page_size()
{
    static int  size;

    if (size == 0)
    {
#if defined(win_nt)
        SYSTEM_INFO si;
    
        GetSystemInfo(&si);
        size = si.dwPageSize;
#else
        size = getpagesize();
#endif
    }
    return (size);
}


/* sp_get_time_us  - return elapsed time in microseconds.
 *
 *      The caller can discard the upper 32 bits *only* when timing intervals
 *      less than ~4000 seconds = 1 hour 11 minutes.
 */
static uint64_t sp_get_time_us(void)
{
    struct timeval      time;
    static uint64_t     begin;

    if (begin == 0)  /* if first time */
    {
        if (gettimeofday(&time, NULL) < 0)
            die("gettimeofday() failure\n");
        begin = (time.tv_sec * (uint64_t)1000 * 1000) + time.tv_usec;
    }
    if (gettimeofday(&time, NULL) < 0)
        die("gettimeofday() failure\n");
    return (time.tv_sec * (uint64_t)1000 * 1000) + time.tv_usec - begin;
}


static FILE    *TraceFp;
#define TRACE if (TraceFp != NULL) trace

/* trace - print a trace message
 */
static void trace(const char *fmt, ...)
{
    va_list     ap;
    uint64_t    diff = sp_get_time_us();
    int	        seconds = (int)(diff / 1000000);
    int         fractions = (int)(diff % 1000000);

    fprintf(TraceFp, "%2d.%06d: ", seconds, fractions);
    
    va_start(ap, fmt);
    vfprintf(TraceFp, fmt, ap);
    va_end(ap);
    fflush(TraceFp);
}


/* sp_get_version - get the subversion version for sump pump
 *
 * Returns: a string containing the subversion version. The string should
 *          NOT be free()'d.
 */
const char *sp_get_version(void)
{
    return (sp_version);
}


/* sp_get_id - get the subversion id keyword substitution for sump pump
 *
 * Returns: a string containing the subversion id keyword. The string should
 *          NOT be free()'d.
 */
const char *sp_get_id(void)
{
    return ("$Id$");
}


/* get_error_msg - place a system error message in the provided buffer
 */
static char *get_error_msg(int error, char *err_buf, size_t err_buf_size)
{
#if defined(win_nt)
    char		*lpMsgBuf;
    char		*eol;

    if (error == 0)  /* if no error specified, get last error */
        error = GetLastError();
    FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | 
                  FORMAT_MESSAGE_FROM_SYSTEM | 
                  FORMAT_MESSAGE_IGNORE_INSERTS,
                  GetModuleHandle(NULL),
                  error,
                  MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
                  (LPTSTR) &lpMsgBuf,
                  0,
                  NULL);
    eol = lpMsgBuf + strlen(lpMsgBuf);
    /* remove trailing \r\n so (%d) appears on the same line */
    if (eol > lpMsgBuf && eol[-1] == '\n' && eol[-2] == '\r')
        eol[-2] = '\0';
	
    _snprintf(err_buf, err_buf_size - 1, "%s (%d)", lpMsgBuf, error);
    LocalFree(lpMsgBuf);
    return (err_buf);
#else
    if (error == 0)  /* if no error specified, get last error */
        error = errno;
    return (strerror_r(error, err_buf, err_buf_size));
#endif    
}


/* start_error - raise an error and set error message during sp_start()
 *               but before the sump pump threads are created.
 */
static int start_error(sp_t sp, const char *fmt, ...)
{
    va_list     ap;
    int         ret;

    if (sp->error_code != 0)          /* if prior error */
        return sp->error_code;        /* ignore this one */
    sp->error_code = SP_START_ERROR;

    va_start(ap, fmt);
    ret = vsnprintf(sp->error_buf, sp->error_buf_size, fmt, ap);
    va_end(ap);
#if defined(win_nt)
    if (ret == -1)  /* non-standard vsnprintf overflow indicator on Windows */
    {
        va_start(ap, fmt);
        ret = _vscprintf(fmt, ap);
        va_end(ap);
    }
#endif
    if ((size_t)ret >= sp->error_buf_size)
    {
        if (sp->error_buf_size != 0)
            free(sp->error_buf);
        sp->error_buf_size = (size_t)ret + 1;
        sp->error_buf = (char *)malloc(sp->error_buf_size);
        va_start(ap, fmt);
        vsnprintf(sp->error_buf, sp->error_buf_size, fmt, ap);
        va_end(ap);
    }
    
    return SP_START_ERROR;
}


/* broadcast_all_conds - internal routine to broadcast all sump pump conditions
 *                       The sump_mtx should already be locked.
 */
static void broadcast_all_conds(sp_t sp)
{
    pthread_cond_broadcast(&sp->in_buf_readable_cond); /*multiple sp threads*/
    pthread_cond_broadcast(&sp->in_buf_done_cond);/* sp_write_input() caller */
    pthread_cond_broadcast(&sp->task_avail_cond);   /* multiple sp threads */
    pthread_cond_broadcast(&sp->task_drained_cond);/* sp_write_input() caller*/
    pthread_cond_broadcast(&sp->task_output_ready_cond); /* mult sp threads */
    pthread_cond_broadcast(&sp->task_output_empty_cond); /* mult sp threads */
}


/* sp_raise_error - raise an error for a sump pump
 */
void sp_raise_error(sp_t sp, int error_code, const char *fmt, ...)
{
    va_list     ap;
    int         ret;

    pthread_mutex_lock(&sp->sump_mtx);
    if (sp->error_code != 0)            /* if prior error */
    {
        pthread_mutex_unlock(&sp->sump_mtx);
        return;                /* ignore this one. let prior error stand */
    }
    sp->error_code = error_code;
    
    va_start(ap, fmt);
    ret = vsnprintf(sp->error_buf, sp->error_buf_size, fmt, ap);
    va_end(ap);
#if defined(win_nt)
    if (ret == -1)  /* non-standard vsnprintf overflow indicator on Windows */
    {
        va_start(ap, fmt);
        ret = _vscprintf(fmt, ap);
        va_end(ap);
    }
#endif
    if ((size_t)ret >= sp->error_buf_size)
    {
        if (sp->error_buf_size != 0)
            free(sp->error_buf);
        sp->error_buf_size = (size_t)ret + 1;
        sp->error_buf = (char *)malloc(sp->error_buf_size);
        va_start(ap, fmt);
        vsnprintf(sp->error_buf, sp->error_buf_size, fmt, ap);
        va_end(ap);
    }
    /* Wake up all possible waiting threads for the sump pump.
     * Some of the below pthread_cond_broadcasts could just be signals.
     * But since error handling isn't a performance critical operation,
     * signals are used instead.
     */
    broadcast_all_conds(sp);
    pthread_mutex_unlock(&sp->sump_mtx);
}


#if defined(win_nt)

# include "sump_win.c"

#else

/* init_zero_fd - initialize Zero_fd, the file descriptor for /dev/zero.
 */
static void init_zero_fd()
{
    pthread_mutex_lock(&Global_lock);
    if (Zero_fd <= 0)
        Zero_fd = open("/dev/zero", O_RDWR);
    pthread_mutex_unlock(&Global_lock);
    return;
}

#endif


#if !defined(SUMP_PUMP_NO_SORT)

/* function pointers to nsort library entry points.  These are 
 * non-NULL if the nsort library is linked in.
 */
nsort_msg_t (*Nsort_define)(const char *def,
                            unsigned options,
                            nsort_error_callback_t *callbacks,
                            nsort_t *ctxp);
nsort_msg_t (*Nsort_release_recs)(void *buf, 
                                  size_t size, 
                                  nsort_t *ctxp);
nsort_msg_t (*Nsort_release_end)(nsort_t *ctxp);
nsort_msg_t (*Nsort_return_recs)(void *buf,
                                 size_t *size, 
                                 nsort_t *ctxp);
nsort_msg_t (*Nsort_end)(nsort_t *ctxp);
const char *(*Nsort_get_stats)(nsort_t *ctxp);
char *(*Nsort_message)(nsort_t *ctxp);
char *(*Nsort_version)(void);

typedef nsort_msg_t (*declare_function_t)(char *name, nsort_compare_t func, void *arg);
typedef nsort_msg_t (*define_t)(const char *, unsigned, nsort_callback_t *, nsort_t *ctxp);
typedef nsort_msg_t (*merge_define_t)(const char *def, unsigned options, nsort_error_callback_t *callbacks, int merge_width, nsort_merge_callback_t *merge_input, nsort_t *ctxp);
typedef nsort_msg_t (*release_recs_t)(void *buf, size_t size, nsort_t *ctxp);
typedef nsort_msg_t (*release_end_t)(nsort_t *ctxp);
typedef nsort_msg_t (*return_recs_t)(void *buf, size_t *size, nsort_t *ctxp);
typedef nsort_msg_t (*print_stats_t)(nsort_t *ctxp, FILE *fp);
typedef const char  *(*get_stats_t)(nsort_t *ctxp);
typedef char        *(*message_t)(nsort_t *ctxp);
typedef nsort_msg_t (*end_t)(nsort_t *ctxp);
typedef char        *(*version_t)(void);


/* get_nsort_syms - internal routine to dynamically link to nsort library
 */
static int get_nsort_syms()
{
# if defined(win_nt)
#  define dlsym(handle, name)   GetProcAddress((handle), (name))
    HANDLE      syms;

    if ((syms = LoadLibrary("libnsort.dll")) == NULL)
        return (-1);
# else
    void        *syms;

    if ((syms = dlopen("libnsort.so", RTLD_GLOBAL | RTLD_LAZY)) == NULL)
        return (-1);
# endif
    if ((Nsort_define = (define_t)dlsym(syms, "nsort_define")) == NULL)
        return (-2);
    if ((Nsort_release_recs = (release_recs_t)dlsym(syms, "nsort_release_recs")) == NULL)
        return (-2);
    if ((Nsort_release_end = (release_end_t)dlsym(syms, "nsort_release_end")) == NULL)
        return (-2);
    if ((Nsort_return_recs = (return_recs_t)dlsym(syms, "nsort_return_recs")) == NULL)
        return (-2);
    if ((Nsort_end = (end_t)dlsym(syms, "nsort_end")) == NULL)
        return (-2);
    if ((Nsort_get_stats = (get_stats_t)dlsym(syms, "nsort_get_stats")) == NULL)
        return (-2);
    if ((Nsort_message = (message_t)dlsym(syms, "nsort_message")) == NULL)
        return (-2);
    if ((Nsort_version = (version_t)dlsym(syms, "nsort_version")) == NULL)
        return (-2);
    return (0);
}


/* link_in_nsort - internal routine to, if not already done, dynamically
 *                 link in the nsort library.
 */
static int link_in_nsort()
{
    int ret;
    
# if !defined(win_nt)
    pthread_mutex_lock(&Global_lock);
# endif
    ret = Nsort_define == NULL ? get_nsort_syms() : 0;
# if !defined(win_nt)
    pthread_mutex_unlock(&Global_lock);
# endif
    return (ret);
}    


/* post_nsort_error - internal routine to post an error received from nsort.
 */
static void post_nsort_error(sp_t sp, unsigned ret)
{
    pthread_mutex_lock(&sp->sump_mtx);
    if (sp->error_code == SP_OK)  /* if no other error yet, this is the one */
    {
        char *msg = (*Nsort_message)(&sp->nsort_ctx);
        
        sp->error_code = SP_SORT_EXEC_ERROR;
        sp->sort_error = ret;
        if (msg == NULL)
            msg = "No Nsort error message";
        strncpy(sp->error_buf, msg, sp->error_buf_size);
        sp->error_buf[sp->error_buf_size - 1] = '\0';  /* handle overflow */
    }
    sp->sort_error = ret;
    sp->sort_state = SORT_DONE;
    pthread_cond_broadcast(&sp->task_output_ready_cond);
    pthread_mutex_unlock(&sp->sump_mtx);
}


#define STAT_DRCTV      " -stat"


/* sp_start_sort - start an nsort instance with a sump pump wrapper so
 *                 that its input or output can be assigned to a file or
 *                 linked to another sump pump.  For instance, the sort
 *                 input or output can be linked to sump pump performing
 *                 record pumping such as "map" on input and "reduce"
 *                 for output.
 * Parameters:
 *      sp -       Pointer to where to return newly allocated sp_t 
 *                 identifier that will be used in as the first
 *                 argument to all subsequent sp_*() calls.
 *      def -      Nsort sort definition string.  Besides the Nsort 
 *                 commands listed in the Nsort User Guide, the following
 *                 directives are also recognized:
 *                   -match[=%d] Each output record will be preceded by a
 *                               single byte that indicates the
 *                               specified number of keys in this record
 *                               are the same as in the previous record.
 *                               If no key number is specified, all keys
 *                               are examined for a match condition.
 *
 * Returns: SP_OK or a sump pump error code
 */
int sp_start_sort(sp_t *caller_sp,
                  char *def_fmt,
                  ...)
{
    int                 ret;
    char                *def_plus = NULL;
    sp_t                sp;
    size_t              def_len;
    char                *def;
    char                thread_drctv[30];
    unsigned char       *p;
# if defined(win_nt)
    SYSTEM_INFO         si;
# endif

    *caller_sp = NULL;  /* assume the worst for now */
    sp = (sp_t)calloc(1, sizeof(struct sump));
    if (sp == NULL)
        return (SP_MEM_ALLOC_ERROR);
    sp->error_buf_size = ERROR_BUF_SIZE;
    sp->error_buf = (char *)calloc(1, sp->error_buf_size);
    if (sp->error_buf == NULL)
        return (SP_MEM_ALLOC_ERROR);
    *caller_sp = sp;  /* allow access to error_buf even if failure */
    /* fill in default parameters */
# if defined(win_nt)
    GetSystemInfo(&si);
    sp->num_threads = si.dwNumberOfProcessors;
# else
    sp->num_threads = sysconf(_SC_NPROCESSORS_ONLN);
# endif
    sprintf(thread_drctv, "-threads=%d ", sp->num_threads);
    if (sp->num_outputs > 32)
        sp->num_outputs = 32;
    sp->num_outputs = 1;
    sp->out = (struct sump_out *)calloc(1, sizeof(struct sump_out));
    sp->delimiter = (void *)"\n";
    sp->rec_size = 0;

    if (link_in_nsort() != 0)    /* if error */
    {
        sp->error_code = SP_NSORT_LINK_FAILURE;
        sp->sort_state = SORT_DONE;
        return (sp->error_code);
    }
    
    sp->flags |= SP_SORT;

    if (def_fmt != NULL)
    {
        va_list ap;
        size_t  def_len;
        
        va_start(ap, def_fmt);
#if defined(win_nt)
        def_len = _vscprintf(def_fmt, ap);
#else
        def_len = vsnprintf(NULL, 0, def_fmt, ap);
#endif
        va_end(ap);
        def = (char *)calloc(def_len + 1, 1);
        va_start(ap, def_fmt);
        if (vsnprintf(def, def_len + 1, def_fmt, ap) != def_len)
            die("sp_start_sort: vnsprintf failed to return %d\n", def_len);
        va_end(ap);
    }
    else
        def = "";

    def_len = strlen(def) + 1;  /* plus '\0' */
    def_len += strlen(STAT_DRCTV);
    def_len += strlen(thread_drctv);
    def_plus = (char *)malloc(def_len);
    if (def_plus == NULL)
        return (SP_MEM_ALLOC_ERROR);
    strcpy(def_plus, thread_drctv); /* goes first so caller can override */
    strcat(def_plus, def);
    strcat(def_plus, STAT_DRCTV);

    /* check for input file declaration */
    for (p = (unsigned char *)def_plus; *p != '\0'; p++)
    {
        if (p[0] == '-' || p[0] == '/')
        {
            p++;
            /* detect -IN[_]F[ILE] */
            if (toupper(p[0]) == 'I' && toupper(p[1]) == 'N' &&
                (toupper(p[2]) == 'F' || p[2] == '_'))
            {
                sp->in_file = "<defined to nsort>";
                p += 3;
            }
            /* detect -OUT[_]F[ILE] */
            else if (toupper(p[0]) == 'O' && toupper(p[1]) == 'U' &&
                     toupper(p[2]) == 'T' &&
                     (toupper(p[3]) == 'F' || p[3] == '_'))
            {
                sp->out[0].file = "<defined to nsort>";
                p += 4;
            }
            /* detect -MATCH */
            else if (toupper(p[0]) == 'M' && toupper(p[1]) == 'A' &&
                     toupper(p[2]) == 'T' && toupper(p[3]) == 'C' &&
                     toupper(p[4]) == 'H')
            {
                sp->match_keys = TRUE;
                p += 5;
            }
        }
    }

    /* check for output file declaration */
    
    ret = (*Nsort_define)(def_plus, 0, NULL, &sp->nsort_ctx);
    free(def_plus);
    if (def_fmt != NULL)
        free(def);
    if (ret < 0)
    {
        char *msg = (*Nsort_message)(&sp->nsort_ctx);
        
        sp->sort_error = ret;
        sp->error_code = SP_SORT_DEF_ERROR;
        if (msg == NULL)
            msg = "No Nsort error message";
        strncpy(sp->error_buf, msg, sp->error_buf_size - 1);
        sp->error_buf[sp->error_buf_size - 1] = '\0';  /* handle overflow */
        sp->sort_state = SORT_DONE;
        return (SP_SORT_DEF_ERROR);
    }
    sp->sort_state = sp->in_file != NULL ? SORT_OUTPUT : SORT_INPUT;
    pthread_mutex_init(&sp->sump_mtx, NULL);
    /* use output ready cond for state changes */
    pthread_cond_init(&sp->task_output_ready_cond, NULL);
    sp->sort_temp_buf_size = sp->out[0].buf_size ? sp->out[0].buf_size : 4096;
    if ((sp->sort_temp_buf = (char *)malloc(sp->sort_temp_buf_size)) == NULL)
        return (SP_MEM_ALLOC_ERROR);
    return (SP_OK);
}


/* sp_get_sort_stats - get a string containing the nsort statistics report
 *                     for an nsort sump pump that has completed.
 *
 * Returns: a string containing the Nsort statistics report. The string should
 *          NOT be free()'d and is valid until the passed sp_t is sp_free()'d.
 */
const char *sp_get_sort_stats(sp_t sp)
{
    const char  *ret;
    
    if (sp->error_code)
        return ("no stats because of nsort error");

    if (!(sp->flags & SP_SORT))
        return (NULL);
    if ((ret = (*Nsort_get_stats)(&sp->nsort_ctx)) == NULL)
        ret = "Nsort_get_stats() failure\n";
    return (ret);
}


/* sp_get_nsort_version - get the subversion version for sump pump
 *
 * Returns: a string containing the Nsort version number. The string should
 *          NOT be free()'d.
 */
const char *sp_get_nsort_version(void)
{
    if (link_in_nsort() != 0)    /* if error */
    {
        return ("Nsort could not be linked in");
    }
    return ((*Nsort_version)());
}

#else

/* sp_start_sort - dummy non-sort version.
 */
int sp_start_sort(sp_t *caller_sp,
                  char *def_fmt,
                  ...)
{
    *caller_sp = NULL;
    return (SP_SORT_NOT_COMPILED);
}


/* sp_get_sort_stats - dummy non-sort version.
 */
const char *sp_get_sort_stats(sp_t sp)
{
    return ("");
}


/* sp_get_nsort_version - dummy non-sort version.
 */
const char *sp_get_nsort_version(void)
{
    return ("this SUMP Pump version has not been compiled for nsort");
}


#endif /* !defined(SUMP_PUMP_NO_SORT) */

/* pipe_reader - thread start function for thread that reads either the
 *               stdout or stderr of an external process, writes the output
 *               to either task output 0 or 1, respectively.
 */
void *pipe_reader(void *arg)
{
    struct sump         *sp;
    struct exec_state   *ex;
    struct std_pipe     *pipe;
    sp_task_t           t;
    char                *buf;
    int                 out_index;
    ssize_t             len;
    ssize_t             total = 0;
    ssize_t             buf_size;
    ssize_t             buf_bytes;
    int                 ret;

    pipe = (struct std_pipe *)arg;
    ex = pipe->ex;
    t = ex->t;
    sp = t->sp;

#if defined(SUMP_PIPE_STDERR)
    /* determine if we are reading stdout or stderr */
    if (pipe == &ex->err)
    {
        buf = ex->err_buf;
        out_index = 1;
    }
    else
#endif
    {
        buf = ex->out_buf;
        out_index = 0;
    }
    TRACE("pipe_reader%d: started\n", t->thread_index);

    /* if we are just writing to a whole input buffer.
     */
    if (sp->flags & SP_WHOLE_BUF)
    {
        /* get output buffer */
        ret = pfunc_get_out_buf(t, out_index,
                                (void **)&buf, (size_t *)&buf_size);
        if (ret != 0)
        {
            pfunc_error(t, "internal error: "
                        "bad ret from sp_get_out_buf: %d\n", ret);
            goto reader_return;
        }
        
        /* read from pipe into output buffer, flushing buffer when necessary */
        buf_bytes = 0;
        for (;;)
        {
#if defined(win_nt)
            DWORD       size;
            
            if (ReadFile(pipe->rd_h, buf + buf_bytes,
                         buf_size - buf_bytes, &size, NULL))
                len = size;     /* success */
            else if (GetLastError() == ERROR_BROKEN_PIPE)
                len = 0;        /* eof */
            else
                len = -1;       /* failure */
#else
            len = read(pipe->rd_fd, buf + buf_bytes, buf_size - buf_bytes);
#endif
            if (len == 0)
                break;
            if (len < 0)
            {
                pipe->perrno = ERRNO;
                buf_bytes = 0;
                break;
            }
            buf_bytes += len;
            if (buf_bytes == buf_size)
            {
                /* commit/flush output bytes */
                ret = pfunc_put_out_buf_bytes(t, out_index, buf_bytes);
                if (ret != 0)
                {
                    pfunc_error(t, "internal error: " 
                                "bad ret from sp_put_out_buf_bytes: %d\n",ret);
                    goto reader_return;
                }
                buf_bytes = 0;
            }
        }
        if (buf_bytes != 0)
        {
            /* commit/flush output bytes */
            ret = pfunc_put_out_buf_bytes(t, out_index, buf_bytes);
            if (ret != 0)
            {
                pfunc_error(t, "internal error: " 
                            "bad ret from sp_put_out_buf_bytes: %d\n",ret);
                goto reader_return;
            }
        }
    }
    else
    {
        /* read from pipe and write to task output */
        for (;;)
        {
#if defined(win_nt)
            DWORD       size;
            
            if (ReadFile(pipe->rd_h, buf, PIPE_BUF_SIZE, &size, NULL))
                len = size;     /* success */
            else if (GetLastError() == ERROR_BROKEN_PIPE)
                len = 0;        /* eof */
            else
                len = -1;       /* failure */
#else
            len = read(pipe->rd_fd, buf, PIPE_BUF_SIZE);
#endif
            TRACE("pipe_reader%d: read %d bytes\n", t->thread_index, (int)len);
            if (len == 0)
                break;
            if (len < 0)
            {
                pipe->perrno = ERRNO;
                break;
            }
            total += len;
            if (pfunc_write(t, out_index, buf, len) != len)
            {
                pfunc_error(t, "internal error: pfunc_write failure\n");
                break;
            }
        }
    }
  reader_return:
#if defined(win_nt)
    CloseHandle(pipe->rd_h);
#else
    pfunc_mutex_lock(t);
    {
        close(pipe->rd_fd);
        pipe->rd_fd = INVALID_FD;
    }
    pfunc_mutex_unlock(t);
#endif
    TRACE("pipe_reader%d: returning\n", t->thread_index);

    return (NULL);
}

/* pfunc_exec - internal pump function to pipe the task input to an external
 *              process, and read back the process stdout and stderr and
 *              write to task outputs 0 and 1 respectively.
 */
int pfunc_exec(sp_task_t t, void *unused)
{
    char                *rec;
    struct exec_state   *ex;
    int                 ti = pfunc_get_thread_index(t);
    struct sump         *sp;
    pthread_t           out_thread;
#if defined(SUMP_PIPE_STDERR)
    pthread_t           err_thread;
#endif
    int                 buf_bytes;
    pid_t               child;
    int                 status;
    char	        errmsg[100];
    /* just_input_files is a half-finished implementation using temporary files
     * rather than pipes to transport input data to the external program */
    int                 just_input_files = 0;
#if defined(win_nt)
    DWORD               wr_size;
    PROCESS_INFORMATION pi;
#else
    int                 tempfds[2];
#endif

    sp = t->sp;
    ex = &sp->ex_state[ti];
    ex->t = t;
    ex->in.perrno = 0;
    ex->out.perrno = 0;
#if defined(SUMP_PIPE_STDERR)
    ex->err.perrno = 0;
#endif

    if (!just_input_files)
    {
#if defined(win_nt)
        SECURITY_ATTRIBUTES     sa;
        STARTUPINFOA            sui;
        
        sa.nLength = sizeof(SECURITY_ATTRIBUTES);
        sa.bInheritHandle = TRUE;
        sa.lpSecurityDescriptor = NULL;

        /* create pipe for stdin of external process */
        if (!CreatePipe(&ex->in.rd_h, &ex->in.wr_h, &sa, 0))
            return (pfunc_error(t, "CreatePipe() error: %s\n",
                                get_error_msg(0, errmsg, sizeof(errmsg))));
        /* make sure the write handle is not inherited */
        if (!SetHandleInformation(ex->in.wr_h, HANDLE_FLAG_INHERIT, 0))
            return (pfunc_error(t, "SetHandleInformation() error: %s\n",
                                get_error_msg(0, errmsg, sizeof(errmsg))));

        /* create pipe for stdout of external process */
        if (!CreatePipe(&ex->out.rd_h, &ex->out.wr_h, &sa, 0))
            return (pfunc_error(t, "CreatePipe() error: %s\n",
                                get_error_msg(0, errmsg, sizeof(errmsg))));
        /* make sure the read handle is not inherited */
        if (!SetHandleInformation(ex->out.rd_h, HANDLE_FLAG_INHERIT, 0))
            return (pfunc_error(t, "SetHandleInformation() error: %s\n",
                                get_error_msg(0, errmsg, sizeof(errmsg))));

        ZeroMemory(&sui, sizeof(STARTUPINFOA));
        sui.cb = sizeof(STARTUPINFO);
        sui.hStdInput = ex->in.rd_h;
        sui.hStdOutput = ex->out.wr_h;
        sui.hStdError = GetStdHandle(STD_ERROR_HANDLE); /* use own handle */
        sui.dwFlags |= STARTF_USESTDHANDLES;

        ZeroMemory(&pi, sizeof(PROCESS_INFORMATION));

        if (!CreateProcessA(NULL, /* NULL process name forces use of cmd line
                                   * where the PATH may be used to find exe */
                            sp->exec_argv[0],/* command line */
                            NULL,            /* process security attributes */
                            NULL,            /* thread security attributes */
                            TRUE,            /* handles are inherited */
                            0,               /* flags */
                            NULL,            /* environment */
                            NULL,            /* current directory */
                            &sui,            /* startup info */
                            &pi))            /* process info */
        {
            return (pfunc_error(t, "CreateProcess() error: %s\n",
                                get_error_msg(0, errmsg, sizeof(errmsg))));
        }
        CloseHandle(pi.hThread);

        /* close the original write handle for external proc's stdout */
        if (!CloseHandle(ex->out.wr_h))
        {
            return (pfunc_error(t, "pipe_reader, CloseHandle failure: %d\n", 
                                GetLastError()));
        }
#else
        /* lock out other pump functions (even in other sump pumps)
         * and their fd opens and closes
         */
        pthread_mutex_lock(&Global_lock);
        if (Global_external_count == 0)
            Global_external_sp = sp;
        else if (Global_external_sp != sp)
        {
            /* there is more than one sump pump simultaneously active
             * invoking an external process. This code is not set up to
             * properly close the write fds (to the stdins of its external
             * processes) for the other sump pump.  For now, declare error
             * and return.
             */
            pthread_mutex_unlock(&Global_lock);
            return (pfunc_error(t, "more than one sump active with external processes\n"));
        }
        /* use curly braces to better illustrate mutex block */
        {       
            if (pipe(tempfds) != 0)
            {
                pthread_mutex_unlock(&Global_lock);
                return (pfunc_error(t, "pipe() error: %s\n", strerror(errno)));
            }
            ex->in.rd_fd = tempfds[0];
            ex->in.wr_fd = tempfds[1];
            if (pipe(tempfds) != 0)
            {
                pthread_mutex_unlock(&Global_lock);
                return (pfunc_error(t, "pipe() error: %s\n", strerror(errno)));
            }
            ex->out.rd_fd = tempfds[0];
            ex->out.wr_fd = tempfds[1];
# if defined(SUMP_PIPE_STDERR)
            if (pipe(tempfds) != 0)
            {
                pthread_mutex_unlock(&Global_lock);
                return (pfunc_error(t, "pipe() error: %s\n", strerror(errno)));
            }
            ex->err.rd_fd = tempfds[0];
            ex->err.wr_fd = tempfds[1];
# endif
            child = fork();
            if (child == -1)
            {
                pthread_mutex_unlock(&Global_lock);
                return (pfunc_error(t, "fork() error: %s\n", strerror(errno)));
            }
            else if (child == 0)
            {
                /* current process is the newly created child */
                int         i;

                if (dup2(ex->in.rd_fd, STDIN_FILENO) == -1)
                {
                    perror("dup2() for stdin");
                    exit(1);
                }
                close(ex->in.rd_fd);
                ex->in.rd_fd = INVALID_FD;

                if (dup2(ex->out.wr_fd, STDOUT_FILENO) == -1)
                {
                    perror("dup2() for stdout");
                    exit(1);
                }
                close(ex->out.wr_fd);
                ex->out.wr_fd = INVALID_FD;
# if defined(SUMP_PIPE_STDERR)
                if (dup2(ex->err.wr_fd, STDERR_FILENO) == -1)
                {
                    perror("dup2() for stderr");
                    exit(1);
                }
                close(ex->err.wr_fd);
                ex->err.wr_fd = INVALID_FD;
# endif
                /* close all sump-pump-end pipe file descriptors.
                 * presumably closing these red
                 */
                for (i = 0; i < sp->num_threads; i++)
                {
                    if (sp->ex_state[i].in.wr_fd != INVALID_FD)
                        close(sp->ex_state[i].in.wr_fd);
                    if (sp->ex_state[i].out.rd_fd != INVALID_FD)
                        close(sp->ex_state[i].out.rd_fd);
# if defined(SUMP_PIPE_STDERR)
                    if (sp->ex_state[i].err.rd_fd != INVALID_FD)
                        close(sp->ex_state[i].err.rd_fd);
# endif
                }

                /* exec program */
                execvp(sp->exec_argv[0], sp->exec_argv);

                /* if this path is reached, then the previous exec failed */
                exit(errno);
            }
        
            /* this is the pfunc thread (and not the child process) */
            /* close the child's ends of the 3 pipes */
            close(ex->in.rd_fd);
            ex->in.rd_fd = INVALID_FD;

            close(ex->out.wr_fd);
            ex->out.wr_fd = INVALID_FD;
# if defined(SUMP_PIPE_STDERR)
            close(ex->err.wr_fd);
            ex->err.wr_fd = INVALID_FD;
# endif
        }    
        Global_external_count++;
        pthread_mutex_unlock(&Global_lock);
#endif

        /* create thread for each of stdout and stderr */
        if (pthread_create(&out_thread, NULL, pipe_reader, &ex->out) != 0)
            return (pfunc_error(t, "pthread_create out error: %d\n", ERRNO));
#if defined(SUMP_PIPE_STDERR)
        if (pthread_create(&err_thread, NULL, pipe_reader, &ex->err) != 0)
            return (pfunc_error(t, "pthread_create err error: %d\n", ERRNO));
#endif
    }
    else
    {
#if defined(win_nt)
#else
        char    fname[50];
        
        sprintf(fname, "sptaskinput%llu",
                (long long unsigned int)t->task_number);
        ex->in.wr_fd = open(fname, O_WRONLY | O_CREAT | O_TRUNC, 0777);
        if (ex->in.wr_fd < 0)
        {
            strerror_r(errno, errmsg, sizeof(errmsg));
            return(pfunc_error(t,
                               "can't open debug input file '%s': %s\n",
                               fname, errmsg));
        }
#endif
    }

    /* if we are just reading a whole input buffer, not record by record.
     */
    if (sp->flags & SP_WHOLE_BUF)
    {
        unsigned char   *in;
        size_t          in_size;
        int             ret;

        /* get input buffer */
        if ((ret = pfunc_get_in_buf(t, (void **)&in, &in_size)) != 0)
            return (pfunc_error(t, "internal pfunc error: "
                                "bad ret from sp_get_in_buf: %d\n", ret));
#if defined(win_nt)
        if (!WriteFile(ex->in.wr_h, in, in_size, &wr_size, NULL))
#else
        if (write(ex->in.wr_fd, in, in_size) != in_size)
#endif
            ex->in.perrno = ERRNO;
    }
    else  /* the records are lines of text */
    {
        buf_bytes = 0;
        for (;;)
        {
            /* for each record in the group */
            while (pfunc_get_rec(t, &rec) > 0)
            {
                int     len;
                int     copy_size;
        
                /* buffer and write to in.wr_fd */
                len = strlen(rec);
                while (len)
                {
                    copy_size = len;
                    if (copy_size > PIPE_BUF_SIZE - buf_bytes)
                        copy_size = PIPE_BUF_SIZE - buf_bytes;
                    memcpy(ex->in_buf + buf_bytes, rec, copy_size);
                    buf_bytes += copy_size;
                    if (buf_bytes == PIPE_BUF_SIZE)
                    {
#if defined(win_nt)
                        if (!WriteFile(ex->in.wr_h, ex->in_buf, buf_bytes,
                                       &wr_size, NULL))
#else
                        if (write(ex->in.wr_fd, ex->in_buf, buf_bytes) !=
                            buf_bytes)
#endif
                        {
                            ex->in.perrno = ERRNO;
                            buf_bytes = 0;
                            break;
                        }
                        buf_bytes = 0;
                        rec += copy_size;
                    }
                    len -= copy_size;
                }
            }
            /* if we are doing a "group by" and we have not finished all the
             * records in the first input buffer, then reset the eof state to
             * continue reading records from the first input buffer.
             */
            if ((sp->flags & SP_GROUP_BY) && t->first_in_buf)
            {
                t->input_eof = FALSE;
                t->first_group_rec = TRUE;
            }
            else
                break;  /* done with first input buffer and maybe more */
        }
        /* flush any remainder */
        if (buf_bytes != 0)
        {
#if defined(win_nt)
            if (!WriteFile(ex->in.wr_h, ex->in_buf, buf_bytes, &wr_size, NULL))
#else
            if (write(ex->in.wr_fd, ex->in_buf, buf_bytes) != buf_bytes)
#endif
                ex->in.perrno = ERRNO;
        }
    }
    
#if defined(win_nt)
    CloseHandle(ex->in.wr_h);
#else
    pthread_mutex_lock(&Global_lock);
    close(ex->in.wr_fd);
    ex->in.wr_fd = INVALID_FD;
    Global_external_count--;
    pthread_mutex_unlock(&Global_lock);
#endif
    
    if (!just_input_files)
    {
#if defined(win_nt)
        DWORD   exitCode;
        DWORD   rv;
        
        /* wait for external process */
        rv = WaitForSingleObject(pi.hProcess, INFINITE);
        if (rv != WAIT_OBJECT_0)
            return (pfunc_error(t, "process WaitForSingleObject() ret: %d, error: %s\n",
                                rv, get_error_msg(0, errmsg, sizeof(errmsg))));
        if (!GetExitCodeProcess(pi.hProcess, &exitCode))
            return (pfunc_error(t, "GetExitCodeProcess() error: %s\n",
                                get_error_msg(0, errmsg, sizeof(errmsg))));
        TRACE("pfunc_exec%d: ext proc exited with: %d\n",
              t->thread_index, exitCode);
        if (exitCode != 0)
        {
            pfunc_error(t,
                        "process '%s' instance %I64u exited with status: 0x%x\n",
                        sp->exec_argv[0], t->task_number, exitCode);
        }
        CloseHandle(pi.hProcess);
#else
        /* if error writing to pipe and is not broken pipe error */
        if (ex->in.perrno != 0 && ex->in.perrno != EPIPE)
        {
            strerror_r(ex->in.perrno, errmsg, sizeof(errmsg));
            TRACE("pfunc_exec%d: stdin write error: %d, %s\n", t->thread_index,
                  ex->in.perrno, errmsg);
            return (pfunc_error(t, "pipe stdin write error: %s\n", errmsg));
        }
    
        /* wait for child process to end */
        if (waitpid(child, &status, 0) == -1)
        {
            strerror_r(errno, errmsg, sizeof(errmsg));
            TRACE("pfunc_exec%d: child wait error: %d, %s\n",
                  t->thread_index, errno, errmsg);
            return (pfunc_error(t, "pipe process wait error: %s\n", errmsg));
        }
        if (WIFSIGNALED(status))
        {
            TRACE("pfunc_exec%d: child terminated with signal %d\n",
                  t->thread_index, WTERMSIG(status));
            return (pfunc_error(t,
                                "external process terminated with signal %d\n",
                                WTERMSIG(status)));

        }
        if (WIFSTOPPED(status))
        {
            TRACE("pfunc_exec%d: child stopped with signal %d\n",
                  t->thread_index, WSTOPSIG(status));
            return (pfunc_error(t, "external process stopped with signal %d\n",
                                WSTOPSIG(status)));
        }
        if (WIFEXITED(status) && WEXITSTATUS(status) != 0)
        {
            TRACE("pfunc_exec%d: child exited with status: %d\n",
                  t->thread_index, WEXITSTATUS(status));
            pfunc_error(t,
                        "process '%s' instance %llu exited with status: %d\n",
                        sp->exec_argv[0], t->task_number, WEXITSTATUS(status));
        }
#endif

        /* wait for stdout thread to end */
        pthread_join(out_thread, NULL);
        TRACE("pfunc_exec%d: out thread exited with errno: %d\n",
              t->thread_index, ex->out.perrno);
        if (ex->out.perrno != 0)
        {
            get_error_msg(ex->out.perrno, errmsg, sizeof(errmsg));
            TRACE("pfunc_exec%d: out thread exited on error: %d, %s\n",
                  t->thread_index, ex->out.perrno, errmsg);
            pfunc_error(t, "pipe stdout read error: %s\n", errmsg);
        }
#if defined(SUMP_PIPE_STDERR)
        /* wait for stderr thread to end */
        pthread_join(err_thread, NULL);
        TRACE("pfunc_exec%d: error thread exited with errno: %d\n",
              t->thread_index, ex->err.perrno);
        if (ex->err.perrno != 0)
        {
            get_error_msg(ex->err.perrno, errmsg, sizeof(errmsg));
            TRACE("pfunc_exec%d: error thread exited on error: %d, %s\n",
                  t->thread_index, ex->err.perrno, errmsg);
            pfunc_error(t, "pipe stderr read error: %s\n", errmsg);
        }
#endif
    }

    return (t->error_code);
}


/* file_reader_test - test routine for a file reader thread using 
 *                            normal read() calls into an intermediate
 *                            buffer followed by writes into the sump pump.
 */
static void *file_reader_test(void *arg)
{
    int                 size;
    char                *read_buf;
    sp_file_t           sp_src = (sp_file_t)arg;
    sp_t                sp = sp_src->sp;
    char                err_buf[200];
    
    TRACE("file_reader_intr: allocating %d buffer bytes\n",
          sp_src->transfer_size);
    read_buf = (char *)malloc(sp_src->transfer_size);
    if (read_buf == NULL)
    {
        sp_raise_error(sp, SP_MEM_ALLOC_ERROR,
                       "%s: buffer malloc failure, size %d\n",
                       sp_src->fname, (int)sp_src->transfer_size);
    }
    
    /* keep looping until there is no additional input */
    for ( ; read_buf != NULL; )
    {
#if defined(win_nt)
        DWORD   rlen;
        
        if (ReadFile(sp_src->fd, read_buf, sp_src->transfer_size, &rlen, NULL))
            size = rlen;        /* success */
        else if (GetLastError() == ERROR_BROKEN_PIPE)
            size = 0;           /* eof */
        else
            size = -1;          /* failure */
#else
        size = read(sp_src->fd, read_buf, (unsigned int)sp_src->transfer_size);
#endif
        if (size < 0)
        {
            sp_raise_error(sp, SP_FILE_READ_ERROR,
                           "%s: read() failure: %s\n",
                           sp_src->fname,
                           get_error_msg(0, err_buf, sizeof(err_buf)));
            break;
        }
        if (sp_write_input(sp, read_buf, size) != size)
            break;      /* silently quit on a downstream error */
        if (size == 0)  /* if we just sent 0 bytes to sp_write_input() */
            break;
    }
    if (read_buf != NULL)
        free(read_buf);
    TRACE("file_reader_test done: %d\n", sp_src->error_code);
    return (NULL);
}


/* file_reader_buffered - main routine for a file reader thread using normal
 *                        read() calls directly into sump pump input buffers.
 */
static void *file_reader_buffered(void *arg)
{
    size_t              size;
    size_t              buf_size;
    size_t              filled_bytes;
    size_t              request;
    uint64_t            index;
    char                *read_buf;
    int                 eof;
    int                 ret;
    sp_file_t           sp_src = (sp_file_t)arg;
    sp_t                sp = sp_src->sp;
    char                err_buf[200];
#if defined(win_nt)
    DWORD               rlen;
#endif

    TRACE("file_reader_buffered starting\n");
            
    /* keep looping until there is no additional input */
    for (index = 0; ; index++)
    {
        if (sp_get_in_buf(sp, index, (void **)&read_buf, &buf_size) != SP_OK)
            break;
        
        for (filled_bytes = 0; filled_bytes < buf_size; filled_bytes += size)
        {
            /* calculate read request size */
            request = buf_size - filled_bytes;
            /* limit request size to 2GB */
            if (request > 0x80000000)
                request = 0x80000000;
            if (sp_src->transfer_size != 0 && request > sp_src->transfer_size)
                request = sp_src->transfer_size;
#if defined(win_nt)
            if (ReadFile(sp_src->fd, read_buf + filled_bytes,
                         (DWORD)request, &rlen, NULL))
                size = rlen;    /* success */
            else if (GetLastError() == ERROR_BROKEN_PIPE)
                size = 0;       /* eof */
            else
                size = -1;      /* failure */
#else
            size = read(sp_src->fd, read_buf + filled_bytes,
                        (unsigned int)request);
#endif
            TRACE("file_reader: read %d bytes\n", (int)size);  
            if (size < 0)
            {
                sp_raise_error(sp, SP_FILE_READ_ERROR,
                               "%s: read() failure: %s\n",
                               sp_src->fname,
                               get_error_msg(0, err_buf, sizeof(err_buf)));
                filled_bytes = 0;
                break;
            }
            if (size == 0)  /* if EOF */
                break;
        }
        eof = (filled_bytes < request);
        if ((ret = sp_put_in_buf_bytes(sp, index, filled_bytes, eof)) != SP_OK)
        {
            TRACE("file_reader: sp_put_in_buf_bytes ret: %d\n", ret);
            break;      /* silently quit on a downstream error */
        }
        if (eof)
            break;
    }
#if defined(win_nt)
    CloseHandle(sp_src->fd);
    sp_src->fd = INVALID_HANDLE_VALUE;
#else
    /*close(sp_src->fd);*/
    sp_src->fd = INVALID_FD;
#endif
    TRACE("file_reader_buffered done: %d\n", sp_src->error_code);
    return (NULL);
}


/* file_writer_buffered - main routine for a file writer thread using normal
 *                        write() calls.
 */
static void *file_writer_buffered(void *arg)
{
    char                *buf;
    ssize_t             size;
    sp_file_t           sp_dst = (sp_file_t)arg;
    sp_t                sp = sp_dst->sp;
    int64_t             file_size = 0;
#if defined(win_nt)
    DWORD               wr_size;
#else
    int                 fd = sp_dst->fd;
#endif      
    int                 out_index = sp_dst->out_index;
    char                err_buf[200];
    
    TRACE("file_writer_buffered: allocating %d buffer bytes\n", sp_dst->transfer_size);
    buf = (char *)malloc(sp_dst->transfer_size);
    if (buf == NULL)
    {
        sp_raise_error(sp, SP_MEM_ALLOC_ERROR,
                       "%s: buffer malloc failure, size %d\n",
                       sp_dst->fname, (int)sp_dst->transfer_size);
    }
        
    for ( ; buf != NULL; )
    {
        size = sp_read_output(sp, out_index, buf, sp_dst->transfer_size);
        if (size <= 0)
        {
            if (size < 0)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
            }
            break;
        }
        TRACE("file_writer: writing %d bytes at offset %"PTFlld"\n",
              size, file_size);  
#if defined(win_nt)
        if (!WriteFile(sp_dst->fd, buf, size, &wr_size, NULL))
#else
        if (write(fd, buf, (unsigned int)size) != size)
#endif
        {
            sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                           "%s: write() failure: %s\n",
                           sp_dst->fname,
                           get_error_msg(0, err_buf, sizeof(err_buf)));
            sp_dst->error_code = SP_FILE_WRITE_ERROR;
            TRACE("output file write error: %s\n", err_buf);
            break;
        }
        file_size += size;
    }
    if (buf != NULL && sp_dst->can_seek)
    {
#if defined(win_nt)
        if (!SetEndOfFile(sp_dst->fd))
#else
        if (ftruncate(sp_dst->fd, file_size))
#endif
        {
            sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                           "%s: truncate failure: %s\n",
                           sp_dst->fname,
                           get_error_msg(0, err_buf, sizeof(err_buf)));
            sp_dst->error_code = SP_FILE_WRITE_ERROR;
            TRACE("truncate error: %s\n", err_buf);
        }
    }
    
#if defined(win_nt)
    CloseHandle(sp_dst->fd);
    sp_dst->fd = INVALID_HANDLE_VALUE;
#else
    /*close(sp_dst->fd);*/
    sp_dst->fd = INVALID_FD;
#endif
    if (buf != NULL)
        free(buf);
    TRACE("file_writer_buffered done: %d\n", sp_dst->error_code);
    return (NULL);
}


#if defined(AIO_CAPABLE)

/* file_reader_direct - main routine for a file reader thread using direct
 *                      aio_read() calls on sump pump input buffers.
 */
static void *file_reader_direct(void *arg)
{
    ssize_t             size;
    ssize_t             request;
    size_t              in_buf_size;
    uint64_t            aios_started;
    int64_t             file_read_offset = 0;
    struct sump_aio     *spaio;
    struct aiocb        *aio;
    const struct aiocb  *cb[1];
    int                 eof;
    sp_file_t           sp_src = (sp_file_t)arg;
    sp_t                sp = sp_src->sp;
    char                err_buf[200];
    int                 aio_count;
    int                 start;
    int                 done;
    uint64_t            next_in_buf;
    size_t              next_buf_offset;
    char                *buf;
    int                 put_result;
#if defined(win_nt)
    int                 i;
#endif

    if (sp_src->aio_count <= 0)
        aio_count = 2;
    else
        aio_count = sp_src->aio_count;
    TRACE("file_reader_direct allocating %d aio structs\n", aio_count);
    spaio = (struct sump_aio *)calloc(sizeof(struct sump_aio), aio_count);
    if (spaio == NULL)
    {
        sp_raise_error(sp, SP_MEM_ALLOC_ERROR,
                       "%s: aio malloc failure, size %d\n",
                       sp_src->fname, (int)aio_count);
        TRACE("file_reader_direct done: %d\n", sp_src->error_code);
        return (NULL);
    }
#if defined(win_nt)
    for (i = 0; i < aio_count; i++)
    {
        spaio[i].aio.sump_over.hEvent = CreateEvent(NULL, 1, 0, NULL);
        if (spaio[i].aio.sump_over.hEvent == NULL)
            die("file_reader_direct, CreateEvent failed\n");
    }
#endif
    
    /* keep looping until there is no additional input */
    next_in_buf = 0;
    next_buf_offset = 0;
    for (aios_started = 0; ; aios_started++)
    {
        start = aios_started % aio_count;
        aio = &spaio[start].aio;
        aio->aio_fildes = sp_src->fd;
        /* if getting of the buffer fails. */
        if (sp_get_in_buf(sp, next_in_buf, (void **)&buf, &in_buf_size) != SP_OK)
        {
            sp_raise_error(sp, SP_FILE_READ_ERROR,
                           "sp_get_in_buf() failure with in_buf %lld\n",
                           aios_started);
            break;
        }
        request = in_buf_size - next_buf_offset;
        if (request > sp_src->transfer_size)
            request = sp_src->transfer_size;
        aio->aio_buf = buf + next_buf_offset;
        aio->aio_nbytes = request;
        aio->aio_offset = file_read_offset;
        spaio[start].buf_index = next_in_buf;
        spaio[start].buf_offset = next_buf_offset;
        spaio[start].file_offset = file_read_offset;
        spaio[start].nbytes = request;
        spaio[start].last_buf_io = (next_buf_offset + request == in_buf_size);
        next_buf_offset += request;
        TRACE("file_reader: reading %d bytes at offset %"PTFlld"\n",
              request, file_read_offset);
        if (next_buf_offset == in_buf_size)
        {
            next_in_buf++;
            next_buf_offset = 0;
        }
        if (aio_read(aio) < 0)
        {
            get_error_msg(aio_error(aio), err_buf, sizeof(err_buf));
            TRACE("file_reader_direct: read failed: %s\n", err_buf);
            sp_src->error_code = SP_FILE_READ_ERROR;
            sp_raise_error(sp, SP_FILE_READ_ERROR,
                           "%s: aio_read() failure: %s, "
                           "offset: %"PTFlld", size: %"PTFlld"\n",
                           sp_src->fname, err_buf,
                           aio->aio_offset, (int64_t)request);
            break;
        }
        file_read_offset += request;
        /* if the max number of aios has not been started, then start another.
         */
        if (aios_started < aio_count - 1)
            continue;

        /* wait for a previously issued request.
         */
        done = (aios_started + 1) % aio_count;
        aio = &spaio[done].aio;
        request = spaio[done].nbytes;
        cb[0] = aio;
        if (aio_suspend(cb, 1, NULL) != 0)
        {
            sp_src->error_code = SP_FILE_READ_ERROR;
            sp_raise_error(sp, SP_FILE_READ_ERROR,
                           "%s: aio_suspend() failure: %s, "
                           "offset: %"PTFlld", size: %"PTFlld"\n",
                           sp_src->fname,
                           get_error_msg(0, err_buf, sizeof(err_buf)),
                           spaio[done].file_offset, (int64_t)request);
            break;
        }
        if ((size = aio_return(aio)) < 0)
        {
            sp_src->error_code = SP_FILE_READ_ERROR;
            sp_raise_error(sp, SP_FILE_READ_ERROR,
                           "%s: aio_return() failure: %s, "
                           "offset: %"PTFlld", size: %"PTFlld"\n",
                           sp_src->fname,
                           get_error_msg(aio_error(aio), err_buf,
                                         sizeof(err_buf)),
                           spaio[done].file_offset, (int64_t)request);
            break;
        }

        eof = (size < request);
        
        /* "put" the input buffer bytes if some bytes have been read into
         * the buffer and either eof or that was the last read for the buffer
         */
        size += spaio[done].buf_offset;
        put_result = SP_OK;
        if (size && (eof || spaio[done].last_buf_io))
        {
            put_result =
                sp_put_in_buf_bytes(sp, spaio[done].buf_index, size, eof);
        }
        if (put_result != SP_OK || eof)
        {
            /* clean up remaining aios and finish up */
            if (aio_count != 1)
            {
                /* wait for and ignore all previously issued aio'ess
                 */
                do
                {
                    done = (done + 1) % aio_count;
                    cb[0] = &spaio[done].aio;
                    aio_suspend(cb, 1, NULL);
                } while (done != start);
            }
            break;
        }
    }
#if defined(win_nt)
    for (i = 0; i < aio_count; i++)
        CloseHandle(spaio[i].aio.sump_over.hEvent);
    CloseHandle(sp_src->fd);
    sp_src->fd = INVALID_HANDLE_VALUE;
#else
    /*close(sp_src->fd);*/
    sp_src->fd = INVALID_FD;
#endif
    free(spaio);
    TRACE("file_reader_direct done: %d\n", sp_src->error_code);
    return (NULL);
}


/* file_writer_direct - main routine for a file writer thread using direct
 *                      aio_write() calls.
 */
static void *file_writer_direct(void *arg)
{
    char                *buf;
    ssize_t             size;
    ssize_t             alloc_size;
    sp_file_t           sp_dst = (sp_file_t)arg;
    sp_t                sp = sp_dst->sp;
    int                 out_index = sp_dst->out_index;
    ssize_t             request;
    uint64_t            aios_started;
    uint64_t            aios_completed;
    int64_t             file_write_offset = 0;
    struct sump_aio     *spaio;
    struct aiocb        *aio;
    const struct aiocb  *cb[1];
    int                 aio_count;
    char                err_buf[200];
    int                 start;
    int                 done;
    int                 eof;
    int                 ret;
    char                *remainder_start = NULL;
    int64_t             remainder_size = 0;
#if defined(win_nt)
    LONG                high;
    LONG                low;
    int                 i;
#endif

    if (sp_dst->aio_count <= 0)
        aio_count = 2;
    else
        aio_count = sp_dst->aio_count;
    spaio = (struct sump_aio *)calloc(sizeof(struct sump_aio), aio_count);
    if (spaio == NULL)
    {
        sp_raise_error(sp, SP_MEM_ALLOC_ERROR,
                       "%s: aio malloc failure, size %d\n",
                       sp_dst->fname, (int)aio_count);
        TRACE("file_writer_direct done: %d\n", sp_dst->error_code);
        return (NULL);
    }

    TRACE("file_writer_direct[%d]: allocating %d buffer bytes\n",
          out_index, sp_dst->transfer_size);
    alloc_size = sp_dst->transfer_size * aio_count;
#if defined(win_nt)
    buf = VirtualAlloc(NULL, alloc_size, MEM_COMMIT, PAGE_READWRITE);
    if (buf == NULL)
    {
        sp_raise_error(sp, SP_MEM_ALLOC_ERROR,
                       "VirtualAlloc() failure: %s\n",
                       get_error_msg(0, err_buf, sizeof(err_buf)));
        sp_dst->error_code = SP_MEM_ALLOC_ERROR;
        return NULL;
    }
    for (i = 0; i < aio_count; i++)
    {
        spaio[i].aio.sump_over.hEvent = CreateEvent(NULL, 1, 0, NULL);
        if (spaio[i].aio.sump_over.hEvent == NULL)
            die("file_writer_direct, CreateEvent failed\n");
    }
#else
    init_zero_fd();
    buf = mmap(NULL, alloc_size, PROT_READ | PROT_WRITE, MAP_PRIVATE, Zero_fd, 0);
    if (buf == MAP_FAILED)
    {
        sp_raise_error(sp, SP_MEM_ALLOC_ERROR,
                       "mmap() failure: %s\n",
                       strerror_r(errno, err_buf, sizeof(err_buf)));
        sp_dst->error_code = SP_MEM_ALLOC_ERROR;
        return NULL;
    }
#endif

    aios_completed = 0;
    for (aios_started = 0; ; )
    {
        start = aios_started % aio_count;
        aio = &spaio[start].aio;
        aio->aio_fildes = sp_dst->fd;
        aio->aio_buf = buf + start * sp_dst->transfer_size;
        request = sp_read_output(sp, out_index, (void *)aio->aio_buf,
                              sp_dst->transfer_size);
        if (request < 0)
        {
            sp_dst->error_code = SP_FILE_WRITE_ERROR;
            break;
        }

        /* if not a write or not a full write, then its eof.
         */
        eof = (request < sp_dst->transfer_size);
        if (eof &&
            (remainder_size = request % PAGE_SIZE) != 0)
        {
            request -= remainder_size;
            remainder_start = (char *)aio->aio_buf + request;
        }
        if (request != 0)
        {
            TRACE("file_writer[%d]: writing %"PTFlld" bytes at offset %"PTFlld"\n",
                  out_index, (int64_t)request, file_write_offset);  
            aio->aio_nbytes = request;
            aio->aio_offset = file_write_offset;
            spaio[start].buf_index = 0;    /* not used */
            spaio[start].buf_offset = 0;   /* not used */
            spaio[start].file_offset = file_write_offset;
            spaio[start].nbytes = request;
            spaio[start].last_buf_io = 0;  /* not used */
            if (aio_write(aio) < 0)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
                sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                               "%s: aio_write() failure: %s, "
                               "offset: %lld, size: %lld\n",
                               sp_dst->fname,
                               get_error_msg(aio_error(&aio[start]), err_buf,
                                             sizeof(err_buf)),
                               spaio[done].file_offset, request);
                break;
            }
            file_write_offset += request;
            aios_started++;
        }
        else if (aios_started == 0)  /* if no direct output whatsoever */
            break;
        
        /* if not eof and the max number of aios has not been started,
         * then start another.
         */
        if (!eof && aios_started < aio_count)
            continue;

        /* wait for a previously issued request.
         * if eof, then wait for all previously issued requests.
         */
        do
        {
            done = aios_completed % aio_count;
            aio = &spaio[done].aio;
            request = spaio[done].nbytes;
            cb[0] = aio;
#if defined(win_nt)
            ret = aio_suspend(cb, 1, NULL);
#else
            while ((ret = aio_suspend(cb, 1, NULL)) != 0 && errno == EINTR)
                continue;
#endif
            if (ret != 0)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
                sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                               "%s: aio_suspend() failure: %s, "
                               "offset: %lld, size: %lld\n",
                               sp_dst->fname,
                               get_error_msg(0, err_buf, sizeof(err_buf)),
                               spaio[done].file_offset, request);
                break;
            }
            if ((size = aio_return(aio)) < 0)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
                sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                               "%s: aio_return() failure: %s, "
                               "offset: %lld, size: %lld\n",
                               sp_dst->fname,
                               get_error_msg(aio_error(aio), err_buf,
                                             sizeof(err_buf)),
                               spaio[done].file_offset, request);
                break;
            }
            if (size != request)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
                sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                               "%s: aio_write() return failure: %s, "
                               "offset: %"PTFlld", "
                               "returned size %"PTFlld
                               " != requested size: %"PTFlld"\n",
                               sp_dst->fname,
                               get_error_msg(aio_error(aio), err_buf,
                                             sizeof(err_buf)),
                               spaio[done].file_offset,
                               (int64_t)size, (int64_t)request);
                break;
            }
            aios_completed++;
        } while (eof && aios_completed < aios_started);
        if (eof || sp_dst->error_code != 0)
            break;
    }
    if (sp_dst->error_code == 0 && remainder_size != 0)
    {
        TRACE("file_writer[%d]: writing remaining %"PTFlld" bytes at offset %"PTFlld"\n",
              out_index, (int64_t)remainder_size, file_write_offset);
#if defined(win_nt)
        CloseHandle(sp_dst->fd);
        sp_dst->fd = CreateFile(sp_dst->fname,
                                GENERIC_WRITE,
                                FILE_SHARE_READ,
                                NULL,
                                OPEN_ALWAYS,
                                FILE_ATTRIBUTE_NORMAL,
                                NULL);
        if (sp_dst->fd == INVALID_HANDLE_VALUE)
        {
            sp_dst->error_code = SP_FILE_WRITE_ERROR;
            sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                           "%s: remainder CreateFile() return failure: %s\n",
                           sp_dst->fname,
                           get_error_msg(0, err_buf, sizeof(err_buf)));
        }
        else
        {
            int err, wr_ret;
            
            aio = &spaio[0].aio;
            aio->sump_over.Offset = (int)file_write_offset;
            aio->sump_over.OffsetHigh = (int)(file_write_offset >> 32);
            ResetEvent(aio->sump_over.hEvent);
            wr_ret = WriteFile(sp_dst->fd, remainder_start, remainder_size,
                               NULL, &aio->sump_over);
            err = GetLastError();
            if (wr_ret == 0 && err != ERROR_IO_PENDING)
                ret = -1;
            else
            {
                wr_ret = GetOverlappedResult(sp_dst->fd, &aio->sump_over,
                                             &ret, TRUE);
                if (wr_ret == 0)
                    ret = -1;
            }

            if (ret != remainder_size)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
                sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                               "%s: pwrite() return failure: %s, "
                               "offset: %I64d, "
                               "returned size %d != requested size: %I64d\n",
                               sp_dst->fname,
                               get_error_msg(0, err_buf, sizeof(err_buf)),
                               file_write_offset, ret, remainder_size);
            }
            else
                file_write_offset += remainder_size;
        }
#else
        /* close file descriptor that was opened with O_DIRECT */
        close(sp_dst->fd);
        /* reopen file without O_DIRECT */
        if ((sp_dst->fd = open(sp_dst->fname, O_WRONLY, 0777)) < 0)
        {
            sp_dst->error_code = SP_FILE_WRITE_ERROR;
            sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                           "%s: remainder open() return failure: %s\n",
                           sp_dst->fname,
                           strerror_r(errno, err_buf, sizeof(err_buf)));
        }
        else
        {
            ret = pwrite(sp_dst->fd, remainder_start,
                         remainder_size, file_write_offset);
            if (ret != remainder_size)
            {
                sp_dst->error_code = SP_FILE_WRITE_ERROR;
                sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                               "%s: pwrite() return failure: %s, "
                               "offset: %lld, "
                               "returned size %d != requested size: %lld\n",
                               sp_dst->fname,
                               strerror_r(errno, err_buf, sizeof(err_buf)),
                               file_write_offset, ret, remainder_size);
            }
            else
                file_write_offset += remainder_size;
        }
#endif
    }
#if defined(win_nt)
    /* truncate output file, since we didn't truncate it when we opened it. */
    high = (LONG)(file_write_offset >> 32);
    low = (LONG)(file_write_offset & 0xFFFFFFFF);
    ret = SetFilePointer(sp_dst->fd, low, &high, FILE_BEGIN);
    if (ret == INVALID_SET_FILE_POINTER && GetLastError() != NO_ERROR)
    {
        sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                       "%s: SetFilePointer() failure: %s\n",
                       sp_dst->fname,
                       get_error_msg(0, err_buf, sizeof(err_buf)));
        sp_dst->error_code = SP_FILE_WRITE_ERROR;
        TRACE("file_writer_direct[%d]: SetFilePointer() error: %s\n",
              out_index, err_buf);
    }
    else if (!SetEndOfFile(sp_dst->fd))
    {
        sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                       "%s: SetEndOfFile() failure: %s\n",
                       sp_dst->fname,
                       get_error_msg(0, err_buf, sizeof(err_buf)));
        sp_dst->error_code = SP_FILE_WRITE_ERROR;
        TRACE("file_writer_direct[%d]: SetEndOfFile() error: %s\n",
              out_index, err_buf);
    }
    else
        TRACE("file_writer_direct[%d]: truncated file to %I64d bytes\n",
              out_index, file_write_offset);

    VirtualFree(buf, alloc_size, MEM_RELEASE);
    for (i = 0; i < aio_count; i++)
        CloseHandle(spaio[i].aio.sump_over.hEvent);
    CloseHandle(sp_dst->fd);
    sp_dst->fd = INVALID_HANDLE_VALUE;
#else
    if (ftruncate(sp_dst->fd, file_write_offset))
    {
        sp_raise_error(sp, SP_FILE_WRITE_ERROR,
                       "%s: ftruncate() failure: %s\n",
                       sp_dst->fname,
                       get_error_msg(0, err_buf, sizeof(err_buf)));
        sp_dst->error_code = SP_FILE_WRITE_ERROR;
        TRACE("file_writer_direct[%d]: ftruncate() error: %s\n",
              out_index, err_buf);
    }
    
    munmap(buf, alloc_size);
    /*close(sp_dst->fd);*/
    sp_dst->fd = INVALID_FD;
#endif
    free(spaio);
    TRACE("file_writer_direct[%d] done: %d\n", out_index, sp_dst->error_code);
    return NULL;
}

#endif


/* sp_wait - can be called by an external thread, e.g. the thread that
 *           called sp_start(), to wait for all sump pump activity to cease.
 *
 * Returns: SP_OK or a sump pump error code
 */
int sp_wait(sp_t sp)
{
    unsigned    i;
    int         ret;

    /* if already waited for this sump pump to complete */
    if (sp->wait_done == TRUE)
        return (sp->error_code);
    if (sp->flags & SP_SORT)
    {
#if !defined(SUMP_PUMP_NO_SORT)
        /* if the sort output is going to a file, and we haven't yet
         * waited for it to complete.
         */
        if (sp->sort_state == SORT_OUTPUT && sp->out[0].file != NULL)
        {
            size_t  zero = 0;
            
            ret = (*Nsort_return_recs)(NULL, &zero, &sp->nsort_ctx);
            if (ret != NSORT_END_OF_OUTPUT)
                post_nsort_error(sp, ret);
            sp->sort_state = SORT_DONE;
        }
#endif
        pthread_mutex_lock(&sp->sump_mtx);
        while (sp->error_code == 0 && sp->sort_state != SORT_DONE)
        {
            TRACE("sp_wait() condition wait");
            pthread_cond_wait(&sp->task_output_ready_cond, &sp->sump_mtx);
        }
        pthread_mutex_unlock(&sp->sump_mtx);
        return (sp->error_code);
    }
    else /* normal (non-sort) sump pump */
    {
        TRACE("waiting for input file reader\n");
        if (sp->in_file_sp != NULL)
        {
            if ((ret = sp_file_wait(sp->in_file_sp)) != SP_OK)
                return (sp->error_code == SP_OK ? ret : sp->error_code);
        }
        for (i = 0; i < sp->num_threads; i++)
        {
            TRACE("waiting for pump thread %d\n", i);
            pthread_join(sp->thread[i], NULL);
        }
        for (i = 0; i < sp->num_outputs; i++)
        {
            if (sp->out[i].file_sp != NULL)
            {
                TRACE("waiting for output %d thread\n", i);
                if ((ret = sp_file_wait(sp->out[i].file_sp)) != SP_OK)
                    return (sp->error_code == SP_OK ? ret : sp->error_code);
            }
        }
    }
    sp->wait_done = TRUE;
    return (sp->error_code);
}


/* syntax error - internal routine to internally record a syntax error
 *                message for possible later retrieval.
 */
static void syntax_error(sp_t sp, char *p, char *err_msg)
{
    int         ret;
    const char  *fmt = "syntax error: %s at: \"%s\"\n";

    if (sp->error_code != 0)          /* if prior error */
        return;                       /* ignore this one */
    sp->error_code = SP_SYNTAX_ERROR;

#if defined(win_nt)
    ret = _snprintf(sp->error_buf, sp->error_buf_size, fmt, err_msg, p);
    if (ret == -1)
        ret = _scprintf(fmt, err_msg, p);
#else
    ret = snprintf(sp->error_buf, sp->error_buf_size, fmt, err_msg, p);
#endif
    if ((size_t)ret >= sp->error_buf_size)
    {
        if (sp->error_buf_size != 0)
            free(sp->error_buf);
        sp->error_buf_size = (size_t)ret + 1;
        sp->error_buf = (char *)malloc(sp->error_buf_size);
#if defined(win_nt)
        _snprintf(sp->error_buf, sp->error_buf_size, fmt, err_msg, p);
#else
        snprintf(sp->error_buf, sp->error_buf_size, fmt, err_msg, p);
#endif
    }
    
    return;
}


/* get_numeric_arg - internal routine to convert an ascii number to int64_t
 */
static int64_t get_numeric_arg(sp_t sp, char **caller_p)
{
    int         negative = 0;
    int64_t     result = 0;
    char        *p = *caller_p;

    if (*p == '-')
    {
        p++;
        negative = 1;
    }
    if (*p < '0' || *p > '9')
    {
        syntax_error(sp, p, "expected numeric argument");
        return 0;
    }
    while (*p >= '0' && *p <= '9')
    {
        result = result * 10 + (*p - '0');
        p++;
    }
    *caller_p = p;
    return (negative ? -result : result);
}


/* get_scale - internal routine to convert 'k', 'm' of 'g' to a numeric value
 */
static int64_t get_scale(char **caller_p)
{
    char        *p = *caller_p;
    int64_t     factor = 1;

    if (*p == 'k' || *p == 'K')
        factor = 1024;
    else if (*p == 'm' || *p == 'M')
        factor = 1024 * 1024;
    else if (*p == 'g' || *p == 'G')
        factor = 1024 * 1024 * 1024;
    else
        return (1);
    *caller_p = p + 1;
    return (factor);
}


/* scan - internal routine to scan the sp_start() string for a keyword.
 */
static int scan(char *kw, char **dp)
{
    unsigned char       *kp;
    unsigned char       *p;

    kp = (unsigned char *)kw;
    p = *(unsigned char **)dp;
    
    /* while we haven't hit the end of the keyword or the end of the
     * definition string, and
     * there is a either a match in the current characters or the keyword
     * string contains an '_' and if we skip over it the chars match.
     */
    while (*kp != '\0' &&
           *p != '\0' &&
           (toupper(*kp) == toupper(*p) ||
            (*kp == '_' && (kp++, toupper(*kp) == toupper(*p)))))
    {
        /* go on to next characters in each string */
        kp++;
        p++;
    }

    /* there was a match if we got to the end of the keyword string and
     * either the last character of the keyword was '=' or the ending
     * definition string character is not alphabetic nor underscore
     */
    if (*kp == '\0' &&
        (*(kp - 1) == '=' ||
         !((toupper(*p) >= 'A' && toupper(*p) <= 'Z') || *p == '_')))
    {
        /* update definition pointer */
        *dp = (char *)p;
        return TRUE;
    }
    else
        return FALSE;
}


/* get_file_mods - internal routine to get file name modifiers,
 *                 e.g. access mode and transfer size.
 */
static void get_file_mods(sp_file_t spf, char *mods)
{
    char        *p = mods;

    for (;;)
    {
        while (*p == ',')
            p++;
        if (*p == '\0')
            break;
        else if (scan("BUFFERED", &p) || scan("BUF", &p))
        {
            spf->mode = MODE_BUFFERED;
        }
        else if (scan("DIRECT", &p) || scan("DIR", &p))
        {
            spf->mode = MODE_DIRECT;
        }
        else if (scan("COUNT", &p) || scan("CO", &p))
        {
            if (*p != ':' && *p != '=')
            {
                syntax_error(spf->sp, p, "expected ':' or '=' after 'count'");
                return;
            }
            p++;
            spf->aio_count = (int)get_numeric_arg(spf->sp, &p);
        }
        else if (scan("TRANSFER", &p) || scan("TRANS", &p) || scan("TR", &p))
        {
            if (*p != ':' && *p != '=')
            {
                syntax_error(spf->sp, p, "expected ':' of '='");
                return;
            }
            p++;
            spf->transfer_size = (size_t)get_numeric_arg(spf->sp, &p);
            spf->transfer_size *= (size_t)get_scale(&p);
        }
        else
        {
            syntax_error(spf->sp, p, "unrecognized file modifier");
            break;
        }
    }
}


/* sp_open_file_src - use the specified file as the input for the
 *                    specified sump pump.
 *
 * Parameters:
 *      sp -          sp_t identifier for which we are opening the input file.
 *      fname_mods -  Name of the file, potentially followed by one or more
 *                    of the following modifiers (with no intervening spaces):
 *                    ,BUFFERED or ,BUF The file will be read with normal
 *                                      buffered (not direct) reads.
 *                    ,DIRECT or ,DIR   The file will be read with direct
 *                                      and asynchronous reads.
 *                    ,TRANSFER=%d{k,m,g} or ,TRANS=%d{k,m,g} or ,TR=%d{k,m,g}
 *                                      The transfer size (read request size)
 *                                      is specified in kilo, mega or giga
 *                                      bytes.
 *                    ,COUNT=%d or ,CO=%d  The count of the maximum number of
 *                                      outstanding asynchronous read requests
 *                                      is given.
 *                    Example:
 *                       myfilename,dir,trans=4m,co=4
 *                                      The above example specifies a file
 *                                      name of "myfilename", with direct
 *                                      and asynchronous reads, with a
 *                                      request size of 4 MB, and a maximum
 *                                      of 4 outstanding asynchronous read
 *                                      requests at any time.
 *
 * Returns: NULL if an error occurs in opening the file, otherwise
 *          a valid sump pump file structure.
 */
sp_file_t sp_open_file_src(sp_t sp, const char *fname_mods)
{
    sp_file_t   sp_src;
    char        *comma_char;
    int         fname_len;
    void        *(*reader_main)(void *);
    int         is_stdin;
    int         specified_mode;

    if ((sp_src = (sp_file_t)calloc(1, sizeof(struct sp_file))) == NULL)
        return (NULL);
    sp_src->sp = sp;

    comma_char = strchr(fname_mods, ',');
    fname_len = (int)
        (comma_char == NULL ? strlen(fname_mods) : comma_char - fname_mods);
    sp_src->fname = (char *)calloc(1, fname_len + 1);
    memcpy(sp_src->fname, fname_mods, fname_len);
    sp_src->fname[fname_len] = '\0';
    if (comma_char != NULL)
        get_file_mods(sp_src, comma_char + 1);
    
    specified_mode = sp_src->mode;
    is_stdin = (strcmp(sp_src->fname, "<stdin>") == 0);
#if defined(win_nt)
    if (is_stdin)
    {
        sp_src->fd = GetStdHandle(STD_INPUT_HANDLE);
        if (sp_src->transfer_size == 0)
            /* there is a read size limit for keyboard input, but is this it?*/
            sp_src->transfer_size = 8192;
        sp_src->mode = MODE_BUFFERED;
    }
    else
    {
        sp_src->fd = CreateFile(sp_src->fname,
                                GENERIC_READ,
                                FILE_SHARE_READ,
                                NULL,
                                OPEN_EXISTING,
                                FILE_ATTRIBUTE_NORMAL,
                                NULL);
        if (sp_src->fd == INVALID_HANDLE_VALUE)
            return (NULL);
        sp_src->can_seek = (GetFileType(sp_src->fd) == FILE_TYPE_DISK);
    }
#else
    if (is_stdin)
    {
        sp_src->fd = 0;
        sp_src->mode = MODE_BUFFERED;
    }        
    else
    {
        struct stat     buf;
        
        sp_src->fd = open(sp_src->fname, 0);
        if (sp_src->fd < 0)
            return (NULL);
        if (fstat(sp_src->fd, &buf) != 0)
            return (NULL);
        sp_src->can_seek = S_ISREG(buf.st_mode);
    }
#endif

    /* if direct mode was specified, but it is not a file */
    if (specified_mode == MODE_DIRECT && !sp_src->can_seek)
    {
        start_error(sp, "direct mode reads were requested for file %s, but it"
                    " is either not a normal file or stdin\n", sp_src->fname);
        return (NULL);
    }

    if (sp_src->mode == MODE_UNSPECIFIED)
        sp_src->mode = sp_src->can_seek ? Default_file_mode : MODE_BUFFERED;

    /* if file mode is direct (whether by specification or default)
     */
    if (sp_src->mode == MODE_DIRECT)
    {
        /* if a transfer size has been specified that is not a multiple of the
         * page size, then silently revert to buffered mode.
         * We probably should instead stay with direct mode, allocate separate
         * read buffers, and copy the data into the sump pump input buffers.
         */
        if (sp_src->transfer_size != 0 &&
            sp_src->transfer_size % PAGE_SIZE != 0)
        {
            sp_src->mode = MODE_BUFFERED;
            start_error(sp, "direct mode reads for file %s,"
                        " but the specified transfer size is not a "
                        "multiple of the page size\n", sp_src->fname);
            return (NULL);
        }
        /* if transfer size has been set and the input buffer size is not a
         * multiple of it, then default to buffered.  This is because the 
         * input buffers are directly read into.
         * We could use vectored reads to eliminate this restriction.
         */
        if (sp_src->transfer_size != 0 &&
            sp->in_buf_size % sp_src->transfer_size != 0)
        {
            sp_src->mode = MODE_BUFFERED;
            start_error(sp, "direct mode reads for file %s, but "
                        "the sump pump input buffer size is not a multiple "
                        "of the specified transfer size\n", sp_src->fname);
            return (NULL);
        }
        /* if the input buffer size is not a multiple of the page size
         */
        if (sp->in_buf_size % PAGE_SIZE != 0)
        {
            /* flag error if direct was specified, otherwise use buffered */
            if (specified_mode == MODE_DIRECT)
            {
                start_error(sp,
                            "direct mode reads were specified for file %s, "
                            "but the sump pump input buffer size is not a "
                            "multiple of the page size\n", sp_src->fname);
                return (NULL);
            }
            sp_src->mode = MODE_BUFFERED;
        }
    }

#if defined(AIO_CAPABLE)
    if (sp_src->can_seek && sp_src->mode == MODE_DIRECT)
    {
        reader_main = file_reader_direct;
        if (sp_src->transfer_size == 0)
            sp_src->transfer_size = 512 * 1024; /* probably should be larger
                                                 * for Windows. */
        if (sp_src->aio_count == 0)
            sp_src->aio_count = 4;

        /* now close file and reopen it as direct */
# if defined(win_nt)
        CloseHandle(sp_src->fd);
        sp_src->fd = CreateFile(sp_src->fname,
                                GENERIC_READ,
                                FILE_SHARE_READ,
                                NULL,
                                OPEN_EXISTING,
                                FILE_FLAG_NO_BUFFERING | FILE_FLAG_OVERLAPPED,
                                NULL);
        if (sp_src->fd == INVALID_HANDLE_VALUE)
            return (NULL);
# else
        close(sp_src->fd);
        sp_src->fd = open(sp_src->fname, O_DIRECT);
        if (sp_src->fd < 0)
            return (NULL);
# endif
    }
    else
#endif
    {
        if (Default_rw_test_size != 0)
        {
            /* test mode for some regression tests */
            reader_main = file_reader_test;
            sp_src->transfer_size = Default_rw_test_size;
        }
        else
            reader_main = file_reader_buffered;
        if (sp_src->transfer_size == 0)
        {
            if (sp_src->can_seek)  /* if normal file */
                sp_src->transfer_size = DEFAULT_BUFFERED_TRANSFER_SIZE;
            else
                sp_src->transfer_size = DEFAULT_PIPE_TRANSFER_SIZE;
        }
    }
    
    /* create reader thread */
    if (pthread_create(&sp_src->thread, NULL, reader_main, sp_src) != 0)
        return (NULL);
    return (sp_src);
}


/* sp_open_file_dst - use the specified file as the output for the
 *                    specified output of the specified sump pump.
 *
 * Parameters:
 *      sp -          sp_t identifier for which we are opening an output file.
 *      out_index -   Integer indicating the sump pump output index.
 *      fname_mods -  Name of the file, potentially followed by one or more
 *                    of the following modifiers (with no intervening spaces):
 *                    ,BUFFERED or ,BUF The file will be written with normal
 *                                      buffered (not direct) writes.
 *                    ,DIRECT or ,DIR   The file will be written with direct
 *                                      and asynchronous writes.
 *                    ,TRANSFER=%d{k,m,g} or ,TRANS=%d{k,m,g} or ,TR=%d{k,m,g}
 *                                      The transfer size (write request size)
 *                                      is specified in kilo, mega or giga
 *                                      bytes.
 *                    ,COUNT=%d or ,CO=%d  The count of the maximum number of
 *                                      outstanding asynchronous write requests
 *                                      is given.
 *                    Example:
 *                       myfilename,dir,trans=4m,co=4
 *                                      The above example specifies a file
 *                                      name of "myfilename", with direct
 *                                      and asynchronous writes, with a
 *                                      request size of 4 MB, and a maximum
 *                                      of 4 outstanding asynchronous write
 *                                      requests at any time.
 *
 * Returns: NULL if an error occurs in opening the file, otherwise
 *          a valid sump pump file structure.
 */
sp_file_t sp_open_file_dst(sp_t sp, unsigned out_index, const char *fname_mods)
{
    sp_file_t           sp_dst;
    int                 ret;
    char                *comma_char;
    int                 fname_len;
    void                *(*writer_main)(void *);
    int                 specified_mode;

    if ((sp_dst = (sp_file_t)calloc(1, sizeof(struct sp_file))) == NULL)
        return (NULL);
    sp_dst->sp = sp;

    comma_char = strchr(fname_mods, ',');
    fname_len = (int)
        (comma_char == NULL ? strlen(fname_mods) : comma_char - fname_mods);
    sp_dst->fname = (char *)calloc(1, fname_len + 1);
    memcpy(sp_dst->fname, fname_mods, fname_len);
    sp_dst->fname[fname_len] = '\0';
    if (comma_char != NULL)
        get_file_mods(sp_dst, comma_char + 1);

    specified_mode = sp_dst->mode;
#if defined(win_nt)
    if (strcmp(sp_dst->fname, "<stdout>") == 0)
        sp_dst->fd = GetStdHandle(STD_OUTPUT_HANDLE);
    else if (strcmp(sp_dst->fname, "<stderr>") == 0)
        sp_dst->fd = GetStdHandle(STD_ERROR_HANDLE);
    else
    {
        sp_dst->fd = CreateFile(sp_dst->fname,
                                GENERIC_WRITE,
                                FILE_SHARE_READ,
                                NULL,
                                OPEN_ALWAYS,
                                FILE_ATTRIBUTE_NORMAL,
                                NULL);
        if (sp_dst->fd == INVALID_HANDLE_VALUE)
            return (NULL);
        sp_dst->can_seek = (GetFileType(sp_dst->fd) == FILE_TYPE_DISK);
    }
#else
    if (strcmp(sp_dst->fname, "<stdout>") == 0)
        sp_dst->fd = 1;
    else if (strcmp(sp_dst->fname, "<stderr>") == 0)
        sp_dst->fd = 2;
    else
    {
        struct stat     buf;
        
        sp_dst->fd = open(sp_dst->fname, O_WRONLY | O_CREAT, 0777);
        if (sp_dst->fd < 0)
            return (NULL);
        if (fstat(sp_dst->fd, &buf) != 0)
            return (NULL);
        sp_dst->can_seek = S_ISREG(buf.st_mode);
    }
#endif

    /* if direct mode was specified but file is not a normal file */
    if (specified_mode == MODE_DIRECT && !sp_dst->can_seek)
    {
        start_error(sp, "direct mode writes were requested for file %s, but it"
                    " is either not a file or stdout/stderr\n", sp_dst->fname);
        return (NULL);
    }

    if (sp_dst->mode == MODE_UNSPECIFIED)
        sp_dst->mode = sp_dst->can_seek ? Default_file_mode : MODE_BUFFERED;

    /* if file mode is direct (whether by specification or default)
     */
    if (sp_dst->mode == MODE_DIRECT)
    {
        /* if a transfer size has been specified that is not a multiple of the
         * page size, then silently revert to buffered mode.
         * We probably should instead stay with direct mode, allocate separate
         * read buffers, and copy the data into the sump pump input buffers.
         */
        if (sp_dst->transfer_size != 0 &&
            sp_dst->transfer_size % PAGE_SIZE != 0)
        {
            sp_dst->mode = MODE_BUFFERED;
            start_error(sp, "direct mode writes for file %s"
                        ", but the transfer size is not a "
                        "multiple of the page size\n", sp_dst->fname);
            return (NULL);
        }
    }

#if defined(AIO_CAPABLE)
    if (sp_dst->can_seek && sp_dst->mode == MODE_DIRECT)
    {
        writer_main = file_writer_direct;
        if (sp_dst->transfer_size == 0)
            sp_dst->transfer_size = 512 * 1024;
        if (sp_dst->aio_count == 0)
            sp_dst->aio_count = 4;

        /* now close file and reopen it as direct */
# if defined(win_nt)
        CloseHandle(sp_dst->fd);
        sp_dst->fd = CreateFile(sp_dst->fname,
                                GENERIC_WRITE,
                                FILE_SHARE_READ,
                                NULL,
                                OPEN_ALWAYS,
                                FILE_FLAG_NO_BUFFERING | FILE_FLAG_OVERLAPPED,
                                NULL);
        if (sp_dst->fd == INVALID_HANDLE_VALUE)
            return (NULL);
# else
        close(sp_dst->fd);
        sp_dst->fd = open(sp_dst->fname, O_DIRECT | O_WRONLY | O_CREAT, 0777);
        if (sp_dst->fd < 0)
            return (NULL);
# endif
    }
    else
#endif
    {
        writer_main = file_writer_buffered;
        if (sp_dst->transfer_size == 0)
        {
            if (Default_rw_test_size != 0)
                sp_dst->transfer_size = Default_rw_test_size;
            else if (sp_dst->can_seek)  /* if normal file */
                sp_dst->transfer_size = DEFAULT_BUFFERED_TRANSFER_SIZE;
            else
                sp_dst->transfer_size = DEFAULT_PIPE_TRANSFER_SIZE;
        }
    }
    
    sp_dst->out_index = out_index;

    /* create writer thread */
    if ((ret = pthread_create(&sp_dst->thread, NULL, writer_main, sp_dst)))
        die("sp_open_file_dst: pthread_create() ret: %d\n", ret);
    return (sp_dst);
}


/* sp_file_wait - wait for the specified file connection to complete.
 *
 * Returns: SP_OK or a sump pump error code
 */
int sp_file_wait(sp_file_t sp_file)
{
    if (sp_file->wait_done == TRUE)
        return (sp_file->error_code);
    pthread_join(sp_file->thread, NULL);
    sp_file->wait_done = TRUE;
    return (sp_file->error_code);
}


/* check_task_done - internal routine to make sure there is room for at
 *                   least one new task.  
 *                   Caller must have locked sump_mtx.
 */
static void check_task_done(sp_t sp)
{
    sp_task_t   t;

    TRACE("check_task_done() called\n");

    /* while there are tasks which have not yet been recognized as done.
     */
    if (sp->cnt_task_init > sp->cnt_task_done)
    {
        /* while there is no room for a new task
         */
        while (sp->error_code == 0 &&
               (sp->cnt_task_init >
                sp->cnt_task_drained + sp->num_tasks - 1))
        {
            TRACE("check_task_done() condition wait for task %d\n",
                  sp->cnt_task_drained);
            pthread_cond_wait(&sp->task_drained_cond, &sp->sump_mtx);
        }
        if (sp->error_code != 0)
        {
            TRACE("check_task_done() returning because of error_code: %d\n",
                  sp->error_code);
            return;
        }

        /* Verify the actual ending position of each done task
         * matches its expected ending position.
         */
        while (sp->cnt_task_drained > sp->cnt_task_done)
        {
            t = &sp->task[sp->cnt_task_done % sp->num_tasks];
            TRACE("check_task_done() task %d verify\n",
                  sp->cnt_task_done);
            if (t->curr_in_buf_index != t->expected_end_index ||
                (t->curr_rec - t->in_buf) != t->expected_end_offset)
            {
                die("task %d input ending mismatch: "
                    "ci %d, ei %d, ao %d, eo %d\n",
                    sp->cnt_task_done,
                    t->curr_in_buf_index,
                    t->expected_end_index,
                    (t->curr_rec - t->in_buf),
                    t->expected_end_offset);
            }
            sp->cnt_task_done++;
        }
    }
    TRACE("check_task_done() returning\n");
}


/* init_new_task - internal routine called by sp_write_input() to
 *                 initialize a new sump pump task.
 */
static sp_task_t init_new_task(sp_t sp, in_buf_t *ib, char *curr_rec)
{            
    sp_task_t           t;
    unsigned            i;

    t = &sp->task[sp->cnt_task_init % sp->num_tasks];
    t->task_number = sp->cnt_task_init;
    /* record starting ib and offset */
    t->in_buf = ib->in_buf;
    /* if the size is 0, this empty task indicates EOF */
    t->in_buf_bytes = ib->in_buf_bytes;
    t->curr_rec = curr_rec;
    t->begin_rec = t->curr_rec;
    t->curr_in_buf_index = sp->cnt_in_buf_readable - 1;
    t->begin_in_buf_index = t->curr_in_buf_index;
    t->expected_end_index = -1;
    t->expected_end_offset = -1;
    t->first_in_buf = TRUE;
    t->outs_drained = 0;
    for (i = 0; i < sp->num_outputs; i++)
    {
        t->out[i].bytes_copied = 0;
        t->out[i].stalled = FALSE;
    }
    t->input_eof = FALSE;
    t->output_eof = FALSE;
    sp->cnt_task_init++;
    return (t);
}


/* new_in_buf - wait, if necessary, until an in_buf is available to be filled
 *              with input data.
 */
static void new_in_buf(sp_t sp)
{
    in_buf_t            *ib = NULL;

    TRACE("new_in_buf: waiting for buffer\n"); 
    pthread_mutex_lock(&sp->sump_mtx);
    while (sp->error_code == 0 &&
           sp->cnt_in_buf_readable >= sp->cnt_in_buf_done + sp->num_in_bufs)
    {
        /* get oldest buffer not yet recognized as done */
        ib = &sp->in_buf[sp->cnt_in_buf_done % sp->num_in_bufs];
        /* if all readers of this input buffer are done reading */
        if (ib->num_readers == ib->num_readers_done)
        {
            sp->cnt_in_buf_done++;
            break;
        }
        pthread_cond_wait(&sp->in_buf_done_cond, &sp->sump_mtx);
    }
    pthread_mutex_unlock(&sp->sump_mtx);
}


/* eof_without_new_in_buf_or_task - clean eof, no need to flush an input
 *                                  buffer or start a new task.
 */
static void eof_without_new_in_buf_or_task(sp_t sp)
{
    pthread_mutex_lock(&sp->sump_mtx);
    sp->input_eof = TRUE;
    /* wake sump thread waiting for next input buffer (just signal OK?) */
    pthread_cond_broadcast(&sp->in_buf_readable_cond);
    /* wake all sump threads waiting for new task */
    pthread_cond_broadcast(&sp->task_avail_cond);
    /* wake writer thread as it should exit on EOF */
    pthread_cond_broadcast(&sp->task_output_ready_cond); 
    pthread_mutex_unlock(&sp->sump_mtx);
}


/* flush_in_buf - flush an input buffer and start a new task if necessary
 */
static void flush_in_buf(sp_t sp, size_t buf_bytes, int eof)
{
    in_buf_t            *ib;
    sp_task_t           t;
    char                *curr_rec;
    char                *p;
        
    /* get ready to release in_buf to pump threads executing pump funcs */
    /* initially, no readers (there will be at least one) */
    ib = &sp->in_buf[sp->cnt_in_buf_readable % sp->num_in_bufs];
    ib->num_readers = 0;     
    ib->num_readers_done = 0;
    curr_rec = ib->in_buf;
    ib->in_buf_bytes = buf_bytes;
    sp->in_buf_current_bytes = 0;

    /* if this is not the first buffer and we are not processing
     * whole buffers.
     */
    if (sp->cnt_in_buf_readable != 0 && REC_TYPE(sp) != SP_WHOLE_BUF)
    {
        /* if we are grouping record by key values
         */
        if (sp->flags & SP_GROUP_BY)
        {
            /* the pump thread performing the most previously
             * issued task will always read this input buffer in
             * order to find the end of its input
             */
            ib->num_readers = 1;

            switch (REC_TYPE(sp))
            {
              case SP_UTF_8:
                /* scan until either we find a record whose key
                 * difference index is less than the number of
                 * "group by" keys, or we scan to the end of the buffer.
                 */
                while (curr_rec < ib->in_buf + ib->in_buf_bytes)
                {
                    /* if this is not the beginning of the buffer or
                     * the buffer begins with a whole record.
                     */
                    if (curr_rec != ib->in_buf ||
                        sp->prev_in_buf_ending_rec_partial_bytes == 0)
                    {
                        /* if match character indicates a new key
                         * grouping, then stop as this the boundry
                         * point between tasks.
                         */
                        if (*curr_rec == '0')
                            break;
                    }
                    /* find next instance of newline in buffer, if any */
                    curr_rec = memchr(curr_rec, *(char *)sp->delimiter,
                                      ib->in_buf_bytes -
                                      (curr_rec - ib->in_buf));
                    if (curr_rec == NULL)
                    {
                        curr_rec = ib->in_buf + ib->in_buf_bytes;
                        break;
                    }
                    curr_rec++;     /* step over newline */
                }
                break;

              case SP_FIXED:
                if (sp->prev_in_buf_ending_rec_partial_bytes == 0)
                    curr_rec = ib->in_buf;
                else
                    curr_rec = ib->in_buf + (sp->rec_size + 1) -
                        sp->prev_in_buf_ending_rec_partial_bytes;
                while (curr_rec < ib->in_buf + ib->in_buf_bytes)
                {
                    if (*curr_rec == '0')
                        break;
                    curr_rec += sp->rec_size + 1;
                }
                break;
            }
        }
        else   /* we are not grouping records by key value */
        {
            /* if this in buffer starts with a partial record, then
             * the pump thread performing the previous issued task
             * will read this buffer in order to find the end of its
             * input
             */
            if (sp->prev_in_buf_ending_rec_partial_bytes != 0)
            {
                /* the previously issued task is required to read the
                 * record reamainder at the beginning of this buffer.
                 */
                ib->num_readers = 1;

                switch (REC_TYPE(sp))
                {
                  case SP_UTF_8:
                    /* find next instance of newline in buffer, if any */
                    curr_rec = memchr(curr_rec, *(char *)sp->delimiter,
                                      ib->in_buf_bytes -
                                      (curr_rec - ib->in_buf));
                    if (curr_rec == NULL)
                        curr_rec = ib->in_buf + ib->in_buf_bytes;
                    else
                        curr_rec++;     /* step over newline */
                    break;

                  case SP_FIXED:
                    curr_rec = ib->in_buf + (sp->rec_size -
                                             sp->prev_in_buf_ending_rec_partial_bytes);
                    break;
                }            
            }
        }
    }

    switch (REC_TYPE(sp))
    {
      case SP_UTF_8:
        /* determine how many bytes the are in any partial record at
         * the end of this buffer.  search backwards to find last newline
         * character.
         */
        for (p = ib->in_buf + ib->in_buf_bytes - 1; p >= ib->in_buf; p--)
            if (*p == *(char *)sp->delimiter) 
                break;
        if (p < ib->in_buf)      /* if there was no newline */
        {
            /* add entire buffer size to this partial record */
            sp->prev_in_buf_ending_rec_partial_bytes += ib->in_buf_bytes;
        }
        else
            sp->prev_in_buf_ending_rec_partial_bytes =
                (ib->in_buf + ib->in_buf_bytes - 1) - p;
        break;

      case SP_FIXED:
        sp->prev_in_buf_ending_rec_partial_bytes =
            (sp->prev_in_buf_ending_rec_partial_bytes + ib->in_buf_bytes) %
            (sp->rec_size + ((sp->flags & SP_GROUP_BY) ? 1 : 0));
        break;

      case SP_WHOLE_BUF:
        sp->prev_in_buf_ending_rec_partial_bytes = 0;
        curr_rec = ib->in_buf;
        break;
    }
        
    TRACE("flush_in_buf: ib %d readable with %d bytes\n",
          sp->cnt_in_buf_readable, ib->in_buf_bytes);
    pthread_mutex_lock(&sp->sump_mtx);
    /* make input buffer available to any existing task.
     * in theory there should be only one task waiting */
    sp->cnt_in_buf_readable++;
    pthread_cond_broadcast(&sp->in_buf_readable_cond);

    if (eof)
    {
        /* if we found the starting point for a new task, then add a
         * reader for the task that will start its input with this
         * input buffer.
         */
        if (curr_rec < ib->in_buf + ib->in_buf_bytes)
            ib->num_readers++;

        /* set the expected end point for the previous task as the
         * actual begin point for the task we are about to define.
         */
        if (sp->cnt_task_init != 0) /* if there was a previous task */
        {
            t = &sp->task[(sp->cnt_task_init - 1) % sp->num_tasks];

            /* if we just made availible a non-empty in_buf
             * that did NOT start a new task.
             */
            if (ib->in_buf_bytes != 0 &&
                curr_rec >= ib->in_buf + ib->in_buf_bytes)
            {
                /* the expected ending in_buf is the next (and empty)
                 * one to be issued.
                 */
                t->expected_end_index = sp->cnt_in_buf_readable;
                t->expected_end_offset = 0;
                pthread_mutex_unlock(&sp->sump_mtx);
                /* do not issue a new task here. */
                eof_without_new_in_buf_or_task(sp);
                return;
            }
            else
            {
                /* the expected ending in_buf is the one just issued
                 * and the offset is the end.
                 */
                t->expected_end_index = sp->cnt_in_buf_readable - 1;
                t->expected_end_offset = (int)(curr_rec - ib->in_buf);
            }

            /* Note: it is possible that at this point the previous
             * task has already completed.  This is why tasks should
             * not use their expected ending point to confirm that
             * they ended at the right point.  This thread should
             * perform the check after the task has completed. */
        }

        /* make sure there is at least one available task struct */
        check_task_done(sp);
        if (sp->error_code != 0)
        {
            pthread_mutex_unlock(&sp->sump_mtx);
            return;
        }

        TRACE("flush_in_buf: initializing task %d\n", sp->cnt_task_init);
        t = init_new_task(sp, ib, curr_rec);
        sp->input_eof = TRUE;
        /* wake all sump threads */
        pthread_cond_broadcast(&sp->task_avail_cond);
        /* wake writer thread as it should exit on EOF */
        pthread_cond_broadcast(&sp->task_output_ready_cond); 
        pthread_mutex_unlock(&sp->sump_mtx);
        return;
    }

    /* if we found the ending point for a task (or this was the
     * very first input read), then start a new task.
     */
    if (curr_rec < ib->in_buf + ib->in_buf_bytes)
    {
        /* add a reader for the task that will start its input with
         * this input buffer.
         */
        ib->num_readers++;

        /* set the expected end point for the previous task as the
         * actual begin point for the task we are about to define.
         */
        if (sp->cnt_task_init != 0) /* if there was a previous task */
        {
            t = &sp->task[(sp->cnt_task_init - 1) % sp->num_tasks];

            /* the expected ending in_buf is the one just issued
             * and the offset is.
             */
            t->expected_end_index = sp->cnt_in_buf_readable - 1;
            t->expected_end_offset = (int)(curr_rec - ib->in_buf);

            /* Note: it is possible that at this point the previous
             * task has already completed.  This is why tasks should
             * not use their expected ending point to confirm that
             * they ended at the right point.  This thread should
             * perform the check after the task has completed. */
        }

        /* make sure there is at least one available task struct */
        check_task_done(sp);
        if (sp->error_code != 0)
        {
            pthread_mutex_unlock(&sp->sump_mtx);
            return;
        }

        TRACE("flush_in_buf: initializing task %d\n", sp->cnt_task_init);
        t = init_new_task(sp, ib, curr_rec);

        /* wake 1 sump thread */
        pthread_cond_signal(&sp->task_avail_cond);
    }
    pthread_mutex_unlock(&sp->sump_mtx);
    return;
}


/* sp_write_input - write data that is the input to a sump pump.
 *                  A write size of 0 indicates input EOF.
 */
ssize_t sp_write_input(sp_t sp, void *buf, ssize_t size)
{
    size_t              src_remaining = size;
    size_t              dst_remaining;
    size_t              trans_size;
    char                *trans_src;
    char                *trans_dst;
    in_buf_t            *ib;
#if !defined(SUMP_PUMP_NO_SORT)
    int                 ret;
#endif
            
    TRACE("sp_write_input: size %d\n", size);
    
    if (size <= 0 && sp->input_eof)
        return (0);    /* ignore, already eof */
    if (sp->error_code)
        return (-1);   /* already error */

    if (size <= 0)
    {
        if (size < 0)
        {
            /* TO-DO: need to make sure a partial input record does not
             * cause a sump hang instead of a sump error.  A sump hang
             * shouldn't matter if sump pump invoker waits for sump
             * pumps in upstream-to-downstream order */
            sp->error_code = SP_UPSTREAM_ERROR;
            pthread_mutex_lock(&sp->sump_mtx);
            broadcast_all_conds(sp);
            pthread_mutex_unlock(&sp->sump_mtx);
            size = 0;   /* act as if normal eof */
#if !defined(SUMP_PUMP_NO_SORT)
            if (sp->flags & SP_SORT)
            {
                (*Nsort_end)(&sp->nsort_ctx);
            }
#endif
            return (0);  /* caller is indicating error, return OK status */
        }
        /* else size == 0 */
#if !defined(SUMP_PUMP_NO_SORT)
        if (sp->flags & SP_SORT)
        {
            ret = (*Nsort_release_end)(&sp->nsort_ctx);

            if (ret < 0)        /* if error */
            {
                post_nsort_error(sp, ret);
                return (-1);
            }
            else
            {
                pthread_mutex_lock(&sp->sump_mtx);
                sp->sort_state = SORT_OUTPUT;
                pthread_cond_broadcast(&sp->task_output_ready_cond);
                pthread_mutex_unlock(&sp->sump_mtx);
                return (0);
            }
        }
#endif
        /* for non-sort case, fall through */
    }

#if !defined(SUMP_PUMP_NO_SORT)
    if (sp->flags & SP_SORT)
    {
        if (sp->sort_state != SORT_INPUT)
            return (0);
        ret = (*Nsort_release_recs)(buf, size, &sp->nsort_ctx);
        if (ret < 0)        /* if error */
            post_nsort_error(sp, ret);
        return (ret == NSORT_SUCCESS ? size : 0);
    }
#endif

    /* if EOF and there isn't a partially filled input buffer needing release.
     */
    if (src_remaining == 0 && sp->in_buf_current_bytes == 0)
    {
        eof_without_new_in_buf_or_task(sp);
        return (0);
    }
    
    /* while this is not EOF or there is a partially filled input buffer
     */
    while (src_remaining != 0 || sp->in_buf_current_bytes != 0)
    {
        /* if it is NOT the case that we have already partially filled an
         * input buffer that has not yet been released to the sump pump.
         * then get a new input buffer.
         */
        if (sp->in_buf_current_bytes == 0)
        {
            new_in_buf(sp);
            if (sp->error_code != 0)
                return (-1);
        }

        ib = &sp->in_buf[sp->cnt_in_buf_readable % sp->num_in_bufs];
        TRACE("sp_write_input: readable: %d, partial: %d\n",
              sp->cnt_in_buf_readable, sp->in_buf_current_bytes);

        trans_src = (char *)buf + size - src_remaining;
        trans_size = src_remaining;
        trans_dst = ib->in_buf + sp->in_buf_current_bytes;
        dst_remaining = ib->in_buf_size - sp->in_buf_current_bytes;
        if (trans_size > dst_remaining)
            trans_size = dst_remaining;
        memcpy(trans_dst, trans_src, trans_size);
        src_remaining -= trans_size;
        sp->in_buf_current_bytes += trans_size;
        dst_remaining = ib->in_buf_size - sp->in_buf_current_bytes;

        /* if this isn't EOF and there is more space remaining the in_buf,
         * then return so caller can write more bytes or declare EOF.
         */
        if (size != 0 && dst_remaining > 0)
        {
            if (src_remaining != 0)
                die("sp_write_input: dst_remaining %d and src_remaining %d\n",
                    dst_remaining, src_remaining);
            TRACE("sp_write_input() returning %d, dst_remaining: %d\n",
                  size, dst_remaining);
            return size;
        }

        flush_in_buf(sp, sp->in_buf_current_bytes, size == 0);
    }

    if (sp->error_code)
        size = -1;
    TRACE("sp_write_input: returning %d\n", size);
    return (size);
}


/* sp_get_in_buf - get a pointer to an input buffer that an external
 *                    thread can fill with input data.
 *
 * Returns: SP_OK or a sump pump error code
 */
int sp_get_in_buf(sp_t sp, uint64_t buf_index, void **buf, size_t *size)
{
    in_buf_t    *ib;
    
    if (sp->flags & SP_SORT)
        return (SP_SORT_INCOMPATIBLE);
    *buf = NULL;
    *size = 0;
    TRACE("sp_get_in_buf: waiting for input buffer\n");
    if (buf_index < sp->cnt_in_buf_readable ||
        buf_index >= sp->cnt_in_buf_readable + sp->num_in_bufs)
    {
        return (SP_BUF_INDEX_ERROR);
    }
    pthread_mutex_lock(&sp->sump_mtx);
    while (sp->error_code == 0 &&
           buf_index >= sp->cnt_in_buf_done + sp->num_in_bufs)
    {
        /* get oldest buffer not yet recognized as done */
        ib = &sp->in_buf[sp->cnt_in_buf_done % sp->num_in_bufs];
        /* if all readers of this input buffer are done reading */
        if (ib->num_readers == ib->num_readers_done)
        {
            sp->cnt_in_buf_done++;
            continue;
        }
        pthread_cond_wait(&sp->in_buf_done_cond, &sp->sump_mtx);
    }
    if (sp->error_code == 0)
    {
        ib = &sp->in_buf[buf_index % sp->num_in_bufs];
        *buf = ib->in_buf;
        *size = ib->in_buf_size;
    }
    pthread_mutex_unlock(&sp->sump_mtx);
    if (sp->error_code)
        return (sp->error_code);
    return (SP_OK);
}


/* sp_put_in_buf_bytes - flush bytes that have been placed in a sump
 *                          pump's input buffer by an external thread.
 *                          This function should only be used by first
 *                          calling sp_get_in_buf().
 *
 * Returns: SP_OK or a sump pump error code
 */
int sp_put_in_buf_bytes(sp_t sp, uint64_t buf_index, size_t size, int eof)
{
    if (sp->flags & SP_SORT)
        return (SP_SORT_INCOMPATIBLE);
    if (buf_index != sp->cnt_in_buf_readable)
        return (SP_BUF_INDEX_ERROR);
    if (size == 0)
        eof_without_new_in_buf_or_task(sp);
    else
        flush_in_buf(sp, size, eof);
    if (sp->error_code)
        return (sp->error_code);
    return (SP_OK);
}


/* sp_get_error - get the error code of a sump pump.
 *
 * Returns: SP_OK if no error has occurred, otherwise the error code.
 */
int sp_get_error(sp_t sp)
{
    return (sp->error_code);
}


/* pfunc_get_thread_index - can be used by pump functions to get the
 *                          index of the sump pump thread executing the
 *                          pump func.  For instance, if there are 4
 *                          sump pump threads, this function will return
 *                          0-3 depending on which of the 4 threads is
 *                          invoking it.
 *
 * Returns: The index of the requesting sump pump thread.
 */
int pfunc_get_thread_index(sp_task_t t)
{
    return (t->thread_index);
}


/* pfunc_get_task_number - can be used by pump functions to get the sump
 *                         pump task number being executed by the pump
 *                         function.  This number starts at 0 and
 *                         increases with each subsequent task
 *                         issued/started by the sump pump.
 *
 * Returns: The task number (starting with 0 as the first task) that the
 *          calling thread is currently executing.
 */
uint64_t pfunc_get_task_number(sp_task_t t)
{
    return (t->task_number);
}


/* pfunc_write - write function that can be used by a pump function to
 *               write the output data for the pump function.
 *
 * Returns: the number of bytes written.  If this is not the same as the
 *          requested size, an error has occurred.
 */
size_t pfunc_write(sp_task_t t, unsigned out_index, void *buf, size_t size)
{
    char        *src = (char *)buf;
    size_t      bytes_left = size;
    size_t      copy_bytes;
    sp_t        sp = t->sp;
    struct task_out *out = t->out + out_index;

    if (sp->error_code != SP_OK)
        return (0);
    
    if (out_index >= sp->num_outputs)
    {
        pfunc_error(t, "pfunc_write, out_index %d >= num_outputs %d\n",
                    out_index, sp->num_outputs);
        sp->error_code = SP_OUTPUT_INDEX_ERROR;
        return (0);
    }
    
    /* while can't fit existing map output buffer contents and the new record
     * into the map output buffer.
     */
    while (bytes_left + out->bytes_copied > out->size)
    {
        copy_bytes = out->size - out->bytes_copied;
        if (copy_bytes)
        {
            memmove(out->buf + out->bytes_copied, src, copy_bytes);
            src += copy_bytes;
            bytes_left -= copy_bytes;
            out->bytes_copied += copy_bytes;
        }
        TRACE("pfunc_write: waking output reader\n");
        pthread_mutex_lock(&sp->sump_mtx);
        out->stalled = TRUE;
        pthread_cond_broadcast(&sp->task_output_ready_cond);
        TRACE("pfunc_write: waiting for available output buffer\n");
        while (out->stalled && sp->error_code == 0)
            pthread_cond_wait(&sp->task_output_empty_cond, &sp->sump_mtx);
        pthread_mutex_unlock(&sp->sump_mtx);
        if (sp->error_code != 0)
            return (-1);
    }
    /* copy new record into buffer */
    memmove(out->buf + out->bytes_copied, src, bytes_left);
    out->bytes_copied += bytes_left;
    return (size);
}


/* pfunc_mutex_lock - lock the auto-allocated mutex for pump functions.
 */
void pfunc_mutex_lock(sp_task_t t)
{
    pthread_mutex_lock(&t->sp->sp_mtx);
}


/* pfunc_mutex_unlock - unlock the auto-allocated mutex for pump functions.
 */
void pfunc_mutex_unlock(sp_task_t t)
{
    pthread_mutex_unlock(&t->sp->sp_mtx);
}


/* done_reading_in_buf - internal routine to mark the task's current
 *                       input buffer as done for reading.  This is
 *                       non-trivial because multiple pump threads may
 *                       have to read the same input buffer.
 */
static void done_reading_in_buf(sp_task_t t, int move_to_next_in_buf)
{
    in_buf_t    *ib;
    sp_t        sp = t->sp;

    pthread_mutex_lock(&sp->sump_mtx);

    /* bump up the count of done readers for the input buffer */
    ib = &sp->in_buf[t->curr_in_buf_index % sp->num_in_bufs];
    ib->num_readers_done++;
    /* if all readers are now done, signal the reader thread */
    if (ib->num_readers == ib->num_readers_done)
        pthread_cond_broadcast(&sp->in_buf_done_cond);

    pthread_mutex_unlock(&sp->sump_mtx);

    if (move_to_next_in_buf)
    {
        /* move this task's current input position to the beginning of
         * the next input buffer.
         */
        t->curr_in_buf_index++;   /* on to next input buffer */
        ib = &sp->in_buf[t->curr_in_buf_index % sp->num_in_bufs];
        t->in_buf = ib->in_buf;
        t->in_buf_bytes = 0;
        t->curr_rec = t->in_buf;
    }
}


/* ready_in_buf - internal routine to insure the current input buffer
 *                for a task is ready for reading.
 */
static void ready_in_buf(sp_task_t t)
{
    in_buf_t     *ib;
    sp_t        sp = t->sp;

    pthread_mutex_lock(&sp->sump_mtx);

    t->first_in_buf = FALSE; /* now that this task has proceeded beyond
                              * its first input buffer, it should stop
                              * at end of the current key group */

    while (sp->error_code == 0 && !sp->input_eof && 
           t->curr_in_buf_index == sp->cnt_in_buf_readable) /*not yet readable*/
    {
        pthread_cond_wait(&sp->in_buf_readable_cond, &sp->sump_mtx);
    }
    
    /* if sp_write_input() has indicated eof and no more readable buffers
     * then that indicates eof for this task.
     */
    if (sp->error_code != 0 ||
        (sp->input_eof &&
         t->curr_in_buf_index == sp->cnt_in_buf_readable))
    {
        t->input_eof = TRUE;
        TRACE("pump%d: eof: error_code: %d, in_buf_bytes is 0 at ib %d\n",
              t->thread_index, sp->error_code, t->curr_in_buf_index);
        t->in_buf = NULL;
        t->in_buf_bytes = 0;
        t->curr_rec = NULL;
    }
    else
    {
        ib = &sp->in_buf[t->curr_in_buf_index % sp->num_in_bufs];
        t->in_buf = ib->in_buf;
        t->in_buf_bytes = ib->in_buf_bytes;        
        t->curr_rec = t->in_buf;
    }

    pthread_mutex_unlock(&sp->sump_mtx);
}


/* is_more_input - internal routine to test if there is more input for
 *                 this pump task.  This routine is called after the
 *                 pump function returns to see if there is additonal
 *                 input data for the task and hence the pump function
 *                 should be called again.
 */
static int is_more_input(sp_task_t t)
{
    if (t->input_eof)   /* if input eof, then false (no more input) */
        return (0);
    
    /* if we are past the first input buffer for this task (because we
     * had to read the remainder of the last task record at the
     * beginning of the second input buffer), then we have hit the end
     * of this task's input.  Note that if GROUP_BY is specified,
     * then the task pump function can read past any record remainder
     * at the beginning of the second input buffer.
     */
    if (!t->first_in_buf && !(t->sp->flags & SP_GROUP_BY))
    {
        done_reading_in_buf(t, t->curr_rec == t->in_buf + t->in_buf_bytes);
        t->input_eof = TRUE;
        return (0);
    }
    
    /* if there are bytes left in the current input buffer, then true
     * (more input).
     */
    TRACE("pump: imi, rec %08x, buf %08x, bytes: %x\n",
          t->curr_rec, t->in_buf, t->in_buf_bytes);
    if (t->curr_rec < t->in_buf + t->in_buf_bytes)
        return (1);

    /* this task is done reading its current input buffer */
    done_reading_in_buf(t, TRUE);
    
    /* if we are not grouping by key difference, then we have hit the
     * end of this task's input.
     */
    if (!(t->sp->flags & SP_GROUP_BY))
    {
        t->input_eof = TRUE;
        return (0);
    }

    /* get next input buffer */
    ready_in_buf(t);
    TRACE("pump: imi, eof %d\n", t->input_eof);
    return (!t->input_eof);
}


/* pfunc_get_rec - get a pointer to the next input record for a pump function.
 *                 The sump pump infrastructure allocates the record buffers
 *                 and modifies the pointer-to-a-pointer argument to point to
 *                 the buffer. If the record type is text, a null character
 *                 will terminate the record.
 *
 * Returns: 0 if no more records are in the sump pump task input, otherwise
 *          the number bytes in the record not including the terminating null
 *          character.
 */
size_t pfunc_get_rec(sp_task_t t, void *ptr_to_rec_ptr)
{
    void        *buf;
    char        *rec;
    char        *next_rec;
    int         match_char;
    size_t      len;
    size_t      src_size;
    size_t      trans_size;
    size_t      delim_size = 0;
    sp_t        sp = t->sp;
    int         new_key_group_beginning = FALSE;

    buf = t->rec_buf;
    
    if (t->input_eof)
        return 0;

    if (REC_TYPE(sp) == SP_UTF_8)
        delim_size = 1;
    
    /* if we have used all the records in the current input buffer...
     * note that we must be on a record boundry at this point.
     */
    if (t->curr_rec >= t->in_buf + t->in_buf_bytes)
    {
        done_reading_in_buf(t, TRUE);  /* done reading input buffer */
        
        /* if we are not grouping records then return no record (0) since
         * we are on a record boundry.
         */
        if (!(sp->flags & SP_GROUP_BY))
        {
            t->input_eof = TRUE;
            return 0;
        }

        ready_in_buf(t);
        if (t->input_eof)
            return 0;
    }

    /* now we have at least a partial record in the input buffer */
    rec = t->curr_rec;
    /* if there is an initial key offset character */
    if (sp->flags & SP_GROUP_BY)
    {
        match_char = rec[0];
        if (match_char != '0' && match_char != '1')
        {
            pfunc_error(t, "-group was specified, but an input record does not start with\n"
                        "'0' or '1'. Was nsort -match used to generate input?\n");
            /* return 0; don't return EOF as this can result in infinite loop*/
        }
        rec++;
        new_key_group_beginning =
            (match_char == '0' && !t->first_group_rec);
    }

    /* if the key offset indicates this is the beginning of a new key group &&
     * this is not the actual first record for this current key group.
     */
    if (!(sp->flags & SP_GROUP_BY) || new_key_group_beginning)
    {
        /* if we are beyond the first input buffer for this task, this
         * rec is not only the beginning of the next key group, it also
         * means we have reached the end of the input for this sump pump
         * task.
         */
        if (!t->first_in_buf)
        {
            done_reading_in_buf(t, FALSE);
            TRACE("pump%d: eof: match: %c at ib %d, curr_rec %08x\n",
                  t->thread_index,
                  (sp->flags & SP_GROUP_BY) ? match_char : '0',
                  t->curr_in_buf_index, t->curr_rec);
            t->input_eof = TRUE;   /* no need to be inside the mutex because
                                     * only the current thread reads this */
            return 0;
        }
        /*
         * else we are still reading from the first input buffer for this
         * task, but there is at least one more record key group to be
         * processed by this task.
         */
        else if (new_key_group_beginning)
            return 0;
        /* else we are not doing key grouping, go on to return next rec */
    }
    t->first_group_rec = FALSE;  /* only relevant for key grouping */

    /* now copy the record, which may span multiple in_bufs, into the
     * caller's buffer.
     */
    len = 0;
    for (;;)
    {
        src_size = t->in_buf_bytes - (rec - t->in_buf);
        switch (REC_TYPE(sp))
        {
          case SP_UTF_8:
            next_rec = memchr(rec, *(char *)sp->delimiter, src_size);
            if (next_rec != NULL)
            {
                next_rec++;         /* skip over newline */
                trans_size = next_rec - rec;
            }
            else
                trans_size = src_size;
            break;

          case SP_FIXED:
            trans_size = src_size;
            if (trans_size >= sp->rec_size - len)
            {
                trans_size = sp->rec_size - len;
                next_rec = rec + trans_size;
            }
            else
                next_rec = NULL;
            break;
        }            
        if (trans_size + len + delim_size > t->rec_buf_size)
        {
            size_t new_size;

            if (REC_TYPE(sp) == SP_FIXED)
                new_size = sp->rec_size;
            else
                new_size = trans_size + len + delim_size + 50;
            
            if (t->rec_buf == NULL)
                t->rec_buf = (char *)calloc(1, new_size);
            else
                t->rec_buf = realloc(t->rec_buf, new_size);
            if (t->rec_buf == NULL)
            {
                die("pfunc_get_rec: rec_buf increase failed: old %d new %d\n",
                    t->rec_buf_size, new_size);
            }
            t->rec_buf_size = new_size;
            buf = t->rec_buf;
        }
        memmove((char *)buf + len, rec, trans_size);
        len += trans_size;
        if (next_rec != NULL)  /* if we found the newline delimiter */
        {
            if (REC_TYPE(sp) == SP_UTF_8)
                ((char *)buf)[len] = '\0';
            break;
        }
        else  /* else we need to get remainder of record from next buffer */
        {
            done_reading_in_buf(t, TRUE);
            ready_in_buf(t);
            if (t->input_eof)
            {
                if (REC_TYPE(sp) == SP_FIXED)
                {
                    die("pfunc_get_rec: partial record of "
                        "%d bytes found at end of input\n", len);
                }
                return 0;
            }
            rec = t->curr_rec;
        }
    }
    t->curr_rec = next_rec;
    *(void **)ptr_to_rec_ptr = buf;
    return len;
}


/* pfunc_get_in_buf - get a pointer to the input buffer for a pump function.
 */
int pfunc_get_in_buf(sp_task_t t, void **buf, size_t *size)
{
    *(char **)buf = t->curr_rec;
    *size = (t->in_buf + t->in_buf_bytes) - t->curr_rec;
    t->curr_rec = t->in_buf + t->in_buf_bytes;
    return (0);
}


/* pfunc_get_out_buf - get a pointer to an output buffer and its size for
 *                     a pump function.
 *
 * Returns: SP_OK or a sump pump error code
 */
int pfunc_get_out_buf(sp_task_t t, unsigned out_index, void **buf, size_t *size)
{
    sp_t                sp = t->sp;
    struct task_out     *out = t->out + out_index;

    if (out_index >= sp->num_outputs)
    {
        pfunc_error(t, "pfunc_get_out_buf: out_index %d >= num_outputs %d\n",
                    out_index, sp->num_outputs);
    }

    if (out->bytes_copied == out->size)
    {
        sp_t    sp = t->sp;
        
        TRACE("pfunc_get_out_buf: waking output reader\n");
        pthread_mutex_lock(&sp->sump_mtx);
        out->stalled = TRUE;
        pthread_cond_broadcast(&sp->task_output_ready_cond);
        TRACE("pfunc_get_out_buf: waiting for available output buffer\n");
        while (out->stalled && sp->error_code == 0)
            pthread_cond_wait(&sp->task_output_empty_cond, &sp->sump_mtx);
        pthread_mutex_unlock(&sp->sump_mtx);
        if (sp->error_code != 0)
            return (-1);
    }
    *(char **)buf = out->buf + out->bytes_copied;
    *size = out->size - out->bytes_copied;
    return (0);
}


/* pfunc_put_out_buf_bytes - flush bytes that have been placed in a pump
 *                           function's output buffer.  This routine can only
 *                           be used by first calling sp_get_out_buf().
 *
 * Returns: SP_OK or a sump pump error code
p */
int pfunc_put_out_buf_bytes(sp_task_t t, unsigned out_index, size_t size)
{
    sp_t                sp = t->sp;
    struct task_out     *out = t->out + out_index;

    if (out_index >= sp->num_outputs)
    {
        pfunc_error(t, "pfunc_put_out_buf_bytes: "
                    "out_index %d >= num_outputs %d\n",
                    out_index, sp->num_outputs);
    }
    else if (out->bytes_copied + size > out->size)
    {
        pfunc_error(t, "sp_put_out_buf_bytes: "
                    "aggregate size (%d+%d) is larger than buf size %d\n",
                    out->bytes_copied, size, out->size);
    }
    else
        out->bytes_copied += size;
    return (sp->error_code);
}


/* pfunc_printf - print a formatted string to a pump functions's output.
 *
 * Returns: the number of bytes written.  If this is not the same as the
 *          requested size, an error has occurred.
 */
int pfunc_printf(sp_task_t t, unsigned out_index, const char *fmt, ...)
{
    va_list     ap;
    ssize_t     ret;

    va_start(ap, fmt);
    ret = vsnprintf(t->temp_buf, t->temp_buf_size, fmt, ap);
    va_end(ap);
#if defined(win_nt)
    if (ret == -1)  /* non-standard vsnprintf overflow indicator on Windows */
    {
        va_start(ap, fmt);
        ret = _vscprintf(fmt, ap);
        va_end(ap);
    }
#endif
    if ((size_t)ret >= t->temp_buf_size)
    {
        /* temp buf wasn't big enough.  enlarge it and redo */
        if (t->temp_buf_size != 0)
            free(t->temp_buf);
        t->temp_buf_size = ret + 10;
        t->temp_buf = (char *)malloc(t->temp_buf_size);
        va_start(ap, fmt);
        ret = vsnprintf(t->temp_buf, t->temp_buf_size, fmt, ap);
        va_end(ap);
    }
    return ((int)pfunc_write(t, out_index, t->temp_buf, ret));
}


/* pfunc_error - raise an error for a pump function and define an error 
 *               string that can be retrived by other threads using
 *               sp_get_error_string().
 *
 * Returns: SP_PUMP_FUNCTION_ERROR
 */
int pfunc_error(sp_task_t t, const char *fmt, ...)
{
    va_list     ap;
    int         ret;

    if (t->error_code != 0)            /* if prior error */
        return SP_PUMP_FUNCTION_ERROR;   /* ignore this one */
    t->error_code = SP_PUMP_FUNCTION_ERROR;
    
    va_start(ap, fmt);
    ret = vsnprintf(t->error_buf, t->error_buf_size, fmt, ap);
    va_end(ap);
#if defined(win_nt)
    if (ret == -1)  /* non-standard vsnprintf overflow indicator on Windows */
    {
        va_start(ap, fmt);
        ret = _vscprintf(fmt, ap);
        va_end(ap);
    }
#endif
    if ((size_t)ret >= t->error_buf_size)
    {
        if (t->error_buf_size != 0)
            free(t->error_buf);
        t->error_buf_size = (size_t)ret + 1;
        t->error_buf = (char *)malloc(t->error_buf_size);
        va_start(ap, fmt);
        vsnprintf(t->error_buf, t->error_buf_size, fmt, ap);
        va_end(ap);
    }
    
    return SP_PUMP_FUNCTION_ERROR;
}


/* pump_thread_main - the internal "main" routine of a sump pump thread.
 */
static void *pump_thread_main(void *arg)
{
    sp_task_t           t;
    unsigned            thread_index;
    sp_t                sp = (sp_t)arg;
    int                 ret;

    for (thread_index = 0; thread_index < sp->num_threads; thread_index++)
#if defined(win_nt)
        if (sp->thread[thread_index].id == GetCurrentThreadId())
#else
        if (sp->thread[thread_index] == pthread_self())
#endif
            break;
    TRACE("pump thread %d starting\n", thread_index);
    for (;;)
    {
        TRACE("pump%d: waiting for an available task\n", thread_index);
        pthread_mutex_lock(&sp->sump_mtx);
        while (sp->cnt_task_begun == sp->cnt_task_init &&
               sp->error_code == 0 &&
               sp->input_eof == FALSE)
        {
            pthread_cond_wait(&sp->task_avail_cond, &sp->sump_mtx);
        }
        if (sp->error_code != 0 ||
            (sp->input_eof == TRUE && sp->cnt_task_begun == sp->cnt_task_init))
        {
            pthread_mutex_unlock(&sp->sump_mtx);
            TRACE("pump%d: breaking out of for loop: error_code: %d, input_eof %d\n",
                  thread_index, sp->error_code, sp->input_eof);
            break;
        }
        t = &sp->task[sp->cnt_task_begun % sp->num_tasks];
        sp->cnt_task_begun++;
        t->thread_index = thread_index;
        pthread_mutex_unlock(&sp->sump_mtx);

        TRACE("pump%d: calling pump func with %d input bytes\n",
              thread_index, (int)t->in_buf_bytes);

        if (REC_TYPE(sp) == SP_WHOLE_BUF)
        {
            TRACE("pump%d: calling pump func() block\n", thread_index);
            ret = (*sp->pump_func)(t, sp->pump_arg);
            TRACE("pump%d: pump func returned %d\n", thread_index, ret);
            if (ret)
            {
                if (t->error_code == 0)
                    t->error_code = ret;
            }
            else
            {
                done_reading_in_buf(t, TRUE);
                t->input_eof = TRUE;
            }
        }
        else
        {
            while (is_more_input(t) && t->error_code == 0)
            {
                TRACE("pump%d: calling pump func()\n", thread_index);
                /* indicate first record in key group not yet read */
                t->first_group_rec = TRUE;
                ret = (*sp->pump_func)(t, sp->pump_arg);
                TRACE("pump%d: pump func returned %d, input_eof: %d\n",
                      thread_index, ret, t->input_eof);
                if (ret && t->error_code == 0)
                    t->error_code = ret;
            }
        }
        TRACE("pump%d: pump_func returns with %d out[0] bytes\n",
              thread_index, t->out[0].bytes_copied);
        pthread_mutex_lock(&sp->sump_mtx);
        if (t->error_code && sp->error_code == 0)
        {
            sp->error_code = t->error_code;
            if (sp->error_buf != NULL)
                free(sp->error_buf);
            sp->error_buf = t->error_buf;
            t->error_buf = NULL;
            broadcast_all_conds(sp);
        }
        pthread_mutex_unlock(&sp->sump_mtx);

        TRACE("pump%d: waking output reader\n", thread_index);
        TRACE("pump%d: waking input writer\n", thread_index);
        pthread_mutex_lock(&sp->sump_mtx);
        t->output_eof = TRUE;
        pthread_cond_broadcast(&sp->task_output_ready_cond);
        pthread_mutex_unlock(&sp->sump_mtx);
        /* NOTA BENE: do not use "t" pointer after this point since the
         * struct that it points to can be reused immediately */
    }
    TRACE("pump%d: exiting\n", thread_index);

    return (NULL);
}


/* sp_read_output - read bytes from the specified output of a sump pump.
 *
 * Returns: The number of bytes read.  If 0, then EOF has occurred.
 *          If negative, an error has occurred.
 */
ssize_t sp_read_output(sp_t sp, unsigned index, void *buf, ssize_t size)
{
    int                 out_eof;
    ssize_t             bytes_returned = 0;
    ssize_t             src_remaining = size;
    ssize_t             dst_remaining;
    ssize_t             trans_size;
    char                *trans_src;
    char                *trans_dst;
    sp_task_t           t;
    struct task_out     *out;

    TRACE("sp_read_output[%d]: buf %08x, size %d\n", index, buf, size);

    if (sp->error_code)
        return (-1);

#if !defined(SUMP_PUMP_NO_SORT)
    if (sp->flags & SP_SORT)
    {
        int     ret;

        if (sp->sort_state == SORT_INPUT)
        {
            pthread_mutex_lock(&sp->sump_mtx);
            while (sp->error_code == 0 && sp->sort_state == SORT_INPUT)
                pthread_cond_wait(&sp->task_output_ready_cond, &sp->sump_mtx);
            pthread_mutex_unlock(&sp->sump_mtx);
            if (sp->error_code != 0)
                return (-1);
        }
        if (sp->sort_state == SORT_DONE)
            return (0);

        while (bytes_returned < size)
        {
            /* if there is no data in the sort_temp_buf, then fill it.
             */
            if (sp->out[0].partial_bytes_copied == 0)
            {
                /* call nsort_return_recs() until it does NOT return a buffer
                 * too small error.
                 */
                for (;;)
                {
                    char    *temp;
                
                    sp->sort_temp_buf_bytes = sp->sort_temp_buf_size;
                    ret = (*Nsort_return_recs)(sp->sort_temp_buf,
                                               &sp->sort_temp_buf_bytes,
                                               &sp->nsort_ctx);
                    if (ret != NSORT_RETURN_BUF_SMALL)
                        break;
                    temp = sp->sort_temp_buf;
                    sp->sort_temp_buf_size *= 2;
                    sp->sort_temp_buf = (char *)malloc(sp->sort_temp_buf_size);
                    if (sp->sort_temp_buf == NULL)
                    {
                        sp->error_code = SP_MEM_ALLOC_ERROR;
                        return (-1);
                    }
                    free(temp);
                }
                if (ret == NSORT_END_OF_OUTPUT)
                {
                    pthread_mutex_lock(&sp->sump_mtx);
                    sp->sort_state = SORT_DONE;
                    pthread_cond_broadcast(&sp->task_output_ready_cond);
                    pthread_mutex_unlock(&sp->sump_mtx);
                    return (bytes_returned);
                }
                else if (ret < 0)
                {
                    post_nsort_error(sp, ret);
                    return (-1);
                }
            }

            src_remaining =
                sp->sort_temp_buf_bytes - sp->out[0].partial_bytes_copied;
            dst_remaining = size - bytes_returned;
            trans_size = dst_remaining;
            trans_src = sp->sort_temp_buf + sp->out[0].partial_bytes_copied;
            trans_dst = (char *)buf + bytes_returned;
            if (dst_remaining <= src_remaining)
            {
                /* we will fill the caller's buffer completely.
                 */
                memmove(trans_dst, trans_src, dst_remaining);
                bytes_returned += dst_remaining;
                sp->out[0].partial_bytes_copied += dst_remaining;
                break;
            }

            /* the sort temp buffer will be completely copied out,
             * and we still need more.
             */
            memmove(trans_dst, trans_src, src_remaining);
            bytes_returned += src_remaining;
            /* indicate sort temp output buffer needes to be refilled */
            sp->out[0].partial_bytes_copied = 0;
        }
        return (bytes_returned);
    }
#endif
    
    if (index >= sp->num_outputs)
        return (-1);
            
    for (;;)
    {
        TRACE("sp_read_output: out[%d].cnt_task_drained: %d, "
              "partial_bytes_copied: %d\n",
              index, sp->out[index].cnt_task_drained,
              sp->out[index].partial_bytes_copied);
        t = &sp->task[sp->out[index].cnt_task_drained % sp->num_tasks];
        out = t->out + index;

        /* if we aren't in the middle of copy out a task's buffer, then
         * we must potentially wait for the output of the next task.
         */
        if (sp->out[index].partial_bytes_copied == 0)
        {
            TRACE("sp_read_output: waiting for input\n");
            pthread_mutex_lock(&sp->sump_mtx);

            /* while 1) a sump pump error hasn't occurred.
             * and   2) it's not the case that
             *          a) EOF on input has been reached
             *          b) all initialized tasks have begun (been taken), and
             *          c) all taken tasks have had their output read, and
             * and   3) it's not the case oldest sump pump task is either
             *          done or stalled
             */
            while (sp->error_code == 0 &&
                   !(out_eof = (sp->input_eof &&
                                sp->cnt_task_init == sp->cnt_task_begun &&
                                sp->cnt_task_begun == sp->out[index].cnt_task_drained)) &&
                   !(sp->out[index].cnt_task_drained < sp->cnt_task_begun &&
                     (out->stalled || t->output_eof)))
            {
                TRACE("sp_read_output: waiting\n");
                TRACE("sp_read_output: cnt_task_init: %d\n",
                      sp->cnt_task_init);
                TRACE("sp_read_output: cnt_task_begun: %d\n",
                      sp->cnt_task_begun);
                TRACE("sp_read_output: out[%d].cnt_task_drained: %d\n",
                      index, sp->out[index].cnt_task_drained);
                TRACE("sp_read_output: out->bytes_copied: %d\n",
                      out->bytes_copied);
                TRACE("sp_read_output: t->in_buf_bytes: %d\n",
                      t->in_buf_bytes);
                TRACE("sp_read_output: out->stalled: %d\n", out->stalled);
                TRACE("sp_read_output: t->output_eof: %d\n", t->output_eof);
                pthread_cond_wait(&sp->task_output_ready_cond, &sp->sump_mtx);
            }
            TRACE("sp_read_output: DONE WAITING\n");
            TRACE("sp_read_output: error_code: %d\n", sp->error_code);
            TRACE("sp_read_output: cnt_task_init: %d\n", sp->cnt_task_init);
            TRACE("sp_read_output: cnt_task_begun: %d\n", sp->cnt_task_begun);
            TRACE("sp_read_output: out[%d].cnt_task_drained: %d\n",
                  index, sp->out[index].cnt_task_drained);
            TRACE("sp_read_output: out->bytes_copied: %d\n",
                  out->bytes_copied);
            TRACE("sp_read_output: t->in_buf_bytes: %d\n", t->in_buf_bytes);
            TRACE("sp_read_output: out->stalled: %d\n", out->stalled);
            TRACE("sp_read_output: t->output_eof: %d\n", t->output_eof);
            TRACE("sp_read_output: out_eof: %d\n", out_eof);
            pthread_mutex_unlock(&sp->sump_mtx);
            if (sp->error_code || out_eof)
                break;
        }  
        src_remaining =
            out->bytes_copied - sp->out[index].partial_bytes_copied;
        dst_remaining = size - bytes_returned;
        trans_size = dst_remaining;
        trans_src = t->out[index].buf + sp->out[index].partial_bytes_copied;
        trans_dst = (char *)buf + bytes_returned;
        if (dst_remaining < src_remaining)
        {
            /* we will fill the output buffer and still leave some bytes
             * in the sump pump task's output buffer.
             */
            memmove(trans_dst, trans_src, dst_remaining);
            bytes_returned += dst_remaining;
            sp->out[index].partial_bytes_copied += dst_remaining;
            break;
        }

        /* the sump pump task's output buffer will be completely copied out.
         */
        memmove(trans_dst, trans_src, src_remaining);
        bytes_returned += src_remaining;
        /* indicate new sump pump task output is needed */
        sp->out[index].partial_bytes_copied = 0;  
        
        TRACE("sp_read_output: waking reader thread\n");
        pthread_mutex_lock(&sp->sump_mtx);
        if (t->out[index].stalled)
        {
            /* we have copied the bytes in the buf.  clear the buf
             * and stall indicator, then wake stalled thread.  since
             * there may be more than one thread waiting on the
             * task_output_empty_cond, use broadcast.
             */
            t->out[index].bytes_copied = 0;
            t->out[index].stalled = FALSE;
            pthread_cond_broadcast(&sp->task_output_empty_cond);
        }
        else
        {
            /* increment per-output task output drained count */
            sp->out[index].cnt_task_drained++;
            TRACE("sp_read_output: sp->out[%d].cnt_task_drained incr to %d\n",
                  index, sp->out[index].cnt_task_drained);
            TRACE("sp_read_output: t->outs_drained before incr is: %d\n",
                  t->outs_drained);

            /* if all output buffers for this task have been drained */
            if (++t->outs_drained == sp->num_outputs)  
            {
                /* increment sump pump task drained count */
                sp->cnt_task_drained++;
                TRACE("sp_read_output: sp->cnt_task_drained incr to: %d\n",
                      sp->cnt_task_drained);
                pthread_cond_broadcast(&sp->task_drained_cond);
            }
        }

        pthread_mutex_unlock(&sp->sump_mtx);

        if (bytes_returned == size)
            break;
    }

    if (sp->error_code)
        bytes_returned = -1;
    TRACE("sp_read_output: returning %d bytes\n", bytes_returned);
    return (bytes_returned);
}


/* link_main - internal "main" routine for a thread that links an output
 *             of a sump pump to the input of another sump pump.
 */
static void *link_main(void *arg)
{
    sp_link_t   sp_link = (sp_link_t)arg;
    ssize_t     size;

    while ((size = sp_read_output(sp_link->out_sp,
                                  sp_link->out_index,
                                  sp_link->buf,
                                  sp_link->buf_size)) > 0)
    {
        if (sp_write_input(sp_link->in_sp, sp_link->buf, size) != size)
        {
            TRACE("link_main: sp_write_input() returned wrong size\n");
            sp_link->error_code = SP_WRITE_ERROR;
            return (NULL);
        }
    }
    sp_write_input(sp_link->in_sp, NULL, size);/* need to handle err case?*/
    if (size < 0)
        sp_link->error_code = SP_UPSTREAM_ERROR;
    return (NULL);
}


/* sp_link - start a link/connection between an output
 *           of a sump pump to the input of another sump pump.
 *
 * Returns: SP_OK or a sump pump error code
 */
int sp_link(sp_t out_sp, unsigned out_index, sp_t in_sp)
{
    struct sp_link      *sp_link;
    int                 ret;
    int                 group_by_input;

    sp_link = (struct sp_link *)calloc(1, sizeof(struct sp_link));
    if (sp_link == NULL)
        return (SP_MEM_ALLOC_ERROR);
    group_by_input = (in_sp->flags & SP_GROUP_BY) ? TRUE : FALSE;
    if (out_sp->match_keys ^ group_by_input)
        return (SP_GROUP_BY_MISMATCH);
    sp_link->out_sp = out_sp;
    sp_link->out_index = out_index;
    sp_link->in_sp = in_sp;
    sp_link->buf_size = 4096;
    sp_link->buf = (char *)malloc(sp_link->buf_size);
    if (sp_link->buf == NULL)
        return (SP_MEM_ALLOC_ERROR);
    if ((ret = pthread_create(&sp_link->thread, NULL, link_main, sp_link)))
        die("sp_start_link: pthread_create() ret: %d\n", ret);

    return (SP_OK);
}


/* get_output_index - internal routine to read an output index preceded by
 *                    a "[" and followed by "]=".
 */
static int get_output_index(sp_t sp, char **caller_p)
{
    int         index;
    char        *p = *caller_p;
    
    if (*p++ != '[')
    {
        syntax_error(sp, p, "expected '['");
        return 0;
    }
    index = (int)get_numeric_arg(sp, &p);
    if (index < 0 || (unsigned)index >= sp->num_outputs)
    {
        syntax_error(sp, p,
                     "output index is greater than the number of outputs");
        return 0;
    }
    if (*p++ != ']')
    {
        syntax_error(sp, p, "expected ']'");
        return 0;
    }
    if (*p++ != '=')
    {
        syntax_error(sp, p, "expected '='");
        return 0;
    }
    *caller_p = p;
    return (index);
}


/* get_logical_processor_count - internal routine to get the number of
 *                               logical processors in the system.
 */
static int get_logical_processor_count()
{
#if defined(win_nt)
    {
        SYSTEM_INFO si;

        GetSystemInfo(&si);
        return si.dwNumberOfProcessors;
    }
#else
    return sysconf(_SC_NPROCESSORS_ONLN);
#endif
}


/* get_string_arg - internal routine to scan and return a string
 */
static char *get_string_arg(char **caller_p)
{
    char        *begin_p = *caller_p;
    char        *p;
    char        *ret;
    int         i;

    /* scan string up to the next white space character */
    p = begin_p;
    while (!isspace(*(unsigned char *)p) && *p != '\0')
        p++;
    *caller_p = p;

    /* allocate space for the return string and copy contents to it */
    ret = (char *)calloc(1, p - begin_p + 1);
    for (i = 0; i < p - begin_p; i++)
        ret[i] = begin_p[i];
    ret[p - begin_p] = '\0';
    return (ret);
}


/* sp_argv_to_str - bundle up the specified argv and return it as a string.
 *                  For instance if argc is 2, argv[0] is "TASKS=2" and
 *                  argv[1] is "THEADS=3", then return the string
 *                  "TASKS=2 THREADS=3".
 *
 * Returns: a string containing the command line arguments passed as function
 *          arguments. The string should be free()'d when no longer needed.
 */
char *sp_argv_to_str(char *argv[], int argc)
{
    int                 i;
    char                *str = NULL;
    int                 n_chars;

    n_chars = 1;      /* '\0' */
    for (i = 0; i < argc; i++)
        n_chars += 1 + (int)strlen(argv[i]);   /* ' ' + argv[i] */
    str = (char *)calloc(sizeof(char), n_chars + 1);
    strcpy(str, argc == 0 ? "" : argv[0]);
    for (i = 1; i < argc; i++)
    {
        strcat(str, "\n");
        strcat(str, argv[i]);
    }
    return (str);
}


/* get_exec_args - internal routine to get an external program name and its
 *                 arguments
 */
static void get_exec_args(sp_t sp, char **ep)
{
    char        *p, *begin;
    int         i, cnt;
    
#if defined(win_nt)
    int         cmdlen;
    char        *cmdline;
    
    for (p = *ep, cnt = 0; *p != '\0'; p++)
        if (*p == '\n')
            cnt++;      /* count newlines */
    sp->exec_argv = (char **)calloc(2, sizeof(char *));
    if (sp->exec_argv == NULL)
    {
        sp->error_code = SP_MEM_ALLOC_ERROR;
        return;
    }
    /* allocate space for each character plus 2 double-quote chars per arg */
    cmdline = sp->exec_argv[0] =
        (char *)calloc((p - *ep) + 2 * (cnt + 1), sizeof(char));
    if (cmdline == NULL)
    {
        sp->error_code = SP_MEM_ALLOC_ERROR;
        return;
    }
    cmdlen = 0;
#endif
    
    p = *ep;
    i = 0;
    cnt = 0;
    for (;;)
    {
#if defined(win_nt)
        int     need_quotes = FALSE;
        if (i != 0)
            cmdline[cmdlen++] = ' ';   /* add space between args */
#else           /* grow argv for non-Windows systems */
# define SP_ARGV_INCR    5
        if (i == 0)
        {
            cnt = SP_ARGV_INCR;
            sp->exec_argv = (char **)calloc(cnt + 1, sizeof(char *));
        }
        else if (i == cnt)
        {
            cnt += SP_ARGV_INCR;
            sp->exec_argv = (char **)
                realloc(sp->exec_argv, (cnt + 1) * sizeof(char *));
        }
        if (sp->exec_argv == NULL)
        {
            sp->error_code = SP_MEM_ALLOC_ERROR;
            return;
        }
#endif
        begin = p;
        while (*p != '\n' && *p != '\0')
        {
#if defined(win_nt)
            if (*p == ' ')
                need_quotes = TRUE;
#endif
            p++;
        }

#if defined(win_nt)
        if (need_quotes)
            cmdline[cmdlen++] = '"';   /* begin double-quote */
        memcpy(cmdline + cmdlen, begin, p - begin);
        cmdlen += p - begin;
        if (need_quotes)
            cmdline[cmdlen++] = '"';   /* end double-quote */
#else     
        sp->exec_argv[i] = calloc(sizeof(char), p - begin + 1);
        if (sp->exec_argv[i] == NULL)
        {
            sp->error_code = SP_MEM_ALLOC_ERROR;
            return;
        }
        memcpy(sp->exec_argv[i], begin, p - begin);
        sp->exec_argv[i][begin - p] = '\0';
        TRACE("argv[%d]: %s\n", i, sp->exec_argv[i]);
#endif
        if (*p == '\n')
            p++;
        if (*p == '\0')
            break;
        i++;
    }
    *ep = p;

#if defined(win_nt)
    cmdline[cmdlen] = '\0';
    TRACE("cmdline: %s\n", cmdline);
#endif
}


/* sp_start - Start a sump pump
 *
 * Parameters:
 *      sp -          Pointer to where to return newly allocated sp_t 
 *                    identifier that will be used in as the first argument
 *                    to all subsequent sp_*() calls.
 *      pump_func -   Pointer to pump function that will be called by
 *                    multiple sump pump threads at once. If an external
 *                    program name is specified in the arg_fmt string, this
 *                    parameter can and must be NULL.
 *      arg_fmt -     Printf-format-like string that can be used to specify
 *                    the following sump pump directives:
 *                    -ASCII or -UTF_8    Input records are ascii/utf-8 
 *                                        characters delimited by a newline
 *                                        character.
 *                    -GROUP_BY or -GROUP Group input records for the purpose
 *                                        of reducing them. The sump pump input
 *                                        should be coming from an nsort
 *                                        instance where the "-match"
 *                                        directive has been declared. This
 *                                        directive prevents records with
 *                                        equal keys from being dispersed to
 *                                        more than one sump pump task.
 *                    -IN=%s or -IN_FILE=%s Input file name for the sump pump
 *                                        input.  if not specified, the input
 *                                        should be written into the sump pump
 *                                        either by calls to sp_write_input()
 *                                        or sp_start_link()
 *                                        The input file name can be followed
 *                                        by options that control the file
 *                                        access mode and transfer size.
 *                                        See sp_open_file_src() comments.
 *                    -IN_BUF_SIZE=%d{k,m,g} Overrides default input buffer
 *                                        size (256kb). If a 'k', 'm' or 'g'
 *                                        suffix is specified, the specified
 *                                        size is multiplied by 2^10, 2^20 or
 *                                        2^30 respectively.
 *                    -IN_BUFS=%d         Overrides default number of input
 *                                        buffers (the number of tasks).
 *                    -OUT[%d]=%s or -OUT_FILE[%d]=%s  The output file name for
 *                                        the specified output index, or output
 *                                        0 if no index is specified.  If not 
 *                                        defined, the output should be read
 *                                        either by calls to sp_read_output()
 *                                        or by sp_start_link().
 *                                        The output file name can be followed
 *                                        by options that control the file
 *                                        access mode and transfer size.
 *                                        See sp_open_file_dst() comments.
 *                    -OUT_BUF_SIZE[%d]=%d{x,k,m,g} Overrides default output
 *                                        buffer size (2x the input buf size)
 *                                        for the specified output index, or
 *                                        output 0 if no index is specified.
 *                                        If the size ends with a suffix of
 *                                        'x', the size is used as a multiplier
 *                                        of the input buffer size. If a 'k',
 *                                        'm' or 'g' suffix is specified, the
 *                                        specified size is multiplied by 2^10,
 *                                        2^20 or 2^30 respectively. It is not
 *                                        an error if the output of a task
 *                                        exceeds the output buffer size, but
 *                                        it can potentially result in loss
 *                                        of parallelism.
 *                    -OUTPUTS=%d         Overrides default number of output
 *                                        streams (1).
 *                    -REC_SIZE=%d        Defines the input record size in 
 *                                        bytes. The record contents need not 
 *                                        be ascii nor delimited by a newline
 *                                        character. If not specified, records
 *                                        must consist of ascii or utf-8
 *                                        characters and be terminated by a
 *                                        newline.
 *                    -TASKS=%d           Overrides default number of output
 *                                        tasks (3x the number of threads).
 *                    -THREADS=%d         Overrides default number of threads
 *                                        that are used to execute the pump
 *                                        function in parallel. The default is
 *                                        the number of logical processors in
 *                                        the system.
 *                    -WHOLE or           Processing is not done by input
 *                      -WHOLE_BUF        records so not input record type
 *                                        should be defined.  Instead,
 *                                        processing is done by whole input
 *                                        buffers.
 *                    -DEFAULT_FILE_MODE={BUFFERED,BUF,DIRECT,DIR}  Set the
 *                                        default file access mode for both
 *                                        input and output files.  If none is
 *                                        specified, the direct mode is used
 *                                        to access input and output files for
 *                                        which a BUFFERED file modifier is
 *                                        not specified.
 *                    [external_program_name external_program_arguments]
 *                                        The name of an external program and
 *                                        its arguments. The program name
 *                                        cannot start with the '-' character.
 *                                        The external program name and its 
 *                                        arguments must be last in the
 *                                        arg_fmt string. If the program name
 *                                        does not include a path, the PATH
 *                                        environment variable will be used
 *                                        to find it.
 *      ...           potential subsequent arguments to arg_fmt
 *
 * Returns: SP_OK or a sump pump error code
 *
 * The sump pump directives can also appear in a SUMP_PUMP environment
 * variable.
 */
int sp_start(sp_t *caller_sp,
             sp_pump_t pump_func,
             char *arg_fmt,
             ...)
{
    sp_t                sp;
    char                *s;
    unsigned            i;
    unsigned            j;
    int                 ret;
    char                *p;
    int                 index;
    char                *args;
    char                err_buf[200];
    
    if (TraceFp == NULL &&
        (s = getenv("SUMP_PUMP_TRACE")) != NULL &&
        strlen(s) > 0)
    {
        if (!strcmp(s, "<stdout>"))
            TraceFp = stdout;
        else if (!strcmp(s, "<stderr>"))
            TraceFp = stderr;
        else
        {
            TraceFp = fopen(s, "a+");
            if (TraceFp == NULL)
                fprintf(stderr, "can't open SUMP_PUMP_TRACE=%s\n", s);
            else
            {
                time_t  now = time(NULL);
                trace("begin new sump pump trace, %s", ctime(&now));
            }
        }
    }
    
    *caller_sp = NULL;
    sp = (sp_t)calloc(1, sizeof(struct sump));
    if (sp == NULL)
        return (SP_MEM_ALLOC_ERROR);
    sp->error_buf_size = ERROR_BUF_SIZE;
    sp->error_buf = (char *)calloc(1, sp->error_buf_size);
    if (sp->error_buf == NULL)
        return (SP_MEM_ALLOC_ERROR);
    *caller_sp = sp;
    
    /* fill in default parameters */
    sp->pump_arg = NULL;
    sp->num_threads = get_logical_processor_count(); /* default thread count */
    sp->num_in_bufs = 3 * sp->num_threads;
    sp->num_tasks = 3 * sp->num_threads;
    sp->in_buf_size = (1 << 18);
    sp->num_outputs = 1;
    sp->out = (struct sump_out *)calloc(1, sizeof(struct sump_out));
    sp->out[0].buf_size = (1 << 18);
    sp->delimiter = (void *)"\n";
    sp->rec_size = 0;

    sp->pump_func = pump_func;

    if ((p = getenv("SUMP_PUMP")) == NULL)
    {
        /* allocate minimal string */
        args = (char *)calloc(1, 1);
        args[0] = '\0';
    }
    else
    {
        args = (char *)calloc(strlen(p) + 2, 1);
        memcpy(args, p, strlen(p));
        /* add newline to separate potential following commands */
        args[strlen(p)] = '\n';
        args[strlen(p) + 1] = '\0';
    }

    if (arg_fmt != NULL)
    {
        va_list ap;
        size_t  args_size;

        p = args;
        va_start(ap, arg_fmt);
#if defined(win_nt)
        args_size = _vscprintf(arg_fmt, ap);
#else
        args_size = vsnprintf(NULL, 0, arg_fmt, ap);
#endif
        va_end(ap);
        args = (char *)calloc(strlen(p) + args_size + 1, 1);
        memcpy(args, p, strlen(p));
        va_start(ap, arg_fmt);
        if (vsnprintf(args + strlen(p), args_size + 1, arg_fmt, ap) !=
            args_size)
        {
            start_error(sp, "sp_start: "
                        "vnsprintf failed to return %d\n", args_size);
            return (sp->error_code);
        }
        va_end(ap);
        free(p);   /* free copy of environment string */
        TRACE("sp_start args: '%s'\n", args);
    }

    for (p = args; ; )
    {
        /* ignore leading white space chars */
        while (isspace(*(unsigned char *)p))
            p++;
        if (*p == '\0')
            break;
        if (*p++ != '-')
        {
            /* Must be the name of an external program, and possibly
             * some command line arguments for it.
             */
            p--;
            sp->flags |= SP_EXEC;
#if defined(SUMP_PIPE_STDERR)
            if (sp->num_outputs == 1)
            {
                sp->out = (struct sump_out *)
                    realloc(sp->out, 2 * sizeof(struct sump_out));
                if (sp->out == NULL)
                {
                    start_error(sp, "set_num_outputs, realloc() failed\n");
                    return (sp->error_code);
                }
                memset(sp->out + 1, 0, sizeof(struct sump_out));
                sp->out[1].buf_size = sp->out[0].buf_size;
                sp->num_outputs = 2;
            }
#endif
            get_exec_args(sp, &p);
            if (sp->error_code)
                return (sp->error_code);
        }
        else if (scan("ASCII", &p) || scan("UTF_8", &p))
        {
            sp->flags |= SP_UTF_8;
        }
        else if (scan("DEFAULT_FILE_MODE=", &p))
        {
            if (scan("BUFFERED", &p) || scan("BUF", &p))
                Default_file_mode = MODE_BUFFERED;
            else if (scan("DIRECT", &p) || scan("DIR", &p))
                Default_file_mode = MODE_DIRECT;
            else
                syntax_error(sp, p, "unrecognized file access mode");
        }
        else if (scan("GROUP_BY", &p) || scan("GROUP", &p))
            sp->flags |= SP_GROUP_BY;
        else if (scan("IN_BUFS=", &p))
            sp->num_in_bufs = (unsigned)get_numeric_arg(sp, &p);
        else if (scan("IN_BUF_SIZE=", &p))
        {
            sp->in_buf_size = (ssize_t)get_numeric_arg(sp, &p);
            sp->in_buf_size *= (ssize_t)get_scale(&p);
        }
        else if (scan("IN_FILE=", &p) || scan("IN=", &p))
        {
            /* get input file here */
            sp->in_file = get_string_arg(&p);
            sp->in_file_alloc = TRUE;
        }
        else if (scan("OUT_BUF_SIZE", &p))
        {
            size_t  size;
            double  incr;
                
            if (*p == '=')   /* if no index in square brackets */
            {
                index = 0;   /* default to index 0 */
                p++;
            }
            else
                index = get_output_index(sp, &p);
            if (*p == '.')
                size = 0;
            else
                size = (int)get_numeric_arg(sp, &p);
            if (*p == '.' || *p == 'x' || *p == 'X')
            {
                sp->out[index].buf_size_mult = (double)size;
                if (*p == '.')
                {
                    p++;
                    incr = 0.1;
                    while (*p >= '0' && *p <= '9')
                    {
                        sp->out[index].buf_size_mult +=
                            (*p - '0') * incr;
                        incr *= 0.1;
                        p++;
                    }
                }
                if (*p != 'x' && *p != 'X')
                {
                    start_error(sp, "sp_start: "
                                "out buf size factor must end with 'x'\n");
                    return (sp->error_code);
                }
                p++;
                sp->out[index].size_specified = FALSE;
            }
            else
            {
                sp->out[index].buf_size = size;
                sp->out[index].buf_size *= (size_t)get_scale(&p);
                sp->out[index].size_specified = TRUE;
            }
        }
        else if (scan("OUT_FILE", &p) || scan("OUT", &p))
        {
            if (*p == '=')   /* if no index in square brackets */
            {
                index = 0;   /* default to index 0 */
                p++;
            }
            else
                index = get_output_index(sp, &p);
            if (index >= sp->num_outputs)
            {
                start_error(sp, "sp_start: "
                            "output file index must be less than the "
                            "number of output files\n");
                return (sp->error_code);
            }
            sp->out[index].file = get_string_arg(&p);
            sp->out[index].file_alloc = TRUE;
        }
        else if (scan("OUTPUTS=", &p))
        {
            unsigned        num_outputs;
                
            num_outputs = (unsigned)get_numeric_arg(sp, &p);
            if (num_outputs > sp->num_outputs)
            {
                sp->out = (struct sump_out *)realloc(sp->out,
                                                     num_outputs * sizeof(struct sump_out));
                if (sp->out == NULL)
                {
                    start_error(sp, "set_num_outputs, realloc() failed\n");
                    return (sp->error_code);
                }
                for (i = sp->num_outputs; i < num_outputs; i++)
                {
                    memset(sp->out + i, 0, sizeof(struct sump_out));
                    sp->out[i].buf_size = sp->out[0].buf_size;
                }
            }
            sp->num_outputs = num_outputs;
        }
        else if (scan("WHOLE_BUF", &p) || scan("WHOLE", &p))
        {
            sp->flags &= ~SP_UTF_8;
            sp->flags |= SP_WHOLE_BUF;
        }
        else if (scan("REC_SIZE=", &p))
        {
            sp->rec_size = (int)get_numeric_arg(sp, &p);
            sp->flags &= ~SP_UTF_8;
            sp->flags |= SP_FIXED;
            if (sp->rec_size <= 0)
            {
                start_error(sp, "sp_start: "
                            "REC_SIZE must be greater than 0\n");
                return (sp->error_code);
            }
        }
        else if (scan("RW_TEST_SIZE=", &p))
        {
            Default_rw_test_size = (size_t)get_numeric_arg(sp, &p);
            Default_rw_test_size *= (size_t)get_scale(&p);
        }
        else if (scan("TASKS=", &p))
        {
            sp->num_in_bufs = sp->num_tasks =
                (unsigned)get_numeric_arg(sp, &p);
        }
        else if (scan("THREADS=", &p))
        {
            int     num_threads;
                
            num_threads = (int)get_numeric_arg(sp, &p);
            if (num_threads > 0)
                sp->num_threads = (unsigned)num_threads;
        }
        else
            syntax_error(sp, p, "unrecognized keyword");

        if (sp->error_code)
            return (sp->error_code);
    }
    free(args);

    if (REC_TYPE(sp) == 0)
    {
        start_error(sp, "sp_start: a record type must be specified\n");
        return (sp->error_code);
    }
    if (REC_TYPE(sp) == SP_UTF_8)
    {
        if (strlen((char *)sp->delimiter) > 1)
        {
            start_error(sp, "sp_start: "
                        "currently only single-char delimiters are allowed\n");
            return (sp->error_code);
        }
    }
    else if (REC_TYPE(sp) == SP_UNICODE)
    {
        start_error(sp,
                    "sp_start: unicode records are currently not supported\n");
        return (sp->error_code);
    }
    else if (REC_TYPE(sp) == SP_FIXED)
    {
        /* nothing for now */
    }
    else if (REC_TYPE(sp) == SP_WHOLE_BUF)
    {
        /* nothing for now */
    }
    else
    {
        start_error(sp, "sp_start: multiple record types specified\n");
        return (sp->error_code);
    }

    for (i = 0; i < sp->num_outputs; i++)
    {
        /* if an output buffer absolute size has not been specified
         */
        if (sp->out[i].size_specified == FALSE)
        {
            /* if an output buffer size multiplier was specified, use it */
            if (sp->out[i].buf_size_mult != 0.0)
                sp->out[i].buf_size = (size_t)(sp->in_buf_size *
                                               sp->out[i].buf_size_mult + 0.5);
            else /* default to using 2x the input buffer size */
                sp->out[i].buf_size = 2 * sp->in_buf_size;
        }
        TRACE("out %d: %d\n", i, (int)sp->out[i].buf_size);
    }

    /* alloc rec output buffers for each task */
    sp->task = (sp_task_t)calloc(sp->num_tasks, sizeof(struct sp_task));
    for (i = 0; i < sp->num_tasks; i++)
    {
        sp->task[i].out = (struct task_out *)
            calloc(sp->num_outputs, sizeof (struct task_out));
        sp->task[i].error_buf_size = ERROR_BUF_SIZE;
        sp->task[i].error_buf = (char *)calloc(1, sp->task[i].error_buf_size);
        if (sp->task[i].error_buf == NULL)
            return (SP_MEM_ALLOC_ERROR);
        for (j = 0; j < sp->num_outputs; j++)
        {
            sp->task[i].out[j].buf = (void *)malloc(sp->out[j].buf_size);
            sp->task[i].out[j].size = sp->out[j].buf_size;
        }
        sp->task[i].sp = sp;
    }
    /* alloc input buffers */
    sp->in_buf = (in_buf_t *)calloc(sp->num_in_bufs, sizeof(in_buf_t));
    for (i = 0; i < sp->num_in_bufs; i++)
    {
        size_t  buf_size;

        /* round up buf size to page size multiple */
        buf_size = ((sp->in_buf_size + PAGE_SIZE - 1) / PAGE_SIZE) * PAGE_SIZE;
#if defined(win_nt)
        sp->in_buf[i].in_buf = 
            VirtualAlloc(NULL, buf_size, MEM_COMMIT, PAGE_READWRITE);
        if (sp->in_buf[i].in_buf == NULL)
            return (SP_MEM_ALLOC_ERROR);
#else
        init_zero_fd();
        sp->in_buf[i].in_buf = mmap(NULL, buf_size,
                                    PROT_READ | PROT_WRITE,
                                    MAP_PRIVATE, Zero_fd, 0);
        if (sp->in_buf[i].in_buf == MAP_FAILED)
            return (SP_MEM_ALLOC_ERROR);
#endif
        sp->in_buf[i].in_buf_size = sp->in_buf_size;
        sp->in_buf[i].alloc_size = buf_size;
    }

    if (sp->flags & SP_EXEC)
    {
        if (sp->pump_func != NULL)
        {
            start_error(sp, "can't both define a pump function and external program\n");
            return (sp->error_code);
        }
        sp->ex_state = (struct exec_state *)
            calloc(sizeof(struct exec_state), sp->num_threads);
        /* use own internal pump function to pipe to/from external process */
        sp->pump_func = pfunc_exec;

#if !defined(win_nt)
        /* ignore broken pipe signal, failed write()'s return with error */
        signal(SIGPIPE, SIG_IGN);
#endif
    }
    else if (sp->pump_func == NULL)
    {
        start_error(sp, "an external program or pump function needs to be defined\n");
        return (sp->error_code);
    }

    /* create mutexes and conditions */
    pthread_mutex_init(&sp->sump_mtx, NULL);
    pthread_mutex_init(&sp->sp_mtx, NULL);
    pthread_cond_init(&sp->in_buf_readable_cond, NULL);
    pthread_cond_init(&sp->in_buf_done_cond, NULL);
    pthread_cond_init(&sp->task_avail_cond, NULL);
    pthread_cond_init(&sp->task_drained_cond, NULL);
    pthread_cond_init(&sp->task_output_ready_cond, NULL);
    pthread_cond_init(&sp->task_output_empty_cond, NULL);

    /* create thread sump threads */
    sp->thread = (pthread_t *)calloc(sp->num_threads, sizeof(pthread_t));
    for (i = 0; i < sp->num_threads; i++)
    {
        ret =
            pthread_create(&sp->thread[i], NULL, pump_thread_main, (void *)sp);
        if (ret)
            die("pthread_create() failed: %d\n", ret);
        if (sp->flags & SP_EXEC)
        {
            sp->ex_state[i].in.ex = &sp->ex_state[i];
            sp->ex_state[i].out.ex = &sp->ex_state[i];
#if defined(win_nt)
            sp->ex_state[i].in.rd_h = sp->ex_state[i].in.wr_h =
                INVALID_HANDLE_VALUE;
            sp->ex_state[i].out.rd_h = sp->ex_state[i].out.wr_h = 
                INVALID_HANDLE_VALUE;
#else
            sp->ex_state[i].in.rd_fd = sp->ex_state[i].in.wr_fd = INVALID_FD;
            sp->ex_state[i].out.rd_fd = sp->ex_state[i].out.wr_fd = INVALID_FD;
#endif
#if defined(SUMP_PIPE_STDERR)
            sp->ex_state[i].err.ex = &sp->ex_state[i];
            sp->ex_state[i].err.rd_fd = sp->ex_state[i].err.wr_fd = INVALID_FD;
#endif
        }
    }

    for (i = 0; i < sp->num_outputs; i++)
    {
        if (sp->out[i].file != NULL)
        {
            /* start an output file connection */
            sp->out[i].file_sp = sp_open_file_dst(sp, i, sp->out[i].file);
            if (sp->out[i].file_sp == NULL)
            {
                start_error(sp, "%s: %s\n", sp->out[i].file,
                            get_error_msg(0, err_buf, sizeof(err_buf)));
                return (SP_FILE_OPEN_ERROR);
            }
        }
    }
    
    if (sp->in_file != NULL)
    {
        /* start an input file connection */
        sp->in_file_sp = sp_open_file_src(sp, sp->in_file);
        if (sp->in_file_sp == NULL)
        {
            start_error(sp, "%s: %s\n", sp->in_file,
                        get_error_msg(0, err_buf, sizeof(err_buf)));
            return (SP_FILE_OPEN_ERROR);
        }
    } 
    return (SP_OK);
}


/* sp_get_error_string - return an error message string
 *
 * Returns: a string containing a sump pump error message. The string should
 *          NOT be free()'d and is valid until the passed sp_t is sp_free()'d.
 */
const char *sp_get_error_string(sp_t sp, int error_code)
{
    const char  *err_code_str;

    if (sp != NULL && sp->error_buf[0] != '\0')
        return (sp->error_buf);
    if (error_code == 0 && sp != NULL)
        error_code = sp->error_code;
    switch (error_code)
    {
      case SP_FILE_READ_ERROR:
        err_code_str = "SP_FILE_READ_ERROR: file read error";
        break;

      case SP_FILE_WRITE_ERROR:
        err_code_str = "SP_FILE_WRITE_ERROR: file write error";
        break;

      case SP_UPSTREAM_ERROR:
        err_code_str = "SP_UPSTREAM_ERROR: upstream error";
        break;

      case SP_REDUNDANT_EOF:
        err_code_str = "SP_REDUNDANT_EOF: redundant eof";
        break;

      case SP_MEM_ALLOC_ERROR:
        err_code_str = "SP_MEM_ALLOC_ERROR: memory allocation error";
        break;

      case SP_FILE_OPEN_ERROR:
        err_code_str = "SP_FILE_OPEN_ERROR: file open error";
        break;

      case SP_WRITE_ERROR:
        err_code_str = "SP_WRITE_ERROR: write error";
        break;

      case SP_SORT_DEF_ERROR:
        err_code_str = "SP_SORT_DEF_ERROR: sort definition error";
        break;

      case SP_SORT_EXEC_ERROR:
        err_code_str = "SP_SORT_EXEC_ERROR: sort execution error";
        break;

      case SP_GROUP_BY_MISMATCH:
        err_code_str = "SP_GROUP_BY_MISMATCH: group-by mismatch";
        break;

      case SP_SORT_INCOMPATIBLE:
        err_code_str = "SP_SORT_INCOMPATIBLE: a sort sump pump is incompatible"
            " with buffer-at-a-time i/o, specify the input or"
            " output file as a sort parameter instead";
        break;

      case SP_BUF_INDEX_ERROR:
        err_code_str = "SP_BUF_INDEX_ERROR: the specified buffer index is"
            " out-of-range";
        break;

      case SP_OUTPUT_INDEX_ERROR:
        err_code_str = "SP_OUTPUT_INDEX_ERROR: the specified output index is"
            " equal to or larger than the number of outputs";
        break;

      case SP_START_ERROR:
        err_code_str = "SP_START_ERROR: an error occured during sp_start()";
        break;

      case SP_NSORT_LINK_FAILURE:
        err_code_str =
            "SP_NSORT_LINK_FAILURE: link attempt to nsort library failed. "
            "Is nsort installed?";
        break;

      case SP_SORT_NOT_COMPILED:
        err_code_str =
            "SP_SORT_NOT_COMPILED: sump.c has not been compiled to support "
            "sorting";
        break;

      case SP_PUMP_FUNCTION_ERROR:
        err_code_str = "Pump function error";
        break;

      default:
        err_code_str = "Unknown Error";
    }
    return (err_code_str);
}


/* sp_free - free the specified sp_t and its associated state.
 */
void sp_free(sp_t *caller_sp)
{
    sp_t        sp;
    int         i, j;

    sp = *caller_sp;
    if (sp == NULL)
        return;

    sp_wait(sp); /* make sure sump pump has finished */
    
#if !defined(SUMP_PUMP_NO_SORT)
    if (sp->flags & SP_SORT)
    {
        if (sp->sort_temp_buf != NULL)
            free(sp->sort_temp_buf);

        if (sp->nsort_ctx != 0)
            (*Nsort_end)(&sp->nsort_ctx);
    
        if (sp->out != NULL)
            free(sp->out);
    }
    else
#endif
    {
        if (sp->task != NULL)
        {
            for (i = 0; i < sp->num_tasks; i++)
            {
                if (sp->task[i].out != NULL)
                {
                    for (j = 0; j < sp->num_outputs; j++)
                        if (sp->task[i].out[j].buf != NULL)
                            free(sp->task[i].out[j].buf);
                    free(sp->task[i].out);
                }

                if (sp->task[i].error_buf != NULL)
                    free(sp->task[i].error_buf);
            }
            free(sp->task);
        }
        if (sp->in_buf != NULL)
        {
            for (i = 0; i < sp->num_in_bufs; i++)
            {
                if (sp->in_buf[i].in_buf != NULL)
                {
#if defined(win_nt)
                    VirtualFree(sp->in_buf[i].in_buf,
                                sp->in_buf[i].alloc_size, MEM_RELEASE);
#else
                    munmap(sp->in_buf[i].in_buf, sp->in_buf[i].alloc_size);
#endif
                }
            }
            free(sp->in_buf);
        }
        if (sp->ex_state != NULL)
            free(sp->ex_state);

        if (sp->thread != NULL)
        {
            pthread_mutex_destroy(&sp->sump_mtx);
            pthread_mutex_destroy(&sp->sp_mtx);
            pthread_cond_destroy(&sp->in_buf_readable_cond);
            pthread_cond_destroy(&sp->in_buf_done_cond);
            pthread_cond_destroy(&sp->task_avail_cond);
            pthread_cond_destroy(&sp->task_drained_cond);
            pthread_cond_destroy(&sp->task_output_ready_cond);
            pthread_cond_destroy(&sp->task_output_empty_cond);
            free(sp->thread);
        }
    
        if (sp->out != NULL)
        {
            for (i = 0; i < sp->num_outputs; i++)
            {
                if (sp->out[i].file_sp != NULL)
                    sp_file_free(&sp->out[i].file_sp);
                if (sp->out[i].file_alloc && sp->out[i].file != NULL)
                    free(sp->out[i].file);
            }
            free(sp->out);
        }

        if (sp->in_file_sp != NULL)
            sp_file_free(&sp->in_file_sp);
        if (sp->in_file_alloc && sp->in_file != NULL)
            free(sp->in_file);
    }
    if (sp->error_buf != NULL)
        free(sp->error_buf);

    free(sp);
    *caller_sp = NULL;
}


/* sp_file_free - free the specified sp_file_t and its associated state.
 */
void sp_file_free(sp_file_t *caller_sp_file)
{
    sp_file_t sp_file;

    sp_file = *caller_sp_file;
    if (sp_file == NULL)
        return;

    if (sp_file->fname != NULL)
        free(sp_file->fname);
    
    free(sp_file);
    
    *caller_sp_file = NULL;
}