workqueue.c 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539
  1. /* copyright (c) 2013-2015, The Tor Project, Inc. */
  2. /* See LICENSE for licensing information */
  3. /**
  4. * \file workqueue.c
  5. *
  6. * \brief Implements worker threads, queues of work for them, and mechanisms
  7. * for them to send answers back to the main thread.
  8. *
  9. * The main structure here is a threadpool_t : it manages a set of worker
  10. * threads, a queue of pending work, and a reply queue. Every piece of work
  11. * is a workqueue_entry_t, containing data to process and a function to
  12. * process it with.
  13. *
  14. * The main thread informs the worker threads of pending work by using a
  15. * condition variable. The workers inform the main process of completed work
  16. * by using an alert_sockets_t object, as implemented in compat_threads.c.
  17. *
  18. * The main thread can also queue an "update" that will be handled by all the
  19. * workers. This is useful for updating state that all the workers share.
  20. *
  21. * In Tor today, there is currently only one thread pool, used in cpuworker.c.
  22. */
  23. #include "orconfig.h"
  24. #include "compat.h"
  25. #include "compat_threads.h"
  26. #include "util.h"
  27. #include "workqueue.h"
  28. #include "tor_queue.h"
  29. #include "torlog.h"
  30. struct threadpool_s {
  31. /** An array of pointers to workerthread_t: one for each running worker
  32. * thread. */
  33. struct workerthread_s **threads;
  34. /** Condition variable that we wait on when we have no work, and which
  35. * gets signaled when our queue becomes nonempty. */
  36. tor_cond_t condition;
  37. /** Queue of pending work that we have to do. */
  38. TOR_TAILQ_HEAD(, workqueue_entry_s) work;
  39. /** The current 'update generation' of the threadpool. Any thread that is
  40. * at an earlier generation needs to run the update function. */
  41. unsigned generation;
  42. /** Function that should be run for updates on each thread. */
  43. workqueue_reply_t (*update_fn)(void *, void *);
  44. /** Function to free update arguments if they can't be run. */
  45. void (*free_update_arg_fn)(void *);
  46. /** Array of n_threads update arguments. */
  47. void **update_args;
  48. /** Number of elements in threads. */
  49. int n_threads;
  50. /** Mutex to protect all the above fields. */
  51. tor_mutex_t lock;
  52. /** A reply queue to use when constructing new threads. */
  53. replyqueue_t *reply_queue;
  54. /** Functions used to allocate and free thread state. */
  55. void *(*new_thread_state_fn)(void*);
  56. void (*free_thread_state_fn)(void*);
  57. void *new_thread_state_arg;
  58. };
  59. struct workqueue_entry_s {
  60. /** The next workqueue_entry_t that's pending on the same thread or
  61. * reply queue. */
  62. TOR_TAILQ_ENTRY(workqueue_entry_s) next_work;
  63. /** The threadpool to which this workqueue_entry_t was assigned. This field
  64. * is set when the workqueue_entry_t is created, and won't be cleared until
  65. * after it's handled in the main thread. */
  66. struct threadpool_s *on_pool;
  67. /** True iff this entry is waiting for a worker to start processing it. */
  68. uint8_t pending;
  69. /** Function to run in the worker thread. */
  70. workqueue_reply_t (*fn)(void *state, void *arg);
  71. /** Function to run while processing the reply queue. */
  72. void (*reply_fn)(void *arg);
  73. /** Argument for the above functions. */
  74. void *arg;
  75. };
  76. struct replyqueue_s {
  77. /** Mutex to protect the answers field */
  78. tor_mutex_t lock;
  79. /** Doubly-linked list of answers that the reply queue needs to handle. */
  80. TOR_TAILQ_HEAD(, workqueue_entry_s) answers;
  81. /** Mechanism to wake up the main thread when it is receiving answers. */
  82. alert_sockets_t alert;
  83. };
  84. /** A worker thread represents a single thread in a thread pool. To avoid
  85. * contention, each gets its own queue. This breaks the guarantee that that
  86. * queued work will get executed strictly in order. */
  87. typedef struct workerthread_s {
  88. /** Which thread it this? In range 0..in_pool->n_threads-1 */
  89. int index;
  90. /** The pool this thread is a part of. */
  91. struct threadpool_s *in_pool;
  92. /** User-supplied state field that we pass to the worker functions of each
  93. * work item. */
  94. void *state;
  95. /** Reply queue to which we pass our results. */
  96. replyqueue_t *reply_queue;
  97. /** The current update generation of this thread */
  98. unsigned generation;
  99. } workerthread_t;
  100. static void queue_reply(replyqueue_t *queue, workqueue_entry_t *work);
  101. /** Allocate and return a new workqueue_entry_t, set up to run the function
  102. * <b>fn</b> in the worker thread, and <b>reply_fn</b> in the main
  103. * thread. See threadpool_queue_work() for full documentation. */
  104. static workqueue_entry_t *
  105. workqueue_entry_new(workqueue_reply_t (*fn)(void*, void*),
  106. void (*reply_fn)(void*),
  107. void *arg)
  108. {
  109. workqueue_entry_t *ent = tor_malloc_zero(sizeof(workqueue_entry_t));
  110. ent->fn = fn;
  111. ent->reply_fn = reply_fn;
  112. ent->arg = arg;
  113. return ent;
  114. }
  115. /**
  116. * Release all storage held in <b>ent</b>. Call only when <b>ent</b> is not on
  117. * any queue.
  118. */
  119. static void
  120. workqueue_entry_free(workqueue_entry_t *ent)
  121. {
  122. if (!ent)
  123. return;
  124. memset(ent, 0xf0, sizeof(*ent));
  125. tor_free(ent);
  126. }
  127. /**
  128. * Cancel a workqueue_entry_t that has been returned from
  129. * threadpool_queue_work.
  130. *
  131. * You must not call this function on any work whose reply function has been
  132. * executed in the main thread; that will cause undefined behavior (probably,
  133. * a crash).
  134. *
  135. * If the work is cancelled, this function return the argument passed to the
  136. * work function. It is the caller's responsibility to free this storage.
  137. *
  138. * This function will have no effect if the worker thread has already executed
  139. * or begun to execute the work item. In that case, it will return NULL.
  140. */
  141. void *
  142. workqueue_entry_cancel(workqueue_entry_t *ent)
  143. {
  144. int cancelled = 0;
  145. void *result = NULL;
  146. tor_mutex_acquire(&ent->on_pool->lock);
  147. if (ent->pending) {
  148. TOR_TAILQ_REMOVE(&ent->on_pool->work, ent, next_work);
  149. cancelled = 1;
  150. result = ent->arg;
  151. }
  152. tor_mutex_release(&ent->on_pool->lock);
  153. if (cancelled) {
  154. workqueue_entry_free(ent);
  155. }
  156. return result;
  157. }
  158. /**DOCDOC
  159. must hold lock */
  160. static int
  161. worker_thread_has_work(workerthread_t *thread)
  162. {
  163. return !TOR_TAILQ_EMPTY(&thread->in_pool->work) ||
  164. thread->generation != thread->in_pool->generation;
  165. }
  166. /**
  167. * Main function for the worker thread.
  168. */
  169. static void
  170. worker_thread_main(void *thread_)
  171. {
  172. workerthread_t *thread = thread_;
  173. threadpool_t *pool = thread->in_pool;
  174. workqueue_entry_t *work;
  175. workqueue_reply_t result;
  176. tor_mutex_acquire(&pool->lock);
  177. while (1) {
  178. /* lock must be held at this point. */
  179. while (worker_thread_has_work(thread)) {
  180. /* lock must be held at this point. */
  181. if (thread->in_pool->generation != thread->generation) {
  182. void *arg = thread->in_pool->update_args[thread->index];
  183. thread->in_pool->update_args[thread->index] = NULL;
  184. workqueue_reply_t (*update_fn)(void*,void*) =
  185. thread->in_pool->update_fn;
  186. thread->generation = thread->in_pool->generation;
  187. tor_mutex_release(&pool->lock);
  188. workqueue_reply_t r = update_fn(thread->state, arg);
  189. if (r != WQ_RPL_REPLY) {
  190. return;
  191. }
  192. tor_mutex_acquire(&pool->lock);
  193. continue;
  194. }
  195. work = TOR_TAILQ_FIRST(&pool->work);
  196. TOR_TAILQ_REMOVE(&pool->work, work, next_work);
  197. work->pending = 0;
  198. tor_mutex_release(&pool->lock);
  199. /* We run the work function without holding the thread lock. This
  200. * is the main thread's first opportunity to give us more work. */
  201. result = work->fn(thread->state, work->arg);
  202. /* Queue the reply for the main thread. */
  203. queue_reply(thread->reply_queue, work);
  204. /* We may need to exit the thread. */
  205. if (result != WQ_RPL_REPLY) {
  206. return;
  207. }
  208. tor_mutex_acquire(&pool->lock);
  209. }
  210. /* At this point the lock is held, and there is no work in this thread's
  211. * queue. */
  212. /* TODO: support an idle-function */
  213. /* Okay. Now, wait till somebody has work for us. */
  214. if (tor_cond_wait(&pool->condition, &pool->lock, NULL) < 0) {
  215. log_warn(LD_GENERAL, "Fail tor_cond_wait.");
  216. }
  217. }
  218. }
  219. /** Put a reply on the reply queue. The reply must not currently be on
  220. * any thread's work queue. */
  221. static void
  222. queue_reply(replyqueue_t *queue, workqueue_entry_t *work)
  223. {
  224. int was_empty;
  225. tor_mutex_acquire(&queue->lock);
  226. was_empty = TOR_TAILQ_EMPTY(&queue->answers);
  227. TOR_TAILQ_INSERT_TAIL(&queue->answers, work, next_work);
  228. tor_mutex_release(&queue->lock);
  229. if (was_empty) {
  230. if (queue->alert.alert_fn(queue->alert.write_fd) < 0) {
  231. /* XXXX complain! */
  232. }
  233. }
  234. }
  235. /** Allocate and start a new worker thread to use state object <b>state</b>,
  236. * and send responses to <b>replyqueue</b>. */
  237. static workerthread_t *
  238. workerthread_new(void *state, threadpool_t *pool, replyqueue_t *replyqueue)
  239. {
  240. workerthread_t *thr = tor_malloc_zero(sizeof(workerthread_t));
  241. thr->state = state;
  242. thr->reply_queue = replyqueue;
  243. thr->in_pool = pool;
  244. if (spawn_func(worker_thread_main, thr) < 0) {
  245. //LCOV_EXCL_START
  246. tor_assert_nonfatal_unreached();
  247. log_err(LD_GENERAL, "Can't launch worker thread.");
  248. tor_free(thr);
  249. return NULL;
  250. //LCOV_EXCL_STOP
  251. }
  252. return thr;
  253. }
  254. /**
  255. * Queue an item of work for a thread in a thread pool. The function
  256. * <b>fn</b> will be run in a worker thread, and will receive as arguments the
  257. * thread's state object, and the provided object <b>arg</b>. It must return
  258. * one of WQ_RPL_REPLY, WQ_RPL_ERROR, or WQ_RPL_SHUTDOWN.
  259. *
  260. * Regardless of its return value, the function <b>reply_fn</b> will later be
  261. * run in the main thread when it invokes replyqueue_process(), and will
  262. * receive as its argument the same <b>arg</b> object. It's the reply
  263. * function's responsibility to free the work object.
  264. *
  265. * On success, return a workqueue_entry_t object that can be passed to
  266. * workqueue_entry_cancel(). On failure, return NULL.
  267. *
  268. * Note that because each thread has its own work queue, work items may not
  269. * be executed strictly in order.
  270. */
  271. workqueue_entry_t *
  272. threadpool_queue_work(threadpool_t *pool,
  273. workqueue_reply_t (*fn)(void *, void *),
  274. void (*reply_fn)(void *),
  275. void *arg)
  276. {
  277. workqueue_entry_t *ent = workqueue_entry_new(fn, reply_fn, arg);
  278. ent->on_pool = pool;
  279. ent->pending = 1;
  280. tor_mutex_acquire(&pool->lock);
  281. TOR_TAILQ_INSERT_TAIL(&pool->work, ent, next_work);
  282. tor_cond_signal_one(&pool->condition);
  283. tor_mutex_release(&pool->lock);
  284. return ent;
  285. }
  286. /**
  287. * Queue a copy of a work item for every thread in a pool. This can be used,
  288. * for example, to tell the threads to update some parameter in their states.
  289. *
  290. * Arguments are as for <b>threadpool_queue_work</b>, except that the
  291. * <b>arg</b> value is passed to <b>dup_fn</b> once per each thread to
  292. * make a copy of it.
  293. *
  294. * UPDATE FUNCTIONS MUST BE IDEMPOTENT. We do not guarantee that every update
  295. * will be run. If a new update is scheduled before the old update finishes
  296. * running, then the new will replace the old in any threads that haven't run
  297. * it yet.
  298. *
  299. * Return 0 on success, -1 on failure.
  300. */
  301. int
  302. threadpool_queue_update(threadpool_t *pool,
  303. void *(*dup_fn)(void *),
  304. workqueue_reply_t (*fn)(void *, void *),
  305. void (*free_fn)(void *),
  306. void *arg)
  307. {
  308. int i, n_threads;
  309. void (*old_args_free_fn)(void *arg);
  310. void **old_args;
  311. void **new_args;
  312. tor_mutex_acquire(&pool->lock);
  313. n_threads = pool->n_threads;
  314. old_args = pool->update_args;
  315. old_args_free_fn = pool->free_update_arg_fn;
  316. new_args = tor_calloc(n_threads, sizeof(void*));
  317. for (i = 0; i < n_threads; ++i) {
  318. if (dup_fn)
  319. new_args[i] = dup_fn(arg);
  320. else
  321. new_args[i] = arg;
  322. }
  323. pool->update_args = new_args;
  324. pool->free_update_arg_fn = free_fn;
  325. pool->update_fn = fn;
  326. ++pool->generation;
  327. tor_cond_signal_all(&pool->condition);
  328. tor_mutex_release(&pool->lock);
  329. if (old_args) {
  330. for (i = 0; i < n_threads; ++i) {
  331. if (old_args[i] && old_args_free_fn)
  332. old_args_free_fn(old_args[i]);
  333. }
  334. tor_free(old_args);
  335. }
  336. return 0;
  337. }
  338. /** Don't have more than this many threads per pool. */
  339. #define MAX_THREADS 1024
  340. /** Launch threads until we have <b>n</b>. */
  341. static int
  342. threadpool_start_threads(threadpool_t *pool, int n)
  343. {
  344. if (BUG(n < 0))
  345. return -1; // LCOV_EXCL_LINE
  346. if (n > MAX_THREADS)
  347. n = MAX_THREADS;
  348. tor_mutex_acquire(&pool->lock);
  349. if (pool->n_threads < n)
  350. pool->threads = tor_reallocarray(pool->threads,
  351. sizeof(workerthread_t*), n);
  352. while (pool->n_threads < n) {
  353. void *state = pool->new_thread_state_fn(pool->new_thread_state_arg);
  354. workerthread_t *thr = workerthread_new(state, pool, pool->reply_queue);
  355. if (!thr) {
  356. //LCOV_EXCL_START
  357. tor_assert_nonfatal_unreached();
  358. pool->free_thread_state_fn(state);
  359. tor_mutex_release(&pool->lock);
  360. return -1;
  361. //LCOV_EXCL_STOP
  362. }
  363. thr->index = pool->n_threads;
  364. pool->threads[pool->n_threads++] = thr;
  365. }
  366. tor_mutex_release(&pool->lock);
  367. return 0;
  368. }
  369. /**
  370. * Construct a new thread pool with <b>n</b> worker threads, configured to
  371. * send their output to <b>replyqueue</b>. The threads' states will be
  372. * constructed with the <b>new_thread_state_fn</b> call, receiving <b>arg</b>
  373. * as its argument. When the threads close, they will call
  374. * <b>free_thread_state_fn</b> on their states.
  375. */
  376. threadpool_t *
  377. threadpool_new(int n_threads,
  378. replyqueue_t *replyqueue,
  379. void *(*new_thread_state_fn)(void*),
  380. void (*free_thread_state_fn)(void*),
  381. void *arg)
  382. {
  383. threadpool_t *pool;
  384. pool = tor_malloc_zero(sizeof(threadpool_t));
  385. tor_mutex_init_nonrecursive(&pool->lock);
  386. tor_cond_init(&pool->condition);
  387. TOR_TAILQ_INIT(&pool->work);
  388. pool->new_thread_state_fn = new_thread_state_fn;
  389. pool->new_thread_state_arg = arg;
  390. pool->free_thread_state_fn = free_thread_state_fn;
  391. pool->reply_queue = replyqueue;
  392. if (threadpool_start_threads(pool, n_threads) < 0) {
  393. //LCOV_EXCL_START
  394. tor_assert_nonfatal_unreached();
  395. tor_cond_uninit(&pool->condition);
  396. tor_mutex_uninit(&pool->lock);
  397. tor_free(pool);
  398. return NULL;
  399. //LCOV_EXCL_STOP
  400. }
  401. return pool;
  402. }
  403. /** Return the reply queue associated with a given thread pool. */
  404. replyqueue_t *
  405. threadpool_get_replyqueue(threadpool_t *tp)
  406. {
  407. return tp->reply_queue;
  408. }
  409. /** Allocate a new reply queue. Reply queues are used to pass results from
  410. * worker threads to the main thread. Since the main thread is running an
  411. * IO-centric event loop, it needs to get woken up with means other than a
  412. * condition variable. */
  413. replyqueue_t *
  414. replyqueue_new(uint32_t alertsocks_flags)
  415. {
  416. replyqueue_t *rq;
  417. rq = tor_malloc_zero(sizeof(replyqueue_t));
  418. if (alert_sockets_create(&rq->alert, alertsocks_flags) < 0) {
  419. //LCOV_EXCL_START
  420. tor_free(rq);
  421. return NULL;
  422. //LCOV_EXCL_STOP
  423. }
  424. tor_mutex_init(&rq->lock);
  425. TOR_TAILQ_INIT(&rq->answers);
  426. return rq;
  427. }
  428. /**
  429. * Return the "read socket" for a given reply queue. The main thread should
  430. * listen for read events on this socket, and call replyqueue_process() every
  431. * time it triggers.
  432. */
  433. tor_socket_t
  434. replyqueue_get_socket(replyqueue_t *rq)
  435. {
  436. return rq->alert.read_fd;
  437. }
  438. /**
  439. * Process all pending replies on a reply queue. The main thread should call
  440. * this function every time the socket returned by replyqueue_get_socket() is
  441. * readable.
  442. */
  443. void
  444. replyqueue_process(replyqueue_t *queue)
  445. {
  446. int r = queue->alert.drain_fn(queue->alert.read_fd);
  447. if (r < 0) {
  448. //LCOV_EXCL_START
  449. static ratelim_t warn_limit = RATELIM_INIT(7200);
  450. log_fn_ratelim(&warn_limit, LOG_WARN, LD_GENERAL,
  451. "Failure from drain_fd: %s",
  452. tor_socket_strerror(-r));
  453. //LCOV_EXCL_STOP
  454. }
  455. tor_mutex_acquire(&queue->lock);
  456. while (!TOR_TAILQ_EMPTY(&queue->answers)) {
  457. /* lock must be held at this point.*/
  458. workqueue_entry_t *work = TOR_TAILQ_FIRST(&queue->answers);
  459. TOR_TAILQ_REMOVE(&queue->answers, work, next_work);
  460. tor_mutex_release(&queue->lock);
  461. work->on_pool = NULL;
  462. work->reply_fn(work->arg);
  463. workqueue_entry_free(work);
  464. tor_mutex_acquire(&queue->lock);
  465. }
  466. tor_mutex_release(&queue->lock);
  467. }