#include "uv-common.h"
#include "unix/internal.h"
#include <stdlib.h>

Macros
#define	MAX_THREADPOOL_SIZE 1024

Functions
static unsigned int	slow_work_thread_threshold (void)

static void	uv__cancelled (struct uv__work *w)

static void	worker (void *arg)

static void	post (QUEUE *q, enum uv__work_kind kind)

void	uv__threadpool_cleanup (void)

static void	init_threads (void)

static void	reset_once (void)

static void	init_once (void)

void	uv__work_submit (uv_loop_t loop, struct uv__work w, enum uv__work_kind kind, void(work)(struct uv__work w), void(done)(struct uv__work w, int status))

static int	uv__work_cancel (uv_loop_t loop, uv_req_t req, struct uv__work *w)

void	uv__work_done (uv_async_t *handle)

static void	uv__queue_work (struct uv__work *w)

static void	uv__queue_done (struct uv__work *w, int err)

int	uv_queue_work (uv_loop_t loop, uv_work_t req, uv_work_cb work_cb, uv_after_work_cb after_work_cb)

int	uv_cancel (uv_req_t *req)

Variables
static uv_once_t	once = UV_ONCE_INIT

static uv_cond_t	cond

static uv_mutex_t	mutex

static unsigned int	idle_threads

static unsigned int	slow_io_work_running

static unsigned int	nthreads

static uv_thread_t *	threads

static uv_thread_t	default_threads [4]

static QUEUE	exit_message

static QUEUE	wq

static QUEUE	run_slow_work_message

static QUEUE	slow_io_pending_wq

Macro Definition Documentation

◆ MAX_THREADPOOL_SIZE

#define MAX_THREADPOOL_SIZE 1024

Definition at line 30 of file threadpool.c.

Function Documentation

◆ init_once()

static void init_once ( void )

static

Definition at line 243 of file threadpool.c.

                             {
 #ifndef _WIN32
   /* Re-initialize the threadpool after fork.
    * Note that this discards the global mutex and condition as well
    * as the work queue.
    */
   if (pthread_atfork(NULL, NULL, &reset_once))
     abort();
 #endif
   init_threads();
 }

References init_threads(), NULL, and reset_once().

Referenced by uv__work_submit().

◆ init_threads()

static void init_threads ( void )

static

Definition at line 188 of file threadpool.c.

                                {
   unsigned int i;
   const char* val;
   uv_sem_t sem;
  
   nthreads = ARRAY_SIZE(default_threads);
   val = getenv("UV_THREADPOOL_SIZE");
   if (val != NULL)
     nthreads = atoi(val);
   if (nthreads == 0)
     nthreads = 1;
   if (nthreads > MAX_THREADPOOL_SIZE)
     nthreads = MAX_THREADPOOL_SIZE;
  
   threads = default_threads;
   if (nthreads > ARRAY_SIZE(default_threads)) {
     threads = uv__malloc(nthreads * sizeof(threads[0]));
     if (threads == NULL) {
       nthreads = ARRAY_SIZE(default_threads);
       threads = default_threads;
     }
   }
  
   if (uv_cond_init(&cond))
     abort();
  
   if (uv_mutex_init(&mutex))
     abort();
  
   QUEUE_INIT(&wq);
   QUEUE_INIT(&slow_io_pending_wq);
   QUEUE_INIT(&run_slow_work_message);
  
   if (uv_sem_init(&sem, 0))
     abort();
  
   for (i = 0; i < nthreads; i++)
     if (uv_thread_create(threads + i, worker, &sem))
       abort();
  
   for (i = 0; i < nthreads; i++)
     uv_sem_wait(&sem);
  
   uv_sem_destroy(&sem);
 }

References ARRAY_SIZE, cond, default_threads, getenv(), i, MAX_THREADPOOL_SIZE, mutex, nthreads, NULL, QUEUE_INIT, run_slow_work_message, slow_io_pending_wq, threads, uv__malloc(), uv_cond_init(), uv_mutex_init(), uv_sem_destroy(), uv_sem_init(), uv_sem_wait(), uv_thread_create(), val, worker(), and wq.

Referenced by init_once().

◆ post()

static void post	(	QUEUE *	q,
		enum uv__work_kind	kind
	)

static

Definition at line 142 of file threadpool.c.

                                                     {
   uv_mutex_lock(&mutex);
   if (kind == UV__WORK_SLOW_IO) {
     /* Insert into a separate queue. */
     QUEUE_INSERT_TAIL(&slow_io_pending_wq, q);
     if (!QUEUE_EMPTY(&run_slow_work_message)) {
       /* Running slow I/O tasks is already scheduled => Nothing to do here.
          The worker that runs said other task will schedule this one as well. */
       uv_mutex_unlock(&mutex);
       return;
     }
     q = &run_slow_work_message;
   }
  
   QUEUE_INSERT_TAIL(&wq, q);
   if (idle_threads > 0)
     uv_cond_signal(&cond);
   uv_mutex_unlock(&mutex);
 }

References cond, idle_threads, mutex, QUEUE_EMPTY, QUEUE_INSERT_TAIL, run_slow_work_message, slow_io_pending_wq, UV__WORK_SLOW_IO, uv_cond_signal(), uv_mutex_lock(), uv_mutex_unlock(), and wq.

Referenced by cs_op_index(), sh_apply_effects(), sh_il_set_param_pc_ctx(), uv__threadpool_cleanup(), and uv__work_submit().

◆ reset_once()

static void reset_once ( void )

static

Definition at line 236 of file threadpool.c.

                              {
   uv_once_t child_once = UV_ONCE_INIT;
   memcpy(&once, &child_once, sizeof(child_once));
 }

References memcpy(), once, and UV_ONCE_INIT.

Referenced by init_once().

◆ slow_work_thread_threshold()

static unsigned int slow_work_thread_threshold ( void )

static

Definition at line 45 of file threadpool.c.

                                                      {
   return (nthreads + 1) / 2;
 }

References nthreads.

Referenced by worker().

◆ uv__cancelled()

static void uv__cancelled ( struct uv__work * w )

static

Definition at line 49 of file threadpool.c.

                                               {
   abort();
 }

Referenced by uv__work_cancel(), and uv__work_done().

◆ uv__queue_done()

static void uv__queue_done	(	struct uv__work *	w,
		int	err
	)

static

Definition at line 325 of file threadpool.c.

                                                         {
   uv_work_t* req;
  
   req = container_of(w, uv_work_t, work_req);
   uv__req_unregister(req->loop, req);
  
   if (req->after_work_cb == NULL)
     return;
  
   req->after_work_cb(req, err);
 }

References container_of, err, NULL, req, uv__req_unregister, and w.

Referenced by uv_queue_work().

◆ uv__queue_work()

static void uv__queue_work ( struct uv__work * w )

static

Definition at line 318 of file threadpool.c.

                                                {
   uv_work_t* req = container_of(w, uv_work_t, work_req);
  
   req->work_cb(req);
 }

References container_of, req, and w.

Referenced by uv_queue_work().

◆ uv__threadpool_cleanup()

void uv__threadpool_cleanup ( void )

Definition at line 163 of file threadpool.c.

                                   {
 #ifndef _WIN32
   unsigned int i;
  
   if (nthreads == 0)
     return;
  
   post(&exit_message, UV__WORK_CPU);
  
   for (i = 0; i < nthreads; i++)
     if (uv_thread_join(threads + i))
       abort();
  
   if (threads != default_threads)
     uv__free(threads);
  
   uv_mutex_destroy(&mutex);
   uv_cond_destroy(&cond);
  
   threads = NULL;
   nthreads = 0;
 #endif
 }

References cond, default_threads, exit_message, i, mutex, nthreads, NULL, post(), threads, uv__free(), UV__WORK_CPU, uv_cond_destroy(), uv_mutex_destroy(), and uv_thread_join().

Referenced by uv_library_shutdown().

◆ uv__work_cancel()

static int uv__work_cancel	(	uv_loop_t *	loop,
		uv_req_t *	req,
		struct uv__work *	w
	)

static

Definition at line 269 of file threadpool.c.

                                                                                {
   int cancelled;
  
   uv_mutex_lock(&mutex);
   uv_mutex_lock(&w->loop->wq_mutex);
  
   cancelled = !QUEUE_EMPTY(&w->wq) && w->work != NULL;
   if (cancelled)
     QUEUE_REMOVE(&w->wq);
  
   uv_mutex_unlock(&w->loop->wq_mutex);
   uv_mutex_unlock(&mutex);
  
   if (!cancelled)
     return UV_EBUSY;
  
   w->work = uv__cancelled;
   uv_mutex_lock(&loop->wq_mutex);
   QUEUE_INSERT_TAIL(&loop->wq, &w->wq);
   uv_async_send(&loop->wq_async);
   uv_mutex_unlock(&loop->wq_mutex);
  
   return 0;
 }

References loop, mutex, NULL, QUEUE_EMPTY, QUEUE_INSERT_TAIL, QUEUE_REMOVE, uv__cancelled(), uv_async_send(), uv_mutex_lock(), uv_mutex_unlock(), and w.

Referenced by uv_cancel().

◆ uv__work_done()

void uv__work_done ( uv_async_t * handle )

Definition at line 295 of file threadpool.c.

                                        {
   struct uv__work* w;
   uv_loop_t* loop;
   QUEUE* q;
   QUEUE wq;
   int err;
  
   loop = container_of(handle, uv_loop_t, wq_async);
   uv_mutex_lock(&loop->wq_mutex);
   QUEUE_MOVE(&loop->wq, &wq);
   uv_mutex_unlock(&loop->wq_mutex);
  
   while (!QUEUE_EMPTY(&wq)) {
     q = QUEUE_HEAD(&wq);
     QUEUE_REMOVE(q);
  
     w = container_of(q, struct uv__work, wq);
     err = (w->work == uv__cancelled) ? UV_ECANCELED : 0;
     w->done(w, err);
   }
 }

References container_of, err, handle, loop, QUEUE_EMPTY, QUEUE_HEAD, QUEUE_MOVE, QUEUE_REMOVE, uv__cancelled(), uv_mutex_lock(), uv_mutex_unlock(), w, and wq.

Referenced by uv_loop_init().

◆ uv__work_submit()

void uv__work_submit	(	uv_loop_t *	loop,
		struct uv__work *	w,
		enum uv__work_kind	kind,
		void()(struct uv__work w)	work,
		void()(struct uv__work w, int status)	done
	)

Definition at line 256 of file threadpool.c.

                                                                    {
   uv_once(&once, init_once);
   w->loop = loop;
   w->work = work;
   w->done = done;
   post(&w->wq, kind);
 }

References done, init_once(), loop, once, post(), uv_once(), w, and uv__work::work.

Referenced by uv_getaddrinfo(), uv_getnameinfo(), uv_queue_work(), and uv_random().

◆ uv_cancel()

int uv_cancel ( uv_req_t * req )

Definition at line 358 of file threadpool.c.

                              {
   struct uv__work* wreq;
   uv_loop_t* loop;
  
   switch (req->type) {
   case UV_FS:
     loop =  ((uv_fs_t*) req)->loop;
     wreq = &((uv_fs_t*) req)->work_req;
     break;
   case UV_GETADDRINFO:
     loop =  ((uv_getaddrinfo_t*) req)->loop;
     wreq = &((uv_getaddrinfo_t*) req)->work_req;
     break;
   case UV_GETNAMEINFO:
     loop = ((uv_getnameinfo_t*) req)->loop;
     wreq = &((uv_getnameinfo_t*) req)->work_req;
     break;
   case UV_RANDOM:
     loop = ((uv_random_t*) req)->loop;
     wreq = &((uv_random_t*) req)->work_req;
     break;
   case UV_WORK:
     loop =  ((uv_work_t*) req)->loop;
     wreq = &((uv_work_t*) req)->work_req;
     break;
   default:
     return UV_EINVAL;
   }
  
   return uv__work_cancel(loop, req, wreq);
 }

References loop, req, and uv__work_cancel().

Referenced by signal_handler().

◆ uv_queue_work()

int uv_queue_work	(	uv_loop_t *	loop,
		uv_work_t *	req,
		uv_work_cb	work_cb,
		uv_after_work_cb	after_work_cb
	)

Definition at line 338 of file threadpool.c.

                                                   {
   if (work_cb == NULL)
     return UV_EINVAL;
  
   uv__req_init(loop, req, UV_WORK);
   req->loop = loop;
   req->work_cb = work_cb;
   req->after_work_cb = after_work_cb;
   uv__work_submit(loop,
                   &req->work_req,
                   UV__WORK_CPU,
                   uv__queue_work,
                   uv__queue_done);
   return 0;
 }

References loop, NULL, req, uv__queue_done(), uv__queue_work(), uv__req_init, UV__WORK_CPU, and uv__work_submit().

Referenced by main().

◆ worker()

static void worker ( void * arg )

static

Definition at line 57 of file threadpool.c.

                               {
   struct uv__work* w;
   QUEUE* q;
   int is_slow_work;
  
   uv_sem_post((uv_sem_t*) arg);
   arg = NULL;
  
   uv_mutex_lock(&mutex);
   for (;;) {
     /* `mutex` should always be locked at this point. */
  
     /* Keep waiting while either no work is present or only slow I/O
        and we're at the threshold for that. */
     while (QUEUE_EMPTY(&wq) ||
            (QUEUE_HEAD(&wq) == &run_slow_work_message &&
             QUEUE_NEXT(&run_slow_work_message) == &wq &&
             slow_io_work_running >= slow_work_thread_threshold())) {
       idle_threads += 1;
       uv_cond_wait(&cond, &mutex);
       idle_threads -= 1;
     }
  
     q = QUEUE_HEAD(&wq);
     if (q == &exit_message) {
       uv_cond_signal(&cond);
       uv_mutex_unlock(&mutex);
       break;
     }
  
     QUEUE_REMOVE(q);
     QUEUE_INIT(q);  /* Signal uv_cancel() that the work req is executing. */
  
     is_slow_work = 0;
     if (q == &run_slow_work_message) {
       /* If we're at the slow I/O threshold, re-schedule until after all
          other work in the queue is done. */
       if (slow_io_work_running >= slow_work_thread_threshold()) {
         QUEUE_INSERT_TAIL(&wq, q);
         continue;
       }
  
       /* If we encountered a request to run slow I/O work but there is none
          to run, that means it's cancelled => Start over. */
       if (QUEUE_EMPTY(&slow_io_pending_wq))
         continue;
  
       is_slow_work = 1;
       slow_io_work_running++;
  
       q = QUEUE_HEAD(&slow_io_pending_wq);
       QUEUE_REMOVE(q);
       QUEUE_INIT(q);
  
       /* If there is more slow I/O work, schedule it to be run as well. */
       if (!QUEUE_EMPTY(&slow_io_pending_wq)) {
         QUEUE_INSERT_TAIL(&wq, &run_slow_work_message);
         if (idle_threads > 0)
           uv_cond_signal(&cond);
       }
     }
  
     uv_mutex_unlock(&mutex);
  
     w = QUEUE_DATA(q, struct uv__work, wq);
     w->work(w);
  
     uv_mutex_lock(&w->loop->wq_mutex);
     w->work = NULL;  /* Signal uv_cancel() that the work req is done
                         executing. */
     QUEUE_INSERT_TAIL(&w->loop->wq, &w->wq);
     uv_async_send(&w->loop->wq_async);
     uv_mutex_unlock(&w->loop->wq_mutex);
  
     /* Lock `mutex` since that is expected at the start of the next
      * iteration. */
     uv_mutex_lock(&mutex);
     if (is_slow_work) {
       /* `slow_io_work_running` is protected by `mutex`. */
       slow_io_work_running--;
     }
   }
 }