[ tor-browser ].git.dasho

prtpool.c (29322B)
      1 /* -*- Mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
      2 /* This Source Code Form is subject to the terms of the Mozilla Public
      3 * License, v. 2.0. If a copy of the MPL was not distributed with this
      4 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
      5 
      6 #include "nspr.h"
      7 
      8 /*
      9 * Thread pools
     10 *  Thread pools create and manage threads to provide support for
     11 *  scheduling jobs onto one or more threads.
     12 *
     13 */
     14 #ifdef OPT_WINNT
     15 #  include <windows.h>
     16 #endif
     17 
     18 /*
     19 * worker thread
     20 */
     21 typedef struct wthread {
     22  PRCList links;
     23  PRThread* thread;
     24 } wthread;
     25 
     26 /*
     27 * queue of timer jobs
     28 */
     29 typedef struct timer_jobq {
     30  PRCList list;
     31  PRLock* lock;
     32  PRCondVar* cv;
     33  PRInt32 cnt;
     34  PRCList wthreads;
     35 } timer_jobq;
     36 
     37 /*
     38 * queue of jobs
     39 */
     40 typedef struct tp_jobq {
     41  PRCList list;
     42  PRInt32 cnt;
     43  PRLock* lock;
     44  PRCondVar* cv;
     45  PRCList wthreads;
     46 #ifdef OPT_WINNT
     47  HANDLE nt_completion_port;
     48 #endif
     49 } tp_jobq;
     50 
     51 /*
     52 * queue of IO jobs
     53 */
     54 typedef struct io_jobq {
     55  PRCList list;
     56  PRPollDesc* pollfds;
     57  PRInt32 npollfds;
     58  PRJob** polljobs;
     59  PRLock* lock;
     60  PRInt32 cnt;
     61  PRFileDesc* notify_fd;
     62  PRCList wthreads;
     63 } io_jobq;
     64 
     65 /*
     66 * Threadpool
     67 */
     68 struct PRThreadPool {
     69  PRInt32 init_threads;
     70  PRInt32 max_threads;
     71  PRInt32 current_threads;
     72  PRInt32 idle_threads;
     73  PRUint32 stacksize;
     74  tp_jobq jobq;
     75  io_jobq ioq;
     76  timer_jobq timerq;
     77  PRLock* join_lock; /* used with jobp->join_cv */
     78  PRCondVar* shutdown_cv;
     79  PRBool shutdown;
     80 };
     81 
     82 typedef enum io_op_type {
     83  JOB_IO_READ,
     84  JOB_IO_WRITE,
     85  JOB_IO_CONNECT,
     86  JOB_IO_ACCEPT
     87 } io_op_type;
     88 
     89 #ifdef OPT_WINNT
     90 typedef struct NT_notifier {
     91  OVERLAPPED overlapped; /* must be first */
     92  PRJob* jobp;
     93 } NT_notifier;
     94 #endif
     95 
     96 struct PRJob {
     97  PRCList links;    /*  for linking jobs */
     98  PRBool on_ioq;    /* job on ioq */
     99  PRBool on_timerq; /* job on timerq */
    100  PRJobFn job_func;
    101  void* job_arg;
    102  PRCondVar* join_cv;
    103  PRBool join_wait;     /* == PR_TRUE, when waiting to join */
    104  PRCondVar* cancel_cv; /* for cancelling IO jobs */
    105  PRBool cancel_io;     /* for cancelling IO jobs */
    106  PRThreadPool* tpool;  /* back pointer to thread pool */
    107  PRJobIoDesc* iod;
    108  io_op_type io_op;
    109  PRInt16 io_poll_flags;
    110  PRNetAddr* netaddr;
    111  PRIntervalTime timeout; /* relative value */
    112  PRIntervalTime absolute;
    113 #ifdef OPT_WINNT
    114  NT_notifier nt_notifier;
    115 #endif
    116 };
    117 
    118 #define JOB_LINKS_PTR(_qp) ((PRJob*)((char*)(_qp) - offsetof(PRJob, links)))
    119 
    120 #define WTHREAD_LINKS_PTR(_qp) \
    121  ((wthread*)((char*)(_qp) - offsetof(wthread, links)))
    122 
    123 #define JOINABLE_JOB(_jobp) (NULL != (_jobp)->join_cv)
    124 
    125 #define JOIN_NOTIFY(_jobp)            \
    126  PR_BEGIN_MACRO                      \
    127  PR_Lock(_jobp->tpool->join_lock);   \
    128  _jobp->join_wait = PR_FALSE;        \
    129  PR_NotifyCondVar(_jobp->join_cv);   \
    130  PR_Unlock(_jobp->tpool->join_lock); \
    131  PR_END_MACRO
    132 
    133 #define CANCEL_IO_JOB(jobp)              \
    134  PR_BEGIN_MACRO                         \
    135  jobp->cancel_io = PR_FALSE;            \
    136  jobp->on_ioq = PR_FALSE;               \
    137  PR_REMOVE_AND_INIT_LINK(&jobp->links); \
    138  tp->ioq.cnt--;                         \
    139  PR_NotifyCondVar(jobp->cancel_cv);     \
    140  PR_END_MACRO
    141 
    142 static void delete_job(PRJob* jobp);
    143 static PRThreadPool* alloc_threadpool(void);
    144 static PRJob* alloc_job(PRBool joinable, PRThreadPool* tp);
    145 static void notify_ioq(PRThreadPool* tp);
    146 static void notify_timerq(PRThreadPool* tp);
    147 
    148 /*
    149 * locks are acquired in the following order
    150 *
    151 *  tp->ioq.lock,tp->timerq.lock
    152 *          |
    153 *          V
    154 *      tp->jobq->lock
    155 */
    156 
    157 /*
    158 * worker thread function
    159 */
    160 static void wstart(void* arg) {
    161  PRThreadPool* tp = (PRThreadPool*)arg;
    162  PRCList* head;
    163 
    164  /*
    165   * execute jobs until shutdown
    166   */
    167  while (!tp->shutdown) {
    168    PRJob* jobp;
    169 #ifdef OPT_WINNT
    170    BOOL rv;
    171    DWORD unused, shutdown;
    172    LPOVERLAPPED olp;
    173 
    174    PR_Lock(tp->jobq.lock);
    175    tp->idle_threads++;
    176    PR_Unlock(tp->jobq.lock);
    177    rv = GetQueuedCompletionStatus(tp->jobq.nt_completion_port, &unused,
    178                                   &shutdown, &olp, INFINITE);
    179 
    180    PR_ASSERT(rv);
    181    if (shutdown) {
    182      break;
    183    }
    184    jobp = ((NT_notifier*)olp)->jobp;
    185    PR_Lock(tp->jobq.lock);
    186    tp->idle_threads--;
    187    tp->jobq.cnt--;
    188    PR_Unlock(tp->jobq.lock);
    189 #else
    190 
    191    PR_Lock(tp->jobq.lock);
    192    while (PR_CLIST_IS_EMPTY(&tp->jobq.list) && (!tp->shutdown)) {
    193      tp->idle_threads++;
    194      PR_WaitCondVar(tp->jobq.cv, PR_INTERVAL_NO_TIMEOUT);
    195      tp->idle_threads--;
    196    }
    197    if (tp->shutdown) {
    198      PR_Unlock(tp->jobq.lock);
    199      break;
    200    }
    201    head = PR_LIST_HEAD(&tp->jobq.list);
    202    /*
    203     * remove job from queue
    204     */
    205    PR_REMOVE_AND_INIT_LINK(head);
    206    tp->jobq.cnt--;
    207    jobp = JOB_LINKS_PTR(head);
    208    PR_Unlock(tp->jobq.lock);
    209 #endif
    210 
    211    jobp->job_func(jobp->job_arg);
    212    if (!JOINABLE_JOB(jobp)) {
    213      delete_job(jobp);
    214    } else {
    215      JOIN_NOTIFY(jobp);
    216    }
    217  }
    218  PR_Lock(tp->jobq.lock);
    219  tp->current_threads--;
    220  PR_Unlock(tp->jobq.lock);
    221 }
    222 
    223 /*
    224 * add a job to the work queue
    225 */
    226 static void add_to_jobq(PRThreadPool* tp, PRJob* jobp) {
    227  /*
    228   * add to jobq
    229   */
    230 #ifdef OPT_WINNT
    231  PR_Lock(tp->jobq.lock);
    232  tp->jobq.cnt++;
    233  PR_Unlock(tp->jobq.lock);
    234  /*
    235   * notify worker thread(s)
    236   */
    237  PostQueuedCompletionStatus(tp->jobq.nt_completion_port, 0, FALSE,
    238                             &jobp->nt_notifier.overlapped);
    239 #else
    240  PR_Lock(tp->jobq.lock);
    241  PR_APPEND_LINK(&jobp->links, &tp->jobq.list);
    242  tp->jobq.cnt++;
    243  if ((tp->idle_threads < tp->jobq.cnt) &&
    244      (tp->current_threads < tp->max_threads)) {
    245    wthread* wthrp;
    246    /*
    247     * increment thread count and unlock the jobq lock
    248     */
    249    tp->current_threads++;
    250    PR_Unlock(tp->jobq.lock);
    251    /* create new worker thread */
    252    wthrp = PR_NEWZAP(wthread);
    253    if (wthrp) {
    254      wthrp->thread =
    255          PR_CreateThread(PR_USER_THREAD, wstart, tp, PR_PRIORITY_NORMAL,
    256                          PR_GLOBAL_THREAD, PR_JOINABLE_THREAD, tp->stacksize);
    257      if (NULL == wthrp->thread) {
    258        PR_DELETE(wthrp); /* this sets wthrp to NULL */
    259      }
    260    }
    261    PR_Lock(tp->jobq.lock);
    262    if (NULL == wthrp) {
    263      tp->current_threads--;
    264    } else {
    265      PR_APPEND_LINK(&wthrp->links, &tp->jobq.wthreads);
    266    }
    267  }
    268  /*
    269   * wakeup a worker thread
    270   */
    271  PR_NotifyCondVar(tp->jobq.cv);
    272  PR_Unlock(tp->jobq.lock);
    273 #endif
    274 }
    275 
    276 /*
    277 * io worker thread function
    278 */
    279 static void io_wstart(void* arg) {
    280  PRThreadPool* tp = (PRThreadPool*)arg;
    281  int pollfd_cnt, pollfds_used;
    282  int rv;
    283  PRCList *qp, *nextqp;
    284  PRPollDesc* pollfds = NULL;
    285  PRJob** polljobs = NULL;
    286  int poll_timeout;
    287  PRIntervalTime now;
    288 
    289  /*
    290   * scan io_jobq
    291   * construct poll list
    292   * call PR_Poll
    293   * for all fds, for which poll returns true, move the job to
    294   * jobq and wakeup worker thread.
    295   */
    296  while (!tp->shutdown) {
    297    PRJob* jobp;
    298 
    299    pollfd_cnt = tp->ioq.cnt + 10;
    300    if (pollfd_cnt > tp->ioq.npollfds) {
    301      /*
    302       * re-allocate pollfd array if the current one is not large
    303       * enough
    304       */
    305      if (NULL != tp->ioq.pollfds) {
    306        PR_Free(tp->ioq.pollfds);
    307      }
    308      tp->ioq.pollfds = (PRPollDesc*)PR_Malloc(
    309          pollfd_cnt * (sizeof(PRPollDesc) + sizeof(PRJob*)));
    310      PR_ASSERT(NULL != tp->ioq.pollfds);
    311      /*
    312       * array of pollfds
    313       */
    314      pollfds = tp->ioq.pollfds;
    315      tp->ioq.polljobs = (PRJob**)(&tp->ioq.pollfds[pollfd_cnt]);
    316      /*
    317       * parallel array of jobs
    318       */
    319      polljobs = tp->ioq.polljobs;
    320      tp->ioq.npollfds = pollfd_cnt;
    321    }
    322 
    323    pollfds_used = 0;
    324    /*
    325     * add the notify fd; used for unblocking io thread(s)
    326     */
    327    pollfds[pollfds_used].fd = tp->ioq.notify_fd;
    328    pollfds[pollfds_used].in_flags = PR_POLL_READ;
    329    pollfds[pollfds_used].out_flags = 0;
    330    polljobs[pollfds_used] = NULL;
    331    pollfds_used++;
    332    /*
    333     * fill in the pollfd array
    334     */
    335    PR_Lock(tp->ioq.lock);
    336    for (qp = tp->ioq.list.next; qp != &tp->ioq.list; qp = nextqp) {
    337      nextqp = qp->next;
    338      jobp = JOB_LINKS_PTR(qp);
    339      if (jobp->cancel_io) {
    340        CANCEL_IO_JOB(jobp);
    341        continue;
    342      }
    343      if (pollfds_used == (pollfd_cnt)) {
    344        break;
    345      }
    346      pollfds[pollfds_used].fd = jobp->iod->socket;
    347      pollfds[pollfds_used].in_flags = jobp->io_poll_flags;
    348      pollfds[pollfds_used].out_flags = 0;
    349      polljobs[pollfds_used] = jobp;
    350 
    351      pollfds_used++;
    352    }
    353    if (!PR_CLIST_IS_EMPTY(&tp->ioq.list)) {
    354      qp = tp->ioq.list.next;
    355      jobp = JOB_LINKS_PTR(qp);
    356      if (PR_INTERVAL_NO_TIMEOUT == jobp->timeout) {
    357        poll_timeout = PR_INTERVAL_NO_TIMEOUT;
    358      } else if (PR_INTERVAL_NO_WAIT == jobp->timeout) {
    359        poll_timeout = PR_INTERVAL_NO_WAIT;
    360      } else {
    361        poll_timeout = jobp->absolute - PR_IntervalNow();
    362        if (poll_timeout <= 0) { /* already timed out */
    363          poll_timeout = PR_INTERVAL_NO_WAIT;
    364        }
    365      }
    366    } else {
    367      poll_timeout = PR_INTERVAL_NO_TIMEOUT;
    368    }
    369    PR_Unlock(tp->ioq.lock);
    370 
    371    /*
    372     * XXXX
    373     * should retry if more jobs have been added to the queue?
    374     *
    375     */
    376    PR_ASSERT(pollfds_used <= pollfd_cnt);
    377    rv = PR_Poll(tp->ioq.pollfds, pollfds_used, poll_timeout);
    378 
    379    if (tp->shutdown) {
    380      break;
    381    }
    382 
    383    if (rv > 0) {
    384      /*
    385       * at least one io event is set
    386       */
    387      PRStatus rval_status;
    388      PRInt32 index;
    389 
    390      PR_ASSERT(pollfds[0].fd == tp->ioq.notify_fd);
    391      /*
    392       * reset the pollable event, if notified
    393       */
    394      if (pollfds[0].out_flags & PR_POLL_READ) {
    395        rval_status = PR_WaitForPollableEvent(tp->ioq.notify_fd);
    396        PR_ASSERT(PR_SUCCESS == rval_status);
    397      }
    398 
    399      for (index = 1; index < (pollfds_used); index++) {
    400        PRInt16 events = pollfds[index].in_flags;
    401        PRInt16 revents = pollfds[index].out_flags;
    402        jobp = polljobs[index];
    403 
    404        if ((revents & PR_POLL_NVAL) || /* busted in all cases */
    405            (revents & PR_POLL_ERR) ||
    406            ((events & PR_POLL_WRITE) &&
    407             (revents & PR_POLL_HUP))) { /* write op & hup */
    408          PR_Lock(tp->ioq.lock);
    409          if (jobp->cancel_io) {
    410            CANCEL_IO_JOB(jobp);
    411            PR_Unlock(tp->ioq.lock);
    412            continue;
    413          }
    414          PR_REMOVE_AND_INIT_LINK(&jobp->links);
    415          tp->ioq.cnt--;
    416          jobp->on_ioq = PR_FALSE;
    417          PR_Unlock(tp->ioq.lock);
    418 
    419          /* set error */
    420          if (PR_POLL_NVAL & revents) {
    421            jobp->iod->error = PR_BAD_DESCRIPTOR_ERROR;
    422          } else if (PR_POLL_HUP & revents) {
    423            jobp->iod->error = PR_CONNECT_RESET_ERROR;
    424          } else {
    425            jobp->iod->error = PR_IO_ERROR;
    426          }
    427 
    428          /*
    429           * add to jobq
    430           */
    431          add_to_jobq(tp, jobp);
    432        } else if (revents) {
    433          /*
    434           * add to jobq
    435           */
    436          PR_Lock(tp->ioq.lock);
    437          if (jobp->cancel_io) {
    438            CANCEL_IO_JOB(jobp);
    439            PR_Unlock(tp->ioq.lock);
    440            continue;
    441          }
    442          PR_REMOVE_AND_INIT_LINK(&jobp->links);
    443          tp->ioq.cnt--;
    444          jobp->on_ioq = PR_FALSE;
    445          PR_Unlock(tp->ioq.lock);
    446 
    447          if (jobp->io_op == JOB_IO_CONNECT) {
    448            if (PR_GetConnectStatus(&pollfds[index]) == PR_SUCCESS) {
    449              jobp->iod->error = 0;
    450            } else {
    451              jobp->iod->error = PR_GetError();
    452            }
    453          } else {
    454            jobp->iod->error = 0;
    455          }
    456 
    457          add_to_jobq(tp, jobp);
    458        }
    459      }
    460    }
    461    /*
    462     * timeout processing
    463     */
    464    now = PR_IntervalNow();
    465    PR_Lock(tp->ioq.lock);
    466    for (qp = tp->ioq.list.next; qp != &tp->ioq.list; qp = nextqp) {
    467      nextqp = qp->next;
    468      jobp = JOB_LINKS_PTR(qp);
    469      if (jobp->cancel_io) {
    470        CANCEL_IO_JOB(jobp);
    471        continue;
    472      }
    473      if (PR_INTERVAL_NO_TIMEOUT == jobp->timeout) {
    474        break;
    475      }
    476      if ((PR_INTERVAL_NO_WAIT != jobp->timeout) &&
    477          ((PRInt32)(jobp->absolute - now) > 0)) {
    478        break;
    479      }
    480      PR_REMOVE_AND_INIT_LINK(&jobp->links);
    481      tp->ioq.cnt--;
    482      jobp->on_ioq = PR_FALSE;
    483      jobp->iod->error = PR_IO_TIMEOUT_ERROR;
    484      add_to_jobq(tp, jobp);
    485    }
    486    PR_Unlock(tp->ioq.lock);
    487  }
    488 }
    489 
    490 /*
    491 * timer worker thread function
    492 */
    493 static void timer_wstart(void* arg) {
    494  PRThreadPool* tp = (PRThreadPool*)arg;
    495  PRCList* qp;
    496  PRIntervalTime timeout;
    497  PRIntervalTime now;
    498 
    499  /*
    500   * call PR_WaitCondVar with minimum value of all timeouts
    501   */
    502  while (!tp->shutdown) {
    503    PRJob* jobp;
    504 
    505    PR_Lock(tp->timerq.lock);
    506    if (PR_CLIST_IS_EMPTY(&tp->timerq.list)) {
    507      timeout = PR_INTERVAL_NO_TIMEOUT;
    508    } else {
    509      PRCList* qp;
    510 
    511      qp = tp->timerq.list.next;
    512      jobp = JOB_LINKS_PTR(qp);
    513 
    514      timeout = jobp->absolute - PR_IntervalNow();
    515      if (timeout <= 0) {
    516        timeout = PR_INTERVAL_NO_WAIT; /* already timed out */
    517      }
    518    }
    519    if (PR_INTERVAL_NO_WAIT != timeout) {
    520      PR_WaitCondVar(tp->timerq.cv, timeout);
    521    }
    522    if (tp->shutdown) {
    523      PR_Unlock(tp->timerq.lock);
    524      break;
    525    }
    526    /*
    527     * move expired-timer jobs to jobq
    528     */
    529    now = PR_IntervalNow();
    530    while (!PR_CLIST_IS_EMPTY(&tp->timerq.list)) {
    531      qp = tp->timerq.list.next;
    532      jobp = JOB_LINKS_PTR(qp);
    533 
    534      if ((PRInt32)(jobp->absolute - now) > 0) {
    535        break;
    536      }
    537      /*
    538       * job timed out
    539       */
    540      PR_REMOVE_AND_INIT_LINK(&jobp->links);
    541      tp->timerq.cnt--;
    542      jobp->on_timerq = PR_FALSE;
    543      add_to_jobq(tp, jobp);
    544    }
    545    PR_Unlock(tp->timerq.lock);
    546  }
    547 }
    548 
    549 static void delete_threadpool(PRThreadPool* tp) {
    550  if (NULL != tp) {
    551    if (NULL != tp->shutdown_cv) {
    552      PR_DestroyCondVar(tp->shutdown_cv);
    553    }
    554    if (NULL != tp->jobq.cv) {
    555      PR_DestroyCondVar(tp->jobq.cv);
    556    }
    557    if (NULL != tp->jobq.lock) {
    558      PR_DestroyLock(tp->jobq.lock);
    559    }
    560    if (NULL != tp->join_lock) {
    561      PR_DestroyLock(tp->join_lock);
    562    }
    563 #ifdef OPT_WINNT
    564    if (NULL != tp->jobq.nt_completion_port) {
    565      CloseHandle(tp->jobq.nt_completion_port);
    566    }
    567 #endif
    568    /* Timer queue */
    569    if (NULL != tp->timerq.cv) {
    570      PR_DestroyCondVar(tp->timerq.cv);
    571    }
    572    if (NULL != tp->timerq.lock) {
    573      PR_DestroyLock(tp->timerq.lock);
    574    }
    575 
    576    if (NULL != tp->ioq.lock) {
    577      PR_DestroyLock(tp->ioq.lock);
    578    }
    579    if (NULL != tp->ioq.pollfds) {
    580      PR_Free(tp->ioq.pollfds);
    581    }
    582    if (NULL != tp->ioq.notify_fd) {
    583      PR_DestroyPollableEvent(tp->ioq.notify_fd);
    584    }
    585    PR_Free(tp);
    586  }
    587  return;
    588 }
    589 
    590 static PRThreadPool* alloc_threadpool(void) {
    591  PRThreadPool* tp;
    592 
    593  tp = (PRThreadPool*)PR_CALLOC(sizeof(*tp));
    594  if (NULL == tp) {
    595    goto failed;
    596  }
    597  tp->jobq.lock = PR_NewLock();
    598  if (NULL == tp->jobq.lock) {
    599    goto failed;
    600  }
    601  tp->jobq.cv = PR_NewCondVar(tp->jobq.lock);
    602  if (NULL == tp->jobq.cv) {
    603    goto failed;
    604  }
    605  tp->join_lock = PR_NewLock();
    606  if (NULL == tp->join_lock) {
    607    goto failed;
    608  }
    609 #ifdef OPT_WINNT
    610  tp->jobq.nt_completion_port =
    611      CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    612  if (NULL == tp->jobq.nt_completion_port) {
    613    goto failed;
    614  }
    615 #endif
    616 
    617  tp->ioq.lock = PR_NewLock();
    618  if (NULL == tp->ioq.lock) {
    619    goto failed;
    620  }
    621 
    622  /* Timer queue */
    623 
    624  tp->timerq.lock = PR_NewLock();
    625  if (NULL == tp->timerq.lock) {
    626    goto failed;
    627  }
    628  tp->timerq.cv = PR_NewCondVar(tp->timerq.lock);
    629  if (NULL == tp->timerq.cv) {
    630    goto failed;
    631  }
    632 
    633  tp->shutdown_cv = PR_NewCondVar(tp->jobq.lock);
    634  if (NULL == tp->shutdown_cv) {
    635    goto failed;
    636  }
    637  tp->ioq.notify_fd = PR_NewPollableEvent();
    638  if (NULL == tp->ioq.notify_fd) {
    639    goto failed;
    640  }
    641  return tp;
    642 failed:
    643  delete_threadpool(tp);
    644  PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
    645  return NULL;
    646 }
    647 
    648 /* Create thread pool */
    649 PR_IMPLEMENT(PRThreadPool*)
    650 PR_CreateThreadPool(PRInt32 initial_threads, PRInt32 max_threads,
    651                    PRUint32 stacksize) {
    652  PRThreadPool* tp;
    653  PRThread* thr;
    654  int i;
    655  wthread* wthrp;
    656 
    657  tp = alloc_threadpool();
    658  if (NULL == tp) {
    659    return NULL;
    660  }
    661 
    662  tp->init_threads = initial_threads;
    663  tp->max_threads = max_threads;
    664  tp->stacksize = stacksize;
    665  PR_INIT_CLIST(&tp->jobq.list);
    666  PR_INIT_CLIST(&tp->ioq.list);
    667  PR_INIT_CLIST(&tp->timerq.list);
    668  PR_INIT_CLIST(&tp->jobq.wthreads);
    669  PR_INIT_CLIST(&tp->ioq.wthreads);
    670  PR_INIT_CLIST(&tp->timerq.wthreads);
    671  tp->shutdown = PR_FALSE;
    672 
    673  PR_Lock(tp->jobq.lock);
    674  for (i = 0; i < initial_threads; ++i) {
    675    thr = PR_CreateThread(PR_USER_THREAD, wstart, tp, PR_PRIORITY_NORMAL,
    676                          PR_GLOBAL_THREAD, PR_JOINABLE_THREAD, stacksize);
    677    PR_ASSERT(thr);
    678    wthrp = PR_NEWZAP(wthread);
    679    PR_ASSERT(wthrp);
    680    wthrp->thread = thr;
    681    PR_APPEND_LINK(&wthrp->links, &tp->jobq.wthreads);
    682  }
    683  tp->current_threads = initial_threads;
    684 
    685  thr = PR_CreateThread(PR_USER_THREAD, io_wstart, tp, PR_PRIORITY_NORMAL,
    686                        PR_GLOBAL_THREAD, PR_JOINABLE_THREAD, stacksize);
    687  PR_ASSERT(thr);
    688  wthrp = PR_NEWZAP(wthread);
    689  PR_ASSERT(wthrp);
    690  wthrp->thread = thr;
    691  PR_APPEND_LINK(&wthrp->links, &tp->ioq.wthreads);
    692 
    693  thr = PR_CreateThread(PR_USER_THREAD, timer_wstart, tp, PR_PRIORITY_NORMAL,
    694                        PR_GLOBAL_THREAD, PR_JOINABLE_THREAD, stacksize);
    695  PR_ASSERT(thr);
    696  wthrp = PR_NEWZAP(wthread);
    697  PR_ASSERT(wthrp);
    698  wthrp->thread = thr;
    699  PR_APPEND_LINK(&wthrp->links, &tp->timerq.wthreads);
    700 
    701  PR_Unlock(tp->jobq.lock);
    702  return tp;
    703 }
    704 
    705 static void delete_job(PRJob* jobp) {
    706  if (NULL != jobp) {
    707    if (NULL != jobp->join_cv) {
    708      PR_DestroyCondVar(jobp->join_cv);
    709      jobp->join_cv = NULL;
    710    }
    711    if (NULL != jobp->cancel_cv) {
    712      PR_DestroyCondVar(jobp->cancel_cv);
    713      jobp->cancel_cv = NULL;
    714    }
    715    PR_DELETE(jobp);
    716  }
    717 }
    718 
    719 static PRJob* alloc_job(PRBool joinable, PRThreadPool* tp) {
    720  PRJob* jobp;
    721 
    722  jobp = PR_NEWZAP(PRJob);
    723  if (NULL == jobp) {
    724    goto failed;
    725  }
    726  if (joinable) {
    727    jobp->join_cv = PR_NewCondVar(tp->join_lock);
    728    jobp->join_wait = PR_TRUE;
    729    if (NULL == jobp->join_cv) {
    730      goto failed;
    731    }
    732  } else {
    733    jobp->join_cv = NULL;
    734  }
    735 #ifdef OPT_WINNT
    736  jobp->nt_notifier.jobp = jobp;
    737 #endif
    738  return jobp;
    739 failed:
    740  delete_job(jobp);
    741  PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
    742  return NULL;
    743 }
    744 
    745 /* queue a job */
    746 PR_IMPLEMENT(PRJob*)
    747 PR_QueueJob(PRThreadPool* tpool, PRJobFn fn, void* arg, PRBool joinable) {
    748  PRJob* jobp;
    749 
    750  jobp = alloc_job(joinable, tpool);
    751  if (NULL == jobp) {
    752    return NULL;
    753  }
    754 
    755  jobp->job_func = fn;
    756  jobp->job_arg = arg;
    757  jobp->tpool = tpool;
    758 
    759  add_to_jobq(tpool, jobp);
    760  return jobp;
    761 }
    762 
    763 /* queue a job, when a socket is readable or writeable */
    764 static PRJob* queue_io_job(PRThreadPool* tpool, PRJobIoDesc* iod, PRJobFn fn,
    765                           void* arg, PRBool joinable, io_op_type op) {
    766  PRJob* jobp;
    767  PRIntervalTime now;
    768 
    769  jobp = alloc_job(joinable, tpool);
    770  if (NULL == jobp) {
    771    return NULL;
    772  }
    773 
    774  /*
    775   * Add a new job to io_jobq
    776   * wakeup io worker thread
    777   */
    778 
    779  jobp->job_func = fn;
    780  jobp->job_arg = arg;
    781  jobp->tpool = tpool;
    782  jobp->iod = iod;
    783  if (JOB_IO_READ == op) {
    784    jobp->io_op = JOB_IO_READ;
    785    jobp->io_poll_flags = PR_POLL_READ;
    786  } else if (JOB_IO_WRITE == op) {
    787    jobp->io_op = JOB_IO_WRITE;
    788    jobp->io_poll_flags = PR_POLL_WRITE;
    789  } else if (JOB_IO_ACCEPT == op) {
    790    jobp->io_op = JOB_IO_ACCEPT;
    791    jobp->io_poll_flags = PR_POLL_READ;
    792  } else if (JOB_IO_CONNECT == op) {
    793    jobp->io_op = JOB_IO_CONNECT;
    794    jobp->io_poll_flags = PR_POLL_WRITE | PR_POLL_EXCEPT;
    795  } else {
    796    delete_job(jobp);
    797    PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
    798    return NULL;
    799  }
    800 
    801  jobp->timeout = iod->timeout;
    802  if ((PR_INTERVAL_NO_TIMEOUT == iod->timeout) ||
    803      (PR_INTERVAL_NO_WAIT == iod->timeout)) {
    804    jobp->absolute = iod->timeout;
    805  } else {
    806    now = PR_IntervalNow();
    807    jobp->absolute = now + iod->timeout;
    808  }
    809 
    810  PR_Lock(tpool->ioq.lock);
    811 
    812  if (PR_CLIST_IS_EMPTY(&tpool->ioq.list) ||
    813      (PR_INTERVAL_NO_TIMEOUT == iod->timeout)) {
    814    PR_APPEND_LINK(&jobp->links, &tpool->ioq.list);
    815  } else if (PR_INTERVAL_NO_WAIT == iod->timeout) {
    816    PR_INSERT_LINK(&jobp->links, &tpool->ioq.list);
    817  } else {
    818    PRCList* qp;
    819    PRJob* tmp_jobp;
    820    /*
    821     * insert into the timeout-sorted ioq
    822     */
    823    for (qp = tpool->ioq.list.prev; qp != &tpool->ioq.list; qp = qp->prev) {
    824      tmp_jobp = JOB_LINKS_PTR(qp);
    825      if ((PRInt32)(jobp->absolute - tmp_jobp->absolute) >= 0) {
    826        break;
    827      }
    828    }
    829    PR_INSERT_AFTER(&jobp->links, qp);
    830  }
    831 
    832  jobp->on_ioq = PR_TRUE;
    833  tpool->ioq.cnt++;
    834  /*
    835   * notify io worker thread(s)
    836   */
    837  PR_Unlock(tpool->ioq.lock);
    838  notify_ioq(tpool);
    839  return jobp;
    840 }
    841 
    842 /* queue a job, when a socket is readable */
    843 PR_IMPLEMENT(PRJob*)
    844 PR_QueueJob_Read(PRThreadPool* tpool, PRJobIoDesc* iod, PRJobFn fn, void* arg,
    845                 PRBool joinable) {
    846  return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_READ));
    847 }
    848 
    849 /* queue a job, when a socket is writeable */
    850 PR_IMPLEMENT(PRJob*)
    851 PR_QueueJob_Write(PRThreadPool* tpool, PRJobIoDesc* iod, PRJobFn fn, void* arg,
    852                  PRBool joinable) {
    853  return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_WRITE));
    854 }
    855 
    856 /* queue a job, when a socket has a pending connection */
    857 PR_IMPLEMENT(PRJob*)
    858 PR_QueueJob_Accept(PRThreadPool* tpool, PRJobIoDesc* iod, PRJobFn fn, void* arg,
    859                   PRBool joinable) {
    860  return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_ACCEPT));
    861 }
    862 
    863 /* queue a job, when a socket can be connected */
    864 PR_IMPLEMENT(PRJob*)
    865 PR_QueueJob_Connect(PRThreadPool* tpool, PRJobIoDesc* iod,
    866                    const PRNetAddr* addr, PRJobFn fn, void* arg,
    867                    PRBool joinable) {
    868  PRStatus rv;
    869  PRErrorCode err;
    870 
    871  rv = PR_Connect(iod->socket, addr, PR_INTERVAL_NO_WAIT);
    872  if ((rv == PR_FAILURE) && ((err = PR_GetError()) == PR_IN_PROGRESS_ERROR)) {
    873    /* connection pending */
    874    return (queue_io_job(tpool, iod, fn, arg, joinable, JOB_IO_CONNECT));
    875  }
    876  /*
    877   * connection succeeded or failed; add to jobq right away
    878   */
    879  if (rv == PR_FAILURE) {
    880    iod->error = err;
    881  } else {
    882    iod->error = 0;
    883  }
    884  return (PR_QueueJob(tpool, fn, arg, joinable));
    885 }
    886 
    887 /* queue a job, when a timer expires */
    888 PR_IMPLEMENT(PRJob*)
    889 PR_QueueJob_Timer(PRThreadPool* tpool, PRIntervalTime timeout, PRJobFn fn,
    890                  void* arg, PRBool joinable) {
    891  PRIntervalTime now;
    892  PRJob* jobp;
    893 
    894  if (PR_INTERVAL_NO_TIMEOUT == timeout) {
    895    PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
    896    return NULL;
    897  }
    898  if (PR_INTERVAL_NO_WAIT == timeout) {
    899    /*
    900     * no waiting; add to jobq right away
    901     */
    902    return (PR_QueueJob(tpool, fn, arg, joinable));
    903  }
    904  jobp = alloc_job(joinable, tpool);
    905  if (NULL == jobp) {
    906    return NULL;
    907  }
    908 
    909  /*
    910   * Add a new job to timer_jobq
    911   * wakeup timer worker thread
    912   */
    913 
    914  jobp->job_func = fn;
    915  jobp->job_arg = arg;
    916  jobp->tpool = tpool;
    917  jobp->timeout = timeout;
    918 
    919  now = PR_IntervalNow();
    920  jobp->absolute = now + timeout;
    921 
    922  PR_Lock(tpool->timerq.lock);
    923  jobp->on_timerq = PR_TRUE;
    924  if (PR_CLIST_IS_EMPTY(&tpool->timerq.list)) {
    925    PR_APPEND_LINK(&jobp->links, &tpool->timerq.list);
    926  } else {
    927    PRCList* qp;
    928    PRJob* tmp_jobp;
    929    /*
    930     * insert into the sorted timer jobq
    931     */
    932    for (qp = tpool->timerq.list.prev; qp != &tpool->timerq.list;
    933         qp = qp->prev) {
    934      tmp_jobp = JOB_LINKS_PTR(qp);
    935      if ((PRInt32)(jobp->absolute - tmp_jobp->absolute) >= 0) {
    936        break;
    937      }
    938    }
    939    PR_INSERT_AFTER(&jobp->links, qp);
    940  }
    941  tpool->timerq.cnt++;
    942  /*
    943   * notify timer worker thread(s)
    944   */
    945  notify_timerq(tpool);
    946  PR_Unlock(tpool->timerq.lock);
    947  return jobp;
    948 }
    949 
    950 static void notify_timerq(PRThreadPool* tp) {
    951  /*
    952   * wakeup the timer thread(s)
    953   */
    954  PR_NotifyCondVar(tp->timerq.cv);
    955 }
    956 
    957 static void notify_ioq(PRThreadPool* tp) {
    958  PRStatus rval_status;
    959 
    960  /*
    961   * wakeup the io thread(s)
    962   */
    963  rval_status = PR_SetPollableEvent(tp->ioq.notify_fd);
    964  PR_ASSERT(PR_SUCCESS == rval_status);
    965 }
    966 
    967 /*
    968 * cancel a job
    969 *
    970 *  XXXX: is this needed? likely to be removed
    971 */
    972 PR_IMPLEMENT(PRStatus)
    973 PR_CancelJob(PRJob* jobp) {
    974  PRStatus rval = PR_FAILURE;
    975  PRThreadPool* tp;
    976 
    977  if (jobp->on_timerq) {
    978    /*
    979     * now, check again while holding the timerq lock
    980     */
    981    tp = jobp->tpool;
    982    PR_Lock(tp->timerq.lock);
    983    if (jobp->on_timerq) {
    984      jobp->on_timerq = PR_FALSE;
    985      PR_REMOVE_AND_INIT_LINK(&jobp->links);
    986      tp->timerq.cnt--;
    987      PR_Unlock(tp->timerq.lock);
    988      if (!JOINABLE_JOB(jobp)) {
    989        delete_job(jobp);
    990      } else {
    991        JOIN_NOTIFY(jobp);
    992      }
    993      rval = PR_SUCCESS;
    994    } else {
    995      PR_Unlock(tp->timerq.lock);
    996    }
    997  } else if (jobp->on_ioq) {
    998    /*
    999     * now, check again while holding the ioq lock
   1000     */
   1001    tp = jobp->tpool;
   1002    PR_Lock(tp->ioq.lock);
   1003    if (jobp->on_ioq) {
   1004      jobp->cancel_cv = PR_NewCondVar(tp->ioq.lock);
   1005      if (NULL == jobp->cancel_cv) {
   1006        PR_Unlock(tp->ioq.lock);
   1007        PR_SetError(PR_INSUFFICIENT_RESOURCES_ERROR, 0);
   1008        return PR_FAILURE;
   1009      }
   1010      /*
   1011       * mark job 'cancelled' and notify io thread(s)
   1012       * XXXX:
   1013       *      this assumes there is only one io thread; when there
   1014       *      are multiple threads, the io thread processing this job
   1015       *      must be notified.
   1016       */
   1017      jobp->cancel_io = PR_TRUE;
   1018      PR_Unlock(tp->ioq.lock); /* release, reacquire ioq lock */
   1019      notify_ioq(tp);
   1020      PR_Lock(tp->ioq.lock);
   1021      while (jobp->cancel_io) {
   1022        PR_WaitCondVar(jobp->cancel_cv, PR_INTERVAL_NO_TIMEOUT);
   1023      }
   1024      PR_Unlock(tp->ioq.lock);
   1025      PR_ASSERT(!jobp->on_ioq);
   1026      if (!JOINABLE_JOB(jobp)) {
   1027        delete_job(jobp);
   1028      } else {
   1029        JOIN_NOTIFY(jobp);
   1030      }
   1031      rval = PR_SUCCESS;
   1032    } else {
   1033      PR_Unlock(tp->ioq.lock);
   1034    }
   1035  }
   1036  if (PR_FAILURE == rval) {
   1037    PR_SetError(PR_INVALID_STATE_ERROR, 0);
   1038  }
   1039  return rval;
   1040 }
   1041 
   1042 /* join a job, wait until completion */
   1043 PR_IMPLEMENT(PRStatus)
   1044 PR_JoinJob(PRJob* jobp) {
   1045  if (!JOINABLE_JOB(jobp)) {
   1046    PR_SetError(PR_INVALID_ARGUMENT_ERROR, 0);
   1047    return PR_FAILURE;
   1048  }
   1049  PR_Lock(jobp->tpool->join_lock);
   1050  while (jobp->join_wait) {
   1051    PR_WaitCondVar(jobp->join_cv, PR_INTERVAL_NO_TIMEOUT);
   1052  }
   1053  PR_Unlock(jobp->tpool->join_lock);
   1054  delete_job(jobp);
   1055  return PR_SUCCESS;
   1056 }
   1057 
   1058 /* shutdown threadpool */
   1059 PR_IMPLEMENT(PRStatus)
   1060 PR_ShutdownThreadPool(PRThreadPool* tpool) {
   1061  PRStatus rval = PR_SUCCESS;
   1062 
   1063  PR_Lock(tpool->jobq.lock);
   1064  tpool->shutdown = PR_TRUE;
   1065  PR_NotifyAllCondVar(tpool->shutdown_cv);
   1066  PR_Unlock(tpool->jobq.lock);
   1067 
   1068  return rval;
   1069 }
   1070 
   1071 /*
   1072 * join thread pool
   1073 *  wait for termination of worker threads
   1074 *  reclaim threadpool resources
   1075 */
   1076 PR_IMPLEMENT(PRStatus)
   1077 PR_JoinThreadPool(PRThreadPool* tpool) {
   1078  PRStatus rval = PR_SUCCESS;
   1079  PRCList* head;
   1080  PRStatus rval_status;
   1081 
   1082  PR_Lock(tpool->jobq.lock);
   1083  while (!tpool->shutdown) {
   1084    PR_WaitCondVar(tpool->shutdown_cv, PR_INTERVAL_NO_TIMEOUT);
   1085  }
   1086 
   1087  /*
   1088   * wakeup worker threads
   1089   */
   1090 #ifdef OPT_WINNT
   1091  /*
   1092   * post shutdown notification for all threads
   1093   */
   1094  {
   1095    int i;
   1096    for (i = 0; i < tpool->current_threads; i++) {
   1097      PostQueuedCompletionStatus(tpool->jobq.nt_completion_port, 0, TRUE, NULL);
   1098    }
   1099  }
   1100 #else
   1101  PR_NotifyAllCondVar(tpool->jobq.cv);
   1102 #endif
   1103 
   1104  /*
   1105   * wakeup io thread(s)
   1106   */
   1107  notify_ioq(tpool);
   1108 
   1109  /*
   1110   * wakeup timer thread(s)
   1111   */
   1112  PR_Lock(tpool->timerq.lock);
   1113  notify_timerq(tpool);
   1114  PR_Unlock(tpool->timerq.lock);
   1115 
   1116  while (!PR_CLIST_IS_EMPTY(&tpool->jobq.wthreads)) {
   1117    wthread* wthrp;
   1118 
   1119    head = PR_LIST_HEAD(&tpool->jobq.wthreads);
   1120    PR_REMOVE_AND_INIT_LINK(head);
   1121    PR_Unlock(tpool->jobq.lock);
   1122    wthrp = WTHREAD_LINKS_PTR(head);
   1123    rval_status = PR_JoinThread(wthrp->thread);
   1124    PR_ASSERT(PR_SUCCESS == rval_status);
   1125    PR_DELETE(wthrp);
   1126    PR_Lock(tpool->jobq.lock);
   1127  }
   1128  PR_Unlock(tpool->jobq.lock);
   1129  while (!PR_CLIST_IS_EMPTY(&tpool->ioq.wthreads)) {
   1130    wthread* wthrp;
   1131 
   1132    head = PR_LIST_HEAD(&tpool->ioq.wthreads);
   1133    PR_REMOVE_AND_INIT_LINK(head);
   1134    wthrp = WTHREAD_LINKS_PTR(head);
   1135    rval_status = PR_JoinThread(wthrp->thread);
   1136    PR_ASSERT(PR_SUCCESS == rval_status);
   1137    PR_DELETE(wthrp);
   1138  }
   1139 
   1140  while (!PR_CLIST_IS_EMPTY(&tpool->timerq.wthreads)) {
   1141    wthread* wthrp;
   1142 
   1143    head = PR_LIST_HEAD(&tpool->timerq.wthreads);
   1144    PR_REMOVE_AND_INIT_LINK(head);
   1145    wthrp = WTHREAD_LINKS_PTR(head);
   1146    rval_status = PR_JoinThread(wthrp->thread);
   1147    PR_ASSERT(PR_SUCCESS == rval_status);
   1148    PR_DELETE(wthrp);
   1149  }
   1150 
   1151  /*
   1152   * Delete queued jobs
   1153   */
   1154  while (!PR_CLIST_IS_EMPTY(&tpool->jobq.list)) {
   1155    PRJob* jobp;
   1156 
   1157    head = PR_LIST_HEAD(&tpool->jobq.list);
   1158    PR_REMOVE_AND_INIT_LINK(head);
   1159    jobp = JOB_LINKS_PTR(head);
   1160    tpool->jobq.cnt--;
   1161    delete_job(jobp);
   1162  }
   1163 
   1164  /* delete io jobs */
   1165  while (!PR_CLIST_IS_EMPTY(&tpool->ioq.list)) {
   1166    PRJob* jobp;
   1167 
   1168    head = PR_LIST_HEAD(&tpool->ioq.list);
   1169    PR_REMOVE_AND_INIT_LINK(head);
   1170    tpool->ioq.cnt--;
   1171    jobp = JOB_LINKS_PTR(head);
   1172    delete_job(jobp);
   1173  }
   1174 
   1175  /* delete timer jobs */
   1176  while (!PR_CLIST_IS_EMPTY(&tpool->timerq.list)) {
   1177    PRJob* jobp;
   1178 
   1179    head = PR_LIST_HEAD(&tpool->timerq.list);
   1180    PR_REMOVE_AND_INIT_LINK(head);
   1181    tpool->timerq.cnt--;
   1182    jobp = JOB_LINKS_PTR(head);
   1183    delete_job(jobp);
   1184  }
   1185 
   1186  PR_ASSERT(0 == tpool->jobq.cnt);
   1187  PR_ASSERT(0 == tpool->ioq.cnt);
   1188  PR_ASSERT(0 == tpool->timerq.cnt);
   1189 
   1190  delete_threadpool(tpool);
   1191  return rval;
   1192 }
	tor-browser The Tor Browser
	git clone https://git.dasho.dev/tor-browser.git
	Log \| Files \| Refs \| README \| LICENSE