/* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

#include "nspr.h"

/*
 * Thread pools
 *  Thread pools create and manage threads to provide support for
 *  scheduling jobs onto one or more threads.
 *
 */
#ifdef OPT_WINNT
#include <windows.h>
#endif

/*
 * worker thread
 */
typedef struct wthread {
    PRCList     links;
    PRThread    *thread;
} wthread;

/*
 * queue of timer jobs
 */
typedef struct timer_jobq {
    PRCList     list;
    PRLock      *lock;
    PRCondVar   *cv;
    PRInt32     cnt;
    PRCList     wthreads;
} timer_jobq;

/*
 * queue of jobs
 */
typedef struct tp_jobq {
    PRCList     list;
    PRInt32     cnt;
    PRLock      *lock;
    PRCondVar   *cv;
    PRCList     wthreads;
#ifdef OPT_WINNT
    HANDLE      nt_completion_port;
#endif
} tp_jobq;

/*
 * queue of IO jobs
 */
typedef struct io_jobq {
    PRCList     list;
    PRPollDesc  *pollfds;
    PRInt32     npollfds;
    PRJob       **polljobs;
    PRLock      *lock;
    PRInt32     cnt;
    PRFileDesc  *notify_fd;
    PRCList     wthreads;
} io_jobq;

/*
 * Threadpool
 */
struct PRThreadPool {
    PRInt32     init_threads;
    PRInt32     max_threads;
    PRInt32     current_threads;
    PRInt32     idle_threads;
    PRUint32    stacksize;
    tp_jobq     jobq;
    io_jobq     ioq;
    timer_jobq  timerq;
    PRLock      *join_lock;     /* used with jobp->join_cv */
    PRCondVar   *shutdown_cv;
    PRBool      shutdown;
};

typedef enum io_op_type
{ JOB_IO_READ, JOB_IO_WRITE, JOB_IO_CONNECT, JOB_IO_ACCEPT } io_op_type;

#ifdef OPT_WINNT
typedef struct NT_notifier {
    OVERLAPPED overlapped;      /* must be first */
    PRJob   *jobp;
} NT_notifier;
#endif

struct PRJob {
    PRCList         links;      /*  for linking jobs */
    PRBool          on_ioq;     /* job on ioq */
    PRBool          on_timerq;  /* job on timerq */
    PRJobFn         job_func;
    void            *job_arg;
    PRCondVar       *join_cv;
    PRBool          join_wait;  /* == PR_TRUE, when waiting to join */
    PRCondVar       *cancel_cv; /* for cancelling IO jobs */
    PRBool          cancel_io;  /* for cancelling IO jobs */
    PRThreadPool    *tpool;     /* back pointer to thread pool */
    PRJobIoDesc     *iod;
    io_op_type      io_op;
    PRInt16         io_poll_flags;
    PRNetAddr       *netaddr;
    PRIntervalTime  timeout;    /* relative value */
    PRIntervalTime  absolute;
#ifdef OPT_WINNT
    NT_notifier     nt_notifier;
#endif
};

#define JOB_LINKS_PTR(_qp) \
    ((PRJob *) ((char *) (_qp) - offsetof(PRJob, links)))

#define WTHREAD_LINKS_PTR(_qp) \
    ((wthread *) ((char *) (_qp) - offsetof(wthread, links)))

#define JOINABLE_JOB(_jobp) (NULL != (_jobp)->join_cv)

#define JOIN_NOTIFY(_jobp)                              \
                PR_BEGIN_MACRO                          \
                PR_Lock(_jobp->tpool->join_lock);       \
                _jobp->join_wait = PR_FALSE;            \
                PR_NotifyCondVar(_jobp->join_cv);       \
                PR_Unlock(_jobp->tpool->join_lock);     \
                PR_END_MACRO

#define CANCEL_IO_JOB(jobp)                             \
                PR_BEGIN_MACRO                          \
                jobp->cancel_io = PR_FALSE;             \
                jobp->on_ioq = PR_FALSE;                \
                PR_REMOVE_AND_INIT_LINK(&jobp->links);  \
                tp->ioq.cnt--;                          \
                PR_NotifyCondVar(jobp->cancel_cv);      \
                PR_END_MACRO

static void delete_job(PRJob *jobp);
static PRThreadPool * alloc_threadpool(void);
static PRJob * alloc_job(PRBool joinable, PRThreadPool *tp);
static void notify_ioq(PRThreadPool *tp);
static void notify_timerq(PRThreadPool *tp);

/*
 * locks are acquired in the following order
 *
 *  tp->ioq.lock,tp->timerq.lock
 *          |
 *          V
 *      tp->jobq->lock
 */

/*
 * worker thread function
 */
static void wstart(void *arg)
{
    PRThreadPool *tp = (PRThreadPool *) arg;
    PRCList *head;

    /*
     * execute jobs until shutdown
     */
    while (!tp->shutdown) {
        PRJob *jobp;
#ifdef OPT_WINNT
        BOOL rv;
        DWORD unused, shutdown;
        LPOVERLAPPED olp;

        PR_Lock(tp->jobq.lock);
        tp->idle_threads++;
        PR_Unlock(tp->jobq.lock);
        rv = GetQueuedCompletionStatus(tp->jobq.nt_completion_port,
                                       &unused, &shutdown, &olp, INFINITE);

        PR_ASSERT(rv);
        if (shutdown) {
            break;
        }
        jobp = ((NT_notifier *) olp)->jobp;
        PR_Lock(tp->jobq.lock);
        tp->idle_threads--;
        tp->jobq.cnt--;
        PR_Unlock(tp->jobq.lock);
#else

        PR_Lock(tp->jobq.lock);
        while (PR_CLIST_IS_EMPTY(&tp->jobq.list) && (!tp->shutdown)) {
            tp->idle_threads++;
            PR_WaitCondVar(tp->jobq.cv, PR_INTERVAL_NO_TIMEOUT);
            tp->idle_threads--;
        }
        if (tp->shutdown) {
            PR_Unlock(tp->jobq.lock);
            break;
        }
        head = PR_LIST_HEAD(&tp->jobq.list);
        /*
         * remove job from queue
         */
        PR_REMOVE_AND_INIT_LINK(head);
        tp->jobq.cnt--;
        jobp = JOB_LINKS_PTR(head);
        PR_Unlock(tp->jobq.lock);
#endif

        jobp->job_func(jobp->job_arg);
        if (!JOINABLE_JOB(jobp)) {
            delete_job(jobp);
        } else {
            JOIN_NOTIFY(jobp);
        }
    }
    PR_Lock(tp->jobq.lock);
    tp->current_threads--;
    PR_Unlock(tp->jobq.lock);
}

/*
 * add a job to the work queue
 */
static void
add_to_jobq(PRThreadPool *tp, PRJob *jobp)
{
    /*
     * add to jobq
     */
#ifdef OPT_WINNT
    PR_Lock(tp->jobq.lock);
    tp->jobq.cnt++;
    PR_Unlock(tp->jobq.lock);
    /*
     * notify worker thread(s)
     */
    PostQueuedCompletionStatus(tp->jobq.nt_completion_port, 0,
                               FALSE, &jobp->nt_notifier.overlapped);
#else
    PR_Lock(tp->jobq.lock);
    PR_APPEND_LINK(&jobp->links,&tp->jobq.list);
    tp->jobq.cnt++;
    if ((tp->idle_threads < tp->jobq.cnt) &&
        (tp->current_threads < tp->max_threads)) {
        wthread *wthrp;
        /*
         * increment thread count and unlock the jobq lock
         */
        tp->current_threads++;
        PR_Unlock(tp->jobq.lock);
        /* create new worker thread */
        wthrp = PR_NEWZAP(wthread);
        if (wthrp) {
            wthrp->thread = PR_CreateThread(PR_USER_THREAD, wstart,
                                            tp, PR_PRIORITY_NORMAL,
                                            PR_GLOBAL_THREAD,PR_JOINABLE_THREAD,tp->stacksize);
            if (NULL == wthrp->thread) {
                PR_DELETE(wthrp);  /* this sets wthrp to NULL */
            }
        }
        PR_Lock(tp->jobq.lock);
        if (NULL == wthrp) {
            tp->current_threads--;
        } else {
            PR_APPEND_LINK(&wthrp->links, &tp->jobq.wthreads);
        }
    }
    /*
     * wakeup a worker thread
     */
    PR_NotifyCondVar(tp->jobq.cv);
    PR_Unlock(tp->jobq.lock);
#endif
}

/*
 * io worker thread function
 */
static void io_wstart(void *arg)
{
    PRThreadPool *tp = (PRThreadPool *) arg;
    int pollfd_cnt, pollfds_used;
    int rv;
    PRCList *qp, *nextqp;
    PRPollDesc *pollfds = NULL;
    PRJob **polljobs = NULL;
    int poll_timeout;
    PRIntervalTime now;

    /*
     * scan io_jobq
     * construct poll list
     * call PR_Poll
     * for all fds, for which poll returns true, move the job to
     * jobq and wakeup worker thread.
     */
    while (!tp->shutdown) {
        PRJob *jobp;

        pollfd_cnt = tp->ioq.cnt + 10;
        if (pollfd_cnt > tp->ioq.npollfds) {

            /*
             * re-allocate pollfd array if the current one is not large
             * enough
             */
            if (NULL != tp->ioq.pollfds) {
                PR_Free(tp->ioq.pollfds);
            }
            tp->ioq.pollfds = (PRPollDesc *) PR_Malloc(pollfd_cnt *
                              (sizeof(PRPollDesc) + sizeof(PRJob *)));
            PR_ASSERT(NULL != tp->ioq.pollfds);
            /*
             * array of pollfds
             */
            pollfds = tp->ioq.pollfds;
            tp->ioq.polljobs = (PRJob **) (&tp->ioq.pollfds[pollfd_cnt]);
            /*
             * parallel array of jobs
             */
            polljobs = tp->ioq.polljobs;
            tp->ioq.npollfds = pollfd_cnt;
        }

        pollfds_used = 0;
        /*
         * add the notify fd; used for unblocking io thread(s)
         */
        pollfds[pollfds_used].fd = tp->ioq.notify_fd;
        pollfds[pollfds_used].in_flags = PR_POLL_READ;
        pollfds[pollfds_used].out_flags = 0;
        polljobs[pollfds_used] = NULL;
        pollfds_used++;
        /*
         * fill in the pollfd array
         */
        PR_Lock(tp->ioq.lock);
        for (qp = tp->ioq.list.next; qp != &tp->ioq.list; qp = nextqp) {
            nextqp = qp->next;
            jobp = JOB_LINKS_PTR(qp);
            if (jobp->cancel_io) {
                CANCEL_IO_JOB(jobp);
                continue;
            }
            if (pollfds_used == (pollfd_cnt)) {
                break;
            }
            pollfds[pollfds_used].fd = jobp->iod->socket;
            pollfds[pollfds_used].in_flags = jobp->io_poll_flags;
            pollfds[pollfds_used].out_flags = 0;
            polljobs[pollfds_used] = jobp;

            pollfds_used++;
        }
        if (!PR_CLIST_IS_EMPTY(&tp->ioq.list)) {
            qp = tp->ioq.list.next;
            jobp = JOB_LINKS_PTR(qp);
            if (PR_INTERVAL_NO_TIMEOUT == jobp->timeout) {
                poll_timeout = PR_INTERVAL_NO_TIMEOUT;
            }
            else if (PR_INTERVAL_NO_WAIT == jobp->timeout) {
                poll_timeout = PR_INTERVAL_NO_WAIT;
            }
            else {
                poll_timeout = jobp->absolute - PR_IntervalNow();
                if (poll_timeout <= 0) { /* already timed out */
                    poll_timeout = PR_INTERVAL_NO_WAIT;
                }
            }
        } else {
            poll_timeout = PR_INTERVAL_NO_TIMEOUT;
        }
        PR_Unlock(tp->ioq.lock);

        /*
         * XXXX
         * should retry if more jobs have been added to the queue?
         *
         */
        PR_ASSERT(pollfds_used <= pollfd_cnt);
        rv = PR_Poll(tp->ioq.pollfds, pollfds_used, poll_timeout);

        if (tp->shutdown) {
            break;
        }

        if (rv > 0) {
            /*
             * at least one io event is set
             */
            PRStatus rval_status;
            PRInt32 index;

            PR_ASSERT(pollfds[0].fd == tp->ioq.notify_fd);
            /*
             * reset the pollable event, if notified
             */
            if (pollfds[0].out_flags & PR_POLL_READ) {
                rval_status = PR_WaitForPollableEvent(tp->ioq.notify_fd);
                PR_ASSERT(PR_SUCCESS == rval_status);
            }

            for(index = 1; index < (pollfds_used); index++) {
                PRInt16 events = pollfds[index].in_flags;
                PRInt16 revents = pollfds[index].out_flags;
                jobp = polljobs[index];

                if ((revents & PR_POLL_NVAL) ||  /* busted in all cases */
                    (revents & PR_POLL_ERR) ||
                    ((events & PR_POLL_WRITE) &&
                     (revents & PR_POLL_HUP))) { /* write op & hup */
                    PR_Lock(tp->ioq.lock);
                    if (jobp->cancel_io) {
                        CANCEL_IO_JOB(jobp);
                        PR_Unlock(tp->ioq.lock);
                        continue;
                    }
                    PR_REMOVE_AND_INIT_LINK(&jobp->links);
                    tp->ioq.cnt--;
                    jobp->on_ioq = PR_FALSE;
                    PR_Unlock(tp->ioq.lock);

                    /* set error */
                    if (PR_POLL_NVAL & revents) {
                        jobp->iod->error = PR_BAD_DESCRIPTOR_ERROR;
                    }
                    else if (PR_POLL_HUP & revents) {
                        jobp->iod->error = PR_CONNECT_RESET_ERROR;
                    }
                    else {
                        jobp->iod->error = PR_IO_ERROR;
                    }

                    /*
                     * add to jobq
                     */
                    add_to_jobq(tp, jobp);
                } else if (revents) {
                    /*
                     * add to jobq
                     */
                    PR_Lock(tp->ioq.lock);
                    if (jobp->cancel_io) {
                        CANCEL_IO_JOB(jobp);
                        PR_Unlock(tp->ioq.lock);
                        continue;
                    }
                    PR_REMOVE_AND_INIT_LINK(&jobp->links);
                    tp->ioq.cnt--;
                    jobp->on_ioq = PR_FALSE;
                    PR_Unlock(tp->ioq.lock);

                    if (jobp->io_op == JOB_IO_CONNECT) {
                        if (PR_GetConnectStatus(&pollfds[index]) == PR_SUCCESS) {
                            jobp->iod->error = 0;
                        }
                        else {
                            jobp->iod->error = PR_GetError();
                        }
                    } else {
                        jobp->iod->error = 0;
                    }

                    add_to_jobq(tp, jobp);
                }
            }
        }
        /*
         * timeout processing
         */
        now = PR_IntervalNow();
        PR_Lock(tp->ioq.lock);
        for (qp = tp->ioq.list.next; qp != &tp->ioq.list; qp = nextqp) {
            nextqp = qp->next;
            jobp = JOB_LINKS_PTR(qp);
            if (jobp->cancel_io) {
                CANCEL_IO_JOB(jobp);
                continue;
            }
            if (PR_INTERVAL_NO_TIMEOUT == jobp->timeout) {
                break;
            }
            if ((PR_INTERVAL_NO_WAIT != jobp->timeout) &&
                ((PRInt32)(jobp->absolute - now) > 0)) {
                break;
            }
            PR_REMOVE_AND_INIT_LINK(&jobp->links);
            tp->ioq.cnt--;
            jobp->on_ioq = PR_FALSE;
            jobp->iod->error = PR_IO_TIMEOUT_ERROR;
            add_to_jobq(tp, jobp);
        }
        PR_Unlock(tp->ioq.lock);
    }
}

/*
 * timer worker thread function
 */
static void timer_wstart(void *arg)
{
    PRThreadPool *tp = (PRThreadPool *) arg;
    PRCList *qp;
    PRIntervalTime timeout;
    PRIntervalTime now;

    /*
     * call PR_WaitCondVar with minimum value of all timeouts
     */
    while (!tp->shutdown) {
        PRJob *jobp;

        PR_Lock(tp->timerq.lock);
        if (PR_CLIST_IS_EMPTY(&tp->timerq.list)) {
            timeout = PR_INTERVAL_NO_TIMEOUT;
        } else {
            PRCList *qp;

            qp = tp->timerq.list.next;
            jobp = JOB_LINKS_PTR(qp);

            timeout = jobp->absolute - PR_IntervalNow();
            if (timeout <= 0) {
                timeout = PR_INTERVAL_NO_WAIT;    /* already timed out */
            }
        }
        if (PR_INTERVAL_NO_WAIT != timeout) {
            PR_WaitCondVar(tp->timerq.cv, timeout);
        }
        if (tp->shutdown) {
            PR_Unlock(tp->timerq.lock);
            break;
        }
        /*
         * move expired-timer jobs to jobq
         */
        now = PR_IntervalNow();
        while (!PR_CLIST_IS_EMPTY(&tp->timerq.list)) {
            qp = tp->timerq.list.next;
            jobp = JOB_LINKS_PTR(qp);

            if ((PRInt32)(jobp->absolute - now) > 0) {
                break;
            }
            /*
             * job timed out
             */
            PR_REMOVE_AND_INIT_LINK(&jobp->links);
            tp->timerq.cnt--;
            jobp->on_timerq = PR_FALSE;
            add_to_jobq(tp, jobp);
        }
        PR_Unlock(tp->timerq.lock);
    }
}

static void
delete_threadpool(PRThreadPool *tp)
{
    if (NULL != tp) {
        if (NULL != tp->shutdown_cv) {
            PR_DestroyCondVar(tp->shutdown_cv);
        }
        if (NULL != tp->jobq.cv) {
            PR_DestroyCondVar(tp->jobq.cv);
        }
        if (NULL != tp->jobq.lock) {
            PR_DestroyLock(tp->jobq.lock);
        }
        if (NULL != tp->join_lock) {
            PR_DestroyLock(tp->join_lock);
        }
#ifdef OPT_WINNT
        if (NULL != tp->jobq.nt_completion_port) {
            CloseHandle(tp->jobq.nt_completion_port);
        }
#endif
        /* Timer queue */
        if (NULL != tp->timerq.cv) {
            PR_DestroyCondVar(tp->timerq.cv);
        }
        if (NULL != tp->timerq.lock) {
            PR_DestroyLock(tp->timerq.lock);
        }

        if (NULL != tp->ioq.lock) {
            PR_DestroyLock(tp->ioq.lock);
        }
        if (NULL != tp->ioq.pollfds) {
            PR_Free(tp->ioq.pollfds);
        }
        if (NULL != tp->ioq.notify_fd) {
            PR_DestroyPollableEvent(tp->ioq.notify_fd);
        }
        PR_Free(tp);
    }
    return;
}

static PRThreadPool *
alloc_threadpool(void)
{
    PRThreadPool *tp;

    tp = (PRThreadPool *) PR_CALLOC(sizeof(*tp));
    if (NULL == tp) {
        goto failed;
    }
    tp->jobq.lock = PR_NewLock();
    if (NULL == tp->jobq.lock) {
        goto failed;
    }
    tp->jobq.cv = PR_NewCondVar(tp->jobq.lock);
    if (NULL == tp->jobq.cv) {
        goto failed;
    }
    tp->join_lock = PR_NewLock();
    if (NULL == tp->join_lock) {
        goto failed;
    }
#ifdef OPT_WINNT
    tp->jobq.nt_completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE,
                                  NULL, 0, 0);
    if (NULL == tp->jobq.nt_completion_port) {
        goto failed;
    }
#endif

    tp->ioq.lock = PR_NewLock();
    if (NULL == tp->ioq.lock) {
        goto failed;
    }

    /* Timer queue */

    tp->timerq.lock = PR_NewLock();
    if (NULL == tp->timerq.lock) {
        goto failed;
    }
    tp->timerq.cv = PR_NewCondVar(tp->timerq.lock);
    if (NULL == tp->timerq.cv) {
        goto failed;
    }

    tp->shutdown_cv = PR_NewCondVar(tp->jobq.lock);
    if (NULL == tp->shutdown_cv) {
        goto failed;
    }
    tp->ioq.notify_fd = PR_NewPollableEvent();
    if (NULL == tp->ioq.notify_fd) {
        goto failed;
    }
    return tp;
failed:
    delete_threadpool(tp);
    PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
    return NULL;
}

/* Create thread pool */
PR_IMPLEMENT(PRThreadPool *)
PR_CreateThreadPool(PRInt32 initial_threads, PRInt32 max_threads,
                    PRUint32 stacksize)
{
    PRThreadPool *tp;
    PRThread *thr;
    int i;
    wthread *wthrp;

    tp = alloc_threadpool();
    if (NULL == tp) {
        return NULL;
    }

    tp->init_threads = initial_threads;
    tp->max_threads = max_threads;
    tp->stacksize = stacksize;
    PR_INIT_CLIST(&tp->jobq.list);
    PR_INIT_CLIST(&tp->ioq.list);
    PR_INIT_CLIST(&tp->timerq.list);
    PR_INIT_CLIST(&tp->jobq.wthreads);
    PR_INIT_CLIST(&tp->ioq.wthreads);
    PR_INIT_CLIST(&tp->timerq.wthreads);
    tp->shutdown = PR_FALSE;

    PR_Lock(tp->jobq.lock);
    for(i=0; i < initial_threads; ++i) {

        thr = PR_CreateThread(PR_USER_THREAD, wstart,
                              tp, PR_PRIORITY_NORMAL,
                              PR_GLOBAL_THREAD, PR_JOINABLE_THREAD,stacksize);
        PR_ASSERT(thr);
        wthrp = PR_NEWZAP(wthread);
        PR_ASSERT(wthrp);
        wthrp->thread = thr;
        PR_APPEND_LINK(&wthrp->links, &tp->jobq.wthreads);
    }
    tp->current_threads = initial_threads;

    thr = PR_CreateThread(PR_USER_THREAD, io_wstart,
                          tp, PR_PRIORITY_NORMAL,
                          PR_GLOBAL_THREAD,PR_JOINABLE_THREAD,stacksize);
    PR_ASSERT(thr);
    wthrp = PR_NEWZAP(wthread);
    PR_ASSERT(wthrp);
    wthrp->thread = thr;
    PR_APPEND_LINK(&wthrp->links, &tp->ioq.wthreads);

    thr = PR_CreateThread(PR_USER_THREAD, timer_wstart,
                          tp, PR_PRIORITY_NORMAL,
                          PR_GLOBAL_THREAD,PR_JOINABLE_THREAD,stacksize);
    PR_ASSERT(thr);
    wthrp = PR_NEWZAP(wthread);
    PR_ASSERT(wthrp);
    wthrp->thread = thr;
    PR_APPEND_LINK(&wthrp->links, &tp->timerq.wthreads);

    PR_Unlock(tp->jobq.lock);
    return tp;
}

static void
delete_job(PRJob *jobp)
{
    if (NULL != jobp) {
        if (NULL != jobp->join_cv) {
            PR_DestroyCondVar(jobp->join_cv);
            jobp->join_cv = NULL;
        }
        if (NULL != jobp->cancel_cv) {
            PR_DestroyCondVar(jobp->cancel_cv);
            jobp->cancel_cv = NULL;
        }
        PR_DELETE(jobp);
    }
}

static PRJob *
alloc_job(PRBool joinable, PRThreadPool *tp)
{
    PRJob *jobp;

    jobp = PR_NEWZAP(PRJob);
    if (NULL == jobp) {
        goto failed;
    }
    if (joinable) {
        jobp->join_cv = PR_NewCondVar(tp->join_lock);
        jobp->join_wait = PR_TRUE;
        if (NULL == jobp->join_cv) {
            goto failed;
        }
    } else {
        jobp->join_cv = NULL;
    }
#ifdef OPT_WINNT
    jobp->nt_notifier.jobp = jobp;
#endif
    return jobp;
failed:
    delete_job(jobp);
    PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
    return NULL;
}

/* queue a job */
PR_IMPLEMENT(PRJob *)
PR_QueueJob(PRThreadPool *tpool, PRJobFn fn, void *arg, PRBool joinable)
{
    PRJob *jobp;

    jobp = alloc_job(joinable, tpool);
    if (NULL == jobp) {
        return NULL;
    }

    jobp->job_func = fn;
    jobp->job_arg = arg;
    jobp->tpool = tpool;

    add_to_jobq(tpool, jobp);
    return jobp;
}

/* queue a job, when a socket is readable or writeable */
static PRJob *
queue_io_job(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn, void * arg,
             PRBool joinable, io_op_type op)
{
    PRJob *jobp;
    PRIntervalTime now;

    jobp = alloc_job(joinable, tpool);
    if (NULL == jobp) {
        return NULL;
    }

    /*
     * Add a new job to io_jobq
     * wakeup io worker thread
     */

    jobp->job_func = fn;
    jobp->job_arg = arg;
    jobp->tpool = tpool;
    jobp->iod = iod;
    if (JOB_IO_READ == op) {
        jobp->io_op = JOB_IO_READ;
        jobp->io_poll_flags = PR_POLL_READ;
    } else if (JOB_IO_WRITE == op) {
        jobp->io_op = JOB_IO_WRITE;
        jobp->io_poll_flags = PR_POLL_WRITE;
    } else if (JOB_IO_ACCEPT == op) {
        jobp->io_op = JOB_IO_ACCEPT;
        jobp->io_poll_flags = PR_POLL_READ;
    } else if (JOB_IO_CONNECT == op) {
        jobp->io_op = JOB_IO_CONNECT;
        jobp->io_poll_flags = PR_POLL_WRITE|PR_POLL_EXCEPT;
    } else {
        delete_job(jobp);
        PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
        return NULL;
    }

    jobp->timeout = iod->timeout;
    if ((PR_INTERVAL_NO_TIMEOUT == iod->timeout) ||
        (PR_INTERVAL_NO_WAIT == iod->timeout)) {
        jobp->absolute = iod->timeout;
    } else {
        now = PR_IntervalNow();
        jobp->absolute = now + iod->timeout;
    }


    PR_Lock(tpool->ioq.lock);

    if (PR_CLIST_IS_EMPTY(&tpool->ioq.list) ||
        (PR_INTERVAL_NO_TIMEOUT == iod->timeout)) {
        PR_APPEND_LINK(&jobp->links,&tpool->ioq.list);
    } else if (PR_INTERVAL_NO_WAIT == iod->timeout) {
        PR_INSERT_LINK(&jobp->links,&tpool->ioq.list);
    } else {
        PRCList *qp;
        PRJob *tmp_jobp;
        /*
         * insert into the timeout-sorted ioq
         */
        for (qp = tpool->ioq.list.prev; qp != &tpool->ioq.list;
             qp = qp->prev) {
            tmp_jobp = JOB_LINKS_PTR(qp);
            if ((PRInt32)(jobp->absolute - tmp_jobp->absolute) >= 0) {
                break;
            }
        }
        PR_INSERT_AFTER(&jobp->links,qp);
    }

    jobp->on_ioq = PR_TRUE;
    tpool->ioq.cnt++;
    /*
     * notify io worker thread(s)
     */
    PR_Unlock(tpool->ioq.lock);
    notify_ioq(tpool);
    return jobp;
}

/* queue a job, when a socket is readable */
PR_IMPLEMENT(PRJob *)
PR_QueueJob_Read(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn, void * arg,
                 PRBool joinable)
{
    return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_READ));
}

/* queue a job, when a socket is writeable */
PR_IMPLEMENT(PRJob *)
PR_QueueJob_Write(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn,void * arg,
                  PRBool joinable)
{
    return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_WRITE));
}


/* queue a job, when a socket has a pending connection */
PR_IMPLEMENT(PRJob *)
PR_QueueJob_Accept(PRThreadPool *tpool, PRJobIoDesc *iod, PRJobFn fn,
                   void * arg, PRBool joinable)
{
    return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_ACCEPT));
}

/* queue a job, when a socket can be connected */
PR_IMPLEMENT(PRJob *)
PR_QueueJob_Connect(PRThreadPool *tpool, PRJobIoDesc *iod,
                    const PRNetAddr *addr, PRJobFn fn, void * arg, PRBool joinable)
{
    PRStatus rv;
    PRErrorCode err;

    rv = PR_Connect(iod->socket, addr, PR_INTERVAL_NO_WAIT);
    if ((rv == PR_FAILURE) && ((err = PR_GetError()) == PR_IN_PROGRESS_ERROR)) {
        /* connection pending */
        return(queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_CONNECT));
    }
    /*
     * connection succeeded or failed; add to jobq right away
     */
    if (rv == PR_FAILURE) {
        iod->error = err;
    }
    else {
        iod->error = 0;
    }
    return(PR_QueueJob(tpool, fn, arg, joinable));

}

/* queue a job, when a timer expires */
PR_IMPLEMENT(PRJob *)
PR_QueueJob_Timer(PRThreadPool *tpool, PRIntervalTime timeout,
                  PRJobFn fn, void * arg, PRBool joinable)
{
    PRIntervalTime now;
    PRJob *jobp;

    if (PR_INTERVAL_NO_TIMEOUT == timeout) {
        PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
        return NULL;
    }
    if (PR_INTERVAL_NO_WAIT == timeout) {
        /*
         * no waiting; add to jobq right away
         */
        return(PR_QueueJob(tpool, fn, arg, joinable));
    }
    jobp = alloc_job(joinable, tpool);
    if (NULL == jobp) {
        return NULL;
    }

    /*
     * Add a new job to timer_jobq
     * wakeup timer worker thread
     */

    jobp->job_func = fn;
    jobp->job_arg = arg;
    jobp->tpool = tpool;
    jobp->timeout = timeout;

    now = PR_IntervalNow();
    jobp->absolute = now + timeout;


    PR_Lock(tpool->timerq.lock);
    jobp->on_timerq = PR_TRUE;
    if (PR_CLIST_IS_EMPTY(&tpool->timerq.list)) {
        PR_APPEND_LINK(&jobp->links,&tpool->timerq.list);
    }
    else {
        PRCList *qp;
        PRJob *tmp_jobp;
        /*
         * insert into the sorted timer jobq
         */
        for (qp = tpool->timerq.list.prev; qp != &tpool->timerq.list;
             qp = qp->prev) {
            tmp_jobp = JOB_LINKS_PTR(qp);
            if ((PRInt32)(jobp->absolute - tmp_jobp->absolute) >= 0) {
                break;
            }
        }
        PR_INSERT_AFTER(&jobp->links,qp);
    }
    tpool->timerq.cnt++;
    /*
     * notify timer worker thread(s)
     */
    notify_timerq(tpool);
    PR_Unlock(tpool->timerq.lock);
    return jobp;
}

static void
notify_timerq(PRThreadPool *tp)
{
    /*
     * wakeup the timer thread(s)
     */
    PR_NotifyCondVar(tp->timerq.cv);
}

static void
notify_ioq(PRThreadPool *tp)
{
    PRStatus rval_status;

    /*
     * wakeup the io thread(s)
     */
    rval_status = PR_SetPollableEvent(tp->ioq.notify_fd);
    PR_ASSERT(PR_SUCCESS == rval_status);
}

/*
 * cancel a job
 *
 *  XXXX: is this needed? likely to be removed
 */
PR_IMPLEMENT(PRStatus)
PR_CancelJob(PRJob *jobp) {

    PRStatus rval = PR_FAILURE;
    PRThreadPool *tp;

    if (jobp->on_timerq) {
        /*
         * now, check again while holding the timerq lock
         */
        tp = jobp->tpool;
        PR_Lock(tp->timerq.lock);
        if (jobp->on_timerq) {
            jobp->on_timerq = PR_FALSE;
            PR_REMOVE_AND_INIT_LINK(&jobp->links);
            tp->timerq.cnt--;
            PR_Unlock(tp->timerq.lock);
            if (!JOINABLE_JOB(jobp)) {
                delete_job(jobp);
            } else {
                JOIN_NOTIFY(jobp);
            }
            rval = PR_SUCCESS;
        } else {
            PR_Unlock(tp->timerq.lock);
        }
    } else if (jobp->on_ioq) {
        /*
         * now, check again while holding the ioq lock
         */
        tp = jobp->tpool;
        PR_Lock(tp->ioq.lock);
        if (jobp->on_ioq) {
            jobp->cancel_cv = PR_NewCondVar(tp->ioq.lock);
            if (NULL == jobp->cancel_cv) {
                PR_Unlock(tp->ioq.lock);
                PR_SetError(PR_INSUFFICIENT_RESOURCES_ERROR, 0);
                return PR_FAILURE;
            }
            /*
             * mark job 'cancelled' and notify io thread(s)
             * XXXX:
             *      this assumes there is only one io thread; when there
             *      are multiple threads, the io thread processing this job
             *      must be notified.
             */
            jobp->cancel_io = PR_TRUE;
            PR_Unlock(tp->ioq.lock);    /* release, reacquire ioq lock */
            notify_ioq(tp);
            PR_Lock(tp->ioq.lock);
            while (jobp->cancel_io) {
                PR_WaitCondVar(jobp->cancel_cv, PR_INTERVAL_NO_TIMEOUT);
            }
            PR_Unlock(tp->ioq.lock);
            PR_ASSERT(!jobp->on_ioq);
            if (!JOINABLE_JOB(jobp)) {
                delete_job(jobp);
            } else {
                JOIN_NOTIFY(jobp);
            }
            rval = PR_SUCCESS;
        } else {
            PR_Unlock(tp->ioq.lock);
        }
    }
    if (PR_FAILURE == rval) {
        PR_SetError(PR_INVALID_STATE_ERROR, 0);
    }
    return rval;
}

/* join a job, wait until completion */
PR_IMPLEMENT(PRStatus)
PR_JoinJob(PRJob *jobp)
{
    if (!JOINABLE_JOB(jobp)) {
        PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
        return PR_FAILURE;
    }
    PR_Lock(jobp->tpool->join_lock);
    while(jobp->join_wait) {
        PR_WaitCondVar(jobp->join_cv, PR_INTERVAL_NO_TIMEOUT);
    }
    PR_Unlock(jobp->tpool->join_lock);
    delete_job(jobp);
    return PR_SUCCESS;
}

/* shutdown threadpool */
PR_IMPLEMENT(PRStatus)
PR_ShutdownThreadPool(PRThreadPool *tpool)
{
    PRStatus rval = PR_SUCCESS;

    PR_Lock(tpool->jobq.lock);
    tpool->shutdown = PR_TRUE;
    PR_NotifyAllCondVar(tpool->shutdown_cv);
    PR_Unlock(tpool->jobq.lock);

    return rval;
}

/*
 * join thread pool
 *  wait for termination of worker threads
 *  reclaim threadpool resources
 */
PR_IMPLEMENT(PRStatus)
PR_JoinThreadPool(PRThreadPool *tpool)
{
    PRStatus rval = PR_SUCCESS;
    PRCList *head;
    PRStatus rval_status;

    PR_Lock(tpool->jobq.lock);
    while (!tpool->shutdown) {
        PR_WaitCondVar(tpool->shutdown_cv, PR_INTERVAL_NO_TIMEOUT);
    }

    /*
     * wakeup worker threads
     */
#ifdef OPT_WINNT
    /*
     * post shutdown notification for all threads
     */
    {
        int i;
        for(i=0; i < tpool->current_threads; i++) {
            PostQueuedCompletionStatus(tpool->jobq.nt_completion_port, 0,
                                       TRUE, NULL);
        }
    }
#else
    PR_NotifyAllCondVar(tpool->jobq.cv);
#endif

    /*
     * wakeup io thread(s)
     */
    notify_ioq(tpool);

    /*
     * wakeup timer thread(s)
     */
    PR_Lock(tpool->timerq.lock);
    notify_timerq(tpool);
    PR_Unlock(tpool->timerq.lock);

    while (!PR_CLIST_IS_EMPTY(&tpool->jobq.wthreads)) {
        wthread *wthrp;

        head = PR_LIST_HEAD(&tpool->jobq.wthreads);
        PR_REMOVE_AND_INIT_LINK(head);
        PR_Unlock(tpool->jobq.lock);
        wthrp = WTHREAD_LINKS_PTR(head);
        rval_status = PR_JoinThread(wthrp->thread);
        PR_ASSERT(PR_SUCCESS == rval_status);
        PR_DELETE(wthrp);
        PR_Lock(tpool->jobq.lock);
    }
    PR_Unlock(tpool->jobq.lock);
    while (!PR_CLIST_IS_EMPTY(&tpool->ioq.wthreads)) {
        wthread *wthrp;

        head = PR_LIST_HEAD(&tpool->ioq.wthreads);
        PR_REMOVE_AND_INIT_LINK(head);
        wthrp = WTHREAD_LINKS_PTR(head);
        rval_status = PR_JoinThread(wthrp->thread);
        PR_ASSERT(PR_SUCCESS == rval_status);
        PR_DELETE(wthrp);
    }

    while (!PR_CLIST_IS_EMPTY(&tpool->timerq.wthreads)) {
        wthread *wthrp;

        head = PR_LIST_HEAD(&tpool->timerq.wthreads);
        PR_REMOVE_AND_INIT_LINK(head);
        wthrp = WTHREAD_LINKS_PTR(head);
        rval_status = PR_JoinThread(wthrp->thread);
        PR_ASSERT(PR_SUCCESS == rval_status);
        PR_DELETE(wthrp);
    }

    /*
     * Delete queued jobs
     */
    while (!PR_CLIST_IS_EMPTY(&tpool->jobq.list)) {
        PRJob *jobp;

        head = PR_LIST_HEAD(&tpool->jobq.list);
        PR_REMOVE_AND_INIT_LINK(head);
        jobp = JOB_LINKS_PTR(head);
        tpool->jobq.cnt--;
        delete_job(jobp);
    }

    /* delete io jobs */
    while (!PR_CLIST_IS_EMPTY(&tpool->ioq.list)) {
        PRJob *jobp;

        head = PR_LIST_HEAD(&tpool->ioq.list);
        PR_REMOVE_AND_INIT_LINK(head);
        tpool->ioq.cnt--;
        jobp = JOB_LINKS_PTR(head);
        delete_job(jobp);
    }

    /* delete timer jobs */
    while (!PR_CLIST_IS_EMPTY(&tpool->timerq.list)) {
        PRJob *jobp;

        head = PR_LIST_HEAD(&tpool->timerq.list);
        PR_REMOVE_AND_INIT_LINK(head);
        tpool->timerq.cnt--;
        jobp = JOB_LINKS_PTR(head);
        delete_job(jobp);
    }

    PR_ASSERT(0 == tpool->jobq.cnt);
    PR_ASSERT(0 == tpool->ioq.cnt);
    PR_ASSERT(0 == tpool->timerq.cnt);

    delete_threadpool(tpool);
    return rval;
}