tor  0.4.2.0-alpha-dev
Data Structures | Macros | Typedefs | Functions
workqueue.c File Reference
#include "orconfig.h"
#include "lib/evloop/compat_libevent.h"
#include "lib/evloop/workqueue.h"
#include "lib/crypt_ops/crypto_rand.h"
#include "lib/intmath/weakrng.h"
#include "lib/log/ratelim.h"
#include "lib/log/log.h"
#include "lib/log/util_bug.h"
#include "lib/net/alertsock.h"
#include "lib/net/socket.h"
#include "lib/thread/threads.h"
#include "ext/tor_queue.h"
#include <event2/event.h>
#include <string.h>

Go to the source code of this file.

Data Structures

struct  threadpool_s
 
struct  workqueue_entry_s
 
struct  replyqueue_s
 
struct  workerthread_s
 

Macros

#define WORKQUEUE_PRIORITY_FIRST   WQ_PRI_HIGH
 
#define WORKQUEUE_PRIORITY_LAST   WQ_PRI_LOW
 
#define WORKQUEUE_N_PRIORITIES   (((int) WORKQUEUE_PRIORITY_LAST)+1)
 
#define workqueue_priority_bitfield_t   ENUM_BF(workqueue_priority_t)
 
#define WORKQUEUE_PRIORITY_BITS   2
 
#define workqueue_entry_free(ent)   FREE_AND_NULL(workqueue_entry_t, workqueue_entry_free_, (ent))
 
#define MAX_THREADS   1024
 
#define CHANCE_PERMISSIVE   37
 
#define CHANCE_STRICT   INT32_MAX
 

Typedefs

typedef struct work_tailq_t work_tailq_t
 
typedef struct workerthread_s workerthread_t
 

Functions

 TOR_TAILQ_HEAD (work_tailq_t, workqueue_entry_s)
 
static void queue_reply (replyqueue_t *queue, workqueue_entry_t *work)
 
static workqueue_entry_tworkqueue_entry_new (workqueue_reply_t(*fn)(void *, void *), void(*reply_fn)(void *), void *arg)
 
static void workqueue_entry_free_ (workqueue_entry_t *ent)
 
void * workqueue_entry_cancel (workqueue_entry_t *ent)
 
static int worker_thread_has_work (workerthread_t *thread)
 
static workqueue_entry_tworker_thread_extract_next_work (workerthread_t *thread)
 
static void worker_thread_main (void *thread_)
 
static workerthread_tworkerthread_new (int32_t lower_priority_chance, void *state, threadpool_t *pool, replyqueue_t *replyqueue)
 
workqueue_entry_tthreadpool_queue_work_priority (threadpool_t *pool, workqueue_priority_t prio, workqueue_reply_t(*fn)(void *, void *), void(*reply_fn)(void *), void *arg)
 
workqueue_entry_tthreadpool_queue_work (threadpool_t *pool, workqueue_reply_t(*fn)(void *, void *), void(*reply_fn)(void *), void *arg)
 
int threadpool_queue_update (threadpool_t *pool, void *(*dup_fn)(void *), workqueue_reply_t(*fn)(void *, void *), void(*free_fn)(void *), void *arg)
 
static int threadpool_start_threads (threadpool_t *pool, int n)
 
threadpool_tthreadpool_new (int n_threads, replyqueue_t *replyqueue, void *(*new_thread_state_fn)(void *), void(*free_thread_state_fn)(void *), void *arg)
 
replyqueue_tthreadpool_get_replyqueue (threadpool_t *tp)
 
replyqueue_treplyqueue_new (uint32_t alertsocks_flags)
 
static void reply_event_cb (evutil_socket_t sock, short events, void *arg)
 
int threadpool_register_reply_event (threadpool_t *tp, void(*cb)(threadpool_t *tp))
 
void replyqueue_process (replyqueue_t *queue)
 

Detailed Description

Implements worker threads, queues of work for them, and mechanisms for them to send answers back to the main thread.

The main structure here is a threadpool_t : it manages a set of worker threads, a queue of pending work, and a reply queue. Every piece of work is a workqueue_entry_t, containing data to process and a function to process it with.

The main thread informs the worker threads of pending work by using a condition variable. The workers inform the main process of completed work by using an alert_sockets_t object, as implemented in net/alertsock.c.

The main thread can also queue an "update" that will be handled by all the workers. This is useful for updating state that all the workers share.

In Tor today, there is currently only one thread pool, used in cpuworker.c.

Definition in file workqueue.c.

Macro Definition Documentation

◆ CHANCE_PERMISSIVE

#define CHANCE_PERMISSIVE   37

For half of our threads, choose lower priority queues with probability 1/N for each of these values. Both are chosen somewhat arbitrarily. If CHANCE_PERMISSIVE is too low, then we have a risk of low-priority tasks stalling forever. If it's too high, we have a risk of low-priority tasks grabbing half of the threads.

Definition at line 489 of file workqueue.c.

◆ MAX_THREADS

#define MAX_THREADS   1024

Don't have more than this many threads per pool.

Definition at line 482 of file workqueue.c.

◆ workqueue_priority_bitfield_t

#define workqueue_priority_bitfield_t   ENUM_BF(workqueue_priority_t)

Used to put a workqueue_priority_t value into a bitfield.

Definition at line 91 of file workqueue.c.

◆ WORKQUEUE_PRIORITY_BITS

#define WORKQUEUE_PRIORITY_BITS   2

Number of bits needed to hold all legal values of workqueue_priority_t

Definition at line 93 of file workqueue.c.

Typedef Documentation

◆ workerthread_t

A worker thread represents a single thread in a thread pool.

Function Documentation

◆ queue_reply()

static void queue_reply ( replyqueue_t queue,
workqueue_entry_t work 
)
static

Put a reply on the reply queue. The reply must not currently be on any thread's work queue.

Definition at line 324 of file workqueue.c.

References replyqueue_s::lock, tor_mutex_acquire(), tor_mutex_release(), and threadpool_s::work.

◆ reply_event_cb()

static void reply_event_cb ( evutil_socket_t  sock,
short  events,
void *  arg 
)
static

Internal: Run from the libevent mainloop when there is work to handle in the reply queue handler.

Definition at line 607 of file workqueue.c.

References threadpool_s::reply_queue, and replyqueue_process().

◆ replyqueue_new()

replyqueue_t* replyqueue_new ( uint32_t  alertsocks_flags)

Allocate a new reply queue. Reply queues are used to pass results from worker threads to the main thread. Since the main thread is running an IO-centric event loop, it needs to get woken up with means other than a condition variable.

Definition at line 586 of file workqueue.c.

◆ replyqueue_process()

void replyqueue_process ( replyqueue_t queue)

Process all pending replies on a reply queue. The main thread should call this function every time the socket returned by replyqueue_get_socket() is readable.

Definition at line 647 of file workqueue.c.

Referenced by reply_event_cb().

◆ threadpool_get_replyqueue()

replyqueue_t* threadpool_get_replyqueue ( threadpool_t tp)

Return the reply queue associated with a given thread pool.

Definition at line 576 of file workqueue.c.

References threadpool_s::reply_queue.

◆ threadpool_new()

threadpool_t* threadpool_new ( int  n_threads,
replyqueue_t replyqueue,
void *(*)(void *)  new_thread_state_fn,
void(*)(void *)  free_thread_state_fn,
void *  arg 
)

Construct a new thread pool with n worker threads, configured to send their output to replyqueue. The threads' states will be constructed with the new_thread_state_fn call, receiving arg as its argument. When the threads close, they will call free_thread_state_fn on their states.

Definition at line 541 of file workqueue.c.

◆ threadpool_queue_update()

int threadpool_queue_update ( threadpool_t pool,
void *(*)(void *)  dup_fn,
workqueue_reply_t(*)(void *, void *)  fn,
void(*)(void *)  free_fn,
void *  arg 
)

Queue a copy of a work item for every thread in a pool. This can be used, for example, to tell the threads to update some parameter in their states.

Arguments are as for threadpool_queue_work, except that the arg value is passed to dup_fn once per each thread to make a copy of it.

UPDATE FUNCTIONS MUST BE IDEMPOTENT. We do not guarantee that every update will be run. If a new update is scheduled before the old update finishes running, then the new will replace the old in any threads that haven't run it yet.

Return 0 on success, -1 on failure.

Definition at line 437 of file workqueue.c.

References threadpool_s::free_update_arg_fn, threadpool_s::lock, threadpool_s::n_threads, tor_mutex_acquire(), and threadpool_s::update_args.

◆ threadpool_queue_work()

workqueue_entry_t* threadpool_queue_work ( threadpool_t pool,
workqueue_reply_t(*)(void *, void *)  fn,
void(*)(void *)  reply_fn,
void *  arg 
)

As threadpool_queue_work_priority(), but assumes WQ_PRI_HIGH

Definition at line 413 of file workqueue.c.

References threadpool_queue_work_priority().

◆ threadpool_queue_work_priority()

workqueue_entry_t* threadpool_queue_work_priority ( threadpool_t pool,
workqueue_priority_t  prio,
workqueue_reply_t(*)(void *, void *)  fn,
void(*)(void *)  reply_fn,
void *  arg 
)

Queue an item of work for a thread in a thread pool. The function fn will be run in a worker thread, and will receive as arguments the thread's state object, and the provided object arg. It must return one of WQ_RPL_REPLY, WQ_RPL_ERROR, or WQ_RPL_SHUTDOWN.

Regardless of its return value, the function reply_fn will later be run in the main thread when it invokes replyqueue_process(), and will receive as its argument the same arg object. It's the reply function's responsibility to free the work object.

On success, return a workqueue_entry_t object that can be passed to workqueue_entry_cancel(). On failure, return NULL. (Failure is not currently possible, but callers should check anyway.)

Items are executed in a loose priority order – each thread will usually take from the queued work with the highest prioirity, but will occasionally visit lower-priority queues to keep them from starving completely.

Note that because of priorities and thread behavior, work items may not be executed strictly in order.

Definition at line 386 of file workqueue.c.

References tor_assert().

Referenced by threadpool_queue_work().

◆ threadpool_register_reply_event()

int threadpool_register_reply_event ( threadpool_t tp,
void(*)(threadpool_t *tp)  cb 
)

Register the threadpool tp's reply queue with Tor's global libevent mainloop. If cb is provided, it is run after each time there is work to process from the reply queue. Return 0 on success, -1 on failure.

Definition at line 623 of file workqueue.c.

References threadpool_s::reply_event.

◆ threadpool_start_threads()

static int threadpool_start_threads ( threadpool_t pool,
int  n 
)
static

Launch threads until we have n.

Definition at line 494 of file workqueue.c.

◆ worker_thread_extract_next_work()

static workqueue_entry_t* worker_thread_extract_next_work ( workerthread_t thread)
static

Extract the next workqueue_entry_t from the the thread's pool, removing it from the relevant queues and marking it as non-pending.

The caller must hold the lock.

Definition at line 229 of file workqueue.c.

References workerthread_s::in_pool.

◆ worker_thread_has_work()

static int worker_thread_has_work ( workerthread_t thread)
static

DOCDOC

must hold lock

Definition at line 214 of file workqueue.c.

Referenced by worker_thread_main().

◆ worker_thread_main()

static void worker_thread_main ( void *  thread_)
static

◆ workerthread_new()

static workerthread_t* workerthread_new ( int32_t  lower_priority_chance,
void *  state,
threadpool_t pool,
replyqueue_t replyqueue 
)
static

Allocate and start a new worker thread to use state object state, and send responses to replyqueue.

Definition at line 342 of file workqueue.c.

◆ workqueue_entry_cancel()

void* workqueue_entry_cancel ( workqueue_entry_t ent)

Cancel a workqueue_entry_t that has been returned from threadpool_queue_work.

You must not call this function on any work whose reply function has been executed in the main thread; that will cause undefined behavior (probably, a crash).

If the work is cancelled, this function return the argument passed to the work function. It is the caller's responsibility to free this storage.

This function will have no effect if the worker thread has already executed or begun to execute the work item. In that case, it will return NULL.

Definition at line 191 of file workqueue.c.

References tor_mutex_acquire(), and tor_mutex_release().

Referenced by cpuworker_cancel_circ_handshake().

◆ workqueue_entry_free_()

static void workqueue_entry_free_ ( workqueue_entry_t ent)
static

Release all storage held in ent. Call only when ent is not on any queue.

Definition at line 168 of file workqueue.c.

References tor_free.

◆ workqueue_entry_new()

static workqueue_entry_t* workqueue_entry_new ( workqueue_reply_t(*)(void *, void *)  fn,
void(*)(void *)  reply_fn,
void *  arg 
)
static

Allocate and return a new workqueue_entry_t, set up to run the function fn in the worker thread, and reply_fn in the main thread. See threadpool_queue_work() for full documentation.

Definition at line 148 of file workqueue.c.