workqueue.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678
  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_libevent.h"
  26. #include "compat_threads.h"
  27. #include "crypto_rand.h"
  28. #include "util.h"
  29. #include "workqueue.h"
  30. #include "tor_queue.h"
  31. #include "torlog.h"
  32. #include <event2/event.h>
  33. #define WORKQUEUE_PRIORITY_FIRST WQ_PRI_HIGH
  34. #define WORKQUEUE_PRIORITY_LAST WQ_PRI_LOW
  35. #define WORKQUEUE_N_PRIORITIES (((int) WORKQUEUE_PRIORITY_LAST)+1)
  36. TOR_TAILQ_HEAD(work_tailq_t, workqueue_entry_s);
  37. typedef struct work_tailq_t work_tailq_t;
  38. struct threadpool_s {
  39. /** An array of pointers to workerthread_t: one for each running worker
  40. * thread. */
  41. struct workerthread_s **threads;
  42. /** Condition variable that we wait on when we have no work, and which
  43. * gets signaled when our queue becomes nonempty. */
  44. tor_cond_t condition;
  45. /** Queues of pending work that we have to do. The queue with priority
  46. * <b>p</b> is work[p]. */
  47. work_tailq_t work[WORKQUEUE_N_PRIORITIES];
  48. /** Weak RNG, used to decide when to ignore priority. */
  49. tor_weak_rng_t weak_rng;
  50. /** The current 'update generation' of the threadpool. Any thread that is
  51. * at an earlier generation needs to run the update function. */
  52. unsigned generation;
  53. /** Function that should be run for updates on each thread. */
  54. workqueue_reply_t (*update_fn)(void *, void *);
  55. /** Function to free update arguments if they can't be run. */
  56. void (*free_update_arg_fn)(void *);
  57. /** Array of n_threads update arguments. */
  58. void **update_args;
  59. /** Event to notice when another thread has sent a reply. */
  60. struct event *reply_event;
  61. void (*reply_cb)(threadpool_t *);
  62. /** Number of elements in threads. */
  63. int n_threads;
  64. /** Mutex to protect all the above fields. */
  65. tor_mutex_t lock;
  66. /** A reply queue to use when constructing new threads. */
  67. replyqueue_t *reply_queue;
  68. /** Functions used to allocate and free thread state. */
  69. void *(*new_thread_state_fn)(void*);
  70. void (*free_thread_state_fn)(void*);
  71. void *new_thread_state_arg;
  72. };
  73. /** Used to put a workqueue_priority_t value into a bitfield. */
  74. #define workqueue_priority_bitfield_t ENUM_BF(workqueue_priority_t)
  75. /** Number of bits needed to hold all legal values of workqueue_priority_t */
  76. #define WORKQUEUE_PRIORITY_BITS 2
  77. struct workqueue_entry_s {
  78. /** The next workqueue_entry_t that's pending on the same thread or
  79. * reply queue. */
  80. TOR_TAILQ_ENTRY(workqueue_entry_s) next_work;
  81. /** The threadpool to which this workqueue_entry_t was assigned. This field
  82. * is set when the workqueue_entry_t is created, and won't be cleared until
  83. * after it's handled in the main thread. */
  84. struct threadpool_s *on_pool;
  85. /** True iff this entry is waiting for a worker to start processing it. */
  86. uint8_t pending;
  87. /** Priority of this entry. */
  88. workqueue_priority_bitfield_t priority : WORKQUEUE_PRIORITY_BITS;
  89. /** Function to run in the worker thread. */
  90. workqueue_reply_t (*fn)(void *state, void *arg);
  91. /** Function to run while processing the reply queue. */
  92. void (*reply_fn)(void *arg);
  93. /** Argument for the above functions. */
  94. void *arg;
  95. };
  96. struct replyqueue_s {
  97. /** Mutex to protect the answers field */
  98. tor_mutex_t lock;
  99. /** Doubly-linked list of answers that the reply queue needs to handle. */
  100. TOR_TAILQ_HEAD(, workqueue_entry_s) answers;
  101. /** Mechanism to wake up the main thread when it is receiving answers. */
  102. alert_sockets_t alert;
  103. };
  104. /** A worker thread represents a single thread in a thread pool. */
  105. typedef struct workerthread_s {
  106. /** Which thread it this? In range 0..in_pool->n_threads-1 */
  107. int index;
  108. /** The pool this thread is a part of. */
  109. struct threadpool_s *in_pool;
  110. /** User-supplied state field that we pass to the worker functions of each
  111. * work item. */
  112. void *state;
  113. /** Reply queue to which we pass our results. */
  114. replyqueue_t *reply_queue;
  115. /** The current update generation of this thread */
  116. unsigned generation;
  117. /** One over the probability of taking work from a lower-priority queue. */
  118. int32_t lower_priority_chance;
  119. } workerthread_t;
  120. static void queue_reply(replyqueue_t *queue, workqueue_entry_t *work);
  121. /** Allocate and return a new workqueue_entry_t, set up to run the function
  122. * <b>fn</b> in the worker thread, and <b>reply_fn</b> in the main
  123. * thread. See threadpool_queue_work() for full documentation. */
  124. static workqueue_entry_t *
  125. workqueue_entry_new(workqueue_reply_t (*fn)(void*, void*),
  126. void (*reply_fn)(void*),
  127. void *arg)
  128. {
  129. workqueue_entry_t *ent = tor_malloc_zero(sizeof(workqueue_entry_t));
  130. ent->fn = fn;
  131. ent->reply_fn = reply_fn;
  132. ent->arg = arg;
  133. ent->priority = WQ_PRI_HIGH;
  134. return ent;
  135. }
  136. #define workqueue_entry_free(ent) \
  137. FREE_AND_NULL(workqueue_entry_t, workqueue_entry_free_, (ent))
  138. /**
  139. * Release all storage held in <b>ent</b>. Call only when <b>ent</b> is not on
  140. * any queue.
  141. */
  142. static void
  143. workqueue_entry_free_(workqueue_entry_t *ent)
  144. {
  145. if (!ent)
  146. return;
  147. memset(ent, 0xf0, sizeof(*ent));
  148. tor_free(ent);
  149. }
  150. /**
  151. * Cancel a workqueue_entry_t that has been returned from
  152. * threadpool_queue_work.
  153. *
  154. * You must not call this function on any work whose reply function has been
  155. * executed in the main thread; that will cause undefined behavior (probably,
  156. * a crash).
  157. *
  158. * If the work is cancelled, this function return the argument passed to the
  159. * work function. It is the caller's responsibility to free this storage.
  160. *
  161. * This function will have no effect if the worker thread has already executed
  162. * or begun to execute the work item. In that case, it will return NULL.
  163. */
  164. void *
  165. workqueue_entry_cancel(workqueue_entry_t *ent)
  166. {
  167. int cancelled = 0;
  168. void *result = NULL;
  169. tor_mutex_acquire(&ent->on_pool->lock);
  170. workqueue_priority_t prio = ent->priority;
  171. if (ent->pending) {
  172. TOR_TAILQ_REMOVE(&ent->on_pool->work[prio], ent, next_work);
  173. cancelled = 1;
  174. result = ent->arg;
  175. }
  176. tor_mutex_release(&ent->on_pool->lock);
  177. if (cancelled) {
  178. workqueue_entry_free(ent);
  179. }
  180. return result;
  181. }
  182. /**DOCDOC
  183. must hold lock */
  184. static int
  185. worker_thread_has_work(workerthread_t *thread)
  186. {
  187. unsigned i;
  188. for (i = WORKQUEUE_PRIORITY_FIRST; i <= WORKQUEUE_PRIORITY_LAST; ++i) {
  189. if (!TOR_TAILQ_EMPTY(&thread->in_pool->work[i]))
  190. return 1;
  191. }
  192. return thread->generation != thread->in_pool->generation;
  193. }
  194. /** Extract the next workqueue_entry_t from the the thread's pool, removing
  195. * it from the relevant queues and marking it as non-pending.
  196. *
  197. * The caller must hold the lock. */
  198. static workqueue_entry_t *
  199. worker_thread_extract_next_work(workerthread_t *thread)
  200. {
  201. threadpool_t *pool = thread->in_pool;
  202. work_tailq_t *queue = NULL, *this_queue;
  203. unsigned i;
  204. for (i = WORKQUEUE_PRIORITY_FIRST; i <= WORKQUEUE_PRIORITY_LAST; ++i) {
  205. this_queue = &pool->work[i];
  206. if (!TOR_TAILQ_EMPTY(this_queue)) {
  207. queue = this_queue;
  208. if (! tor_weak_random_one_in_n(&pool->weak_rng,
  209. thread->lower_priority_chance)) {
  210. /* Usually we'll just break now, so that we can get out of the loop
  211. * and use the queue where we found work. But with a small
  212. * probability, we'll keep looking for lower priority work, so that
  213. * we don't ignore our low-priority queues entirely. */
  214. break;
  215. }
  216. }
  217. }
  218. if (queue == NULL)
  219. return NULL;
  220. workqueue_entry_t *work = TOR_TAILQ_FIRST(queue);
  221. TOR_TAILQ_REMOVE(queue, work, next_work);
  222. work->pending = 0;
  223. return work;
  224. }
  225. /**
  226. * Main function for the worker thread.
  227. */
  228. static void
  229. worker_thread_main(void *thread_)
  230. {
  231. workerthread_t *thread = thread_;
  232. threadpool_t *pool = thread->in_pool;
  233. workqueue_entry_t *work;
  234. workqueue_reply_t result;
  235. tor_mutex_acquire(&pool->lock);
  236. while (1) {
  237. /* lock must be held at this point. */
  238. while (worker_thread_has_work(thread)) {
  239. /* lock must be held at this point. */
  240. if (thread->in_pool->generation != thread->generation) {
  241. void *arg = thread->in_pool->update_args[thread->index];
  242. thread->in_pool->update_args[thread->index] = NULL;
  243. workqueue_reply_t (*update_fn)(void*,void*) =
  244. thread->in_pool->update_fn;
  245. thread->generation = thread->in_pool->generation;
  246. tor_mutex_release(&pool->lock);
  247. workqueue_reply_t r = update_fn(thread->state, arg);
  248. if (r != WQ_RPL_REPLY) {
  249. return;
  250. }
  251. tor_mutex_acquire(&pool->lock);
  252. continue;
  253. }
  254. work = worker_thread_extract_next_work(thread);
  255. if (BUG(work == NULL))
  256. break;
  257. tor_mutex_release(&pool->lock);
  258. /* We run the work function without holding the thread lock. This
  259. * is the main thread's first opportunity to give us more work. */
  260. result = work->fn(thread->state, work->arg);
  261. /* Queue the reply for the main thread. */
  262. queue_reply(thread->reply_queue, work);
  263. /* We may need to exit the thread. */
  264. if (result != WQ_RPL_REPLY) {
  265. return;
  266. }
  267. tor_mutex_acquire(&pool->lock);
  268. }
  269. /* At this point the lock is held, and there is no work in this thread's
  270. * queue. */
  271. /* TODO: support an idle-function */
  272. /* Okay. Now, wait till somebody has work for us. */
  273. if (tor_cond_wait(&pool->condition, &pool->lock, NULL) < 0) {
  274. log_warn(LD_GENERAL, "Fail tor_cond_wait.");
  275. }
  276. }
  277. }
  278. /** Put a reply on the reply queue. The reply must not currently be on
  279. * any thread's work queue. */
  280. static void
  281. queue_reply(replyqueue_t *queue, workqueue_entry_t *work)
  282. {
  283. int was_empty;
  284. tor_mutex_acquire(&queue->lock);
  285. was_empty = TOR_TAILQ_EMPTY(&queue->answers);
  286. TOR_TAILQ_INSERT_TAIL(&queue->answers, work, next_work);
  287. tor_mutex_release(&queue->lock);
  288. if (was_empty) {
  289. if (queue->alert.alert_fn(queue->alert.write_fd) < 0) {
  290. /* XXXX complain! */
  291. }
  292. }
  293. }
  294. /** Allocate and start a new worker thread to use state object <b>state</b>,
  295. * and send responses to <b>replyqueue</b>. */
  296. static workerthread_t *
  297. workerthread_new(int32_t lower_priority_chance,
  298. void *state, threadpool_t *pool, replyqueue_t *replyqueue)
  299. {
  300. workerthread_t *thr = tor_malloc_zero(sizeof(workerthread_t));
  301. thr->state = state;
  302. thr->reply_queue = replyqueue;
  303. thr->in_pool = pool;
  304. thr->lower_priority_chance = lower_priority_chance;
  305. if (spawn_func(worker_thread_main, thr) < 0) {
  306. //LCOV_EXCL_START
  307. tor_assert_nonfatal_unreached();
  308. log_err(LD_GENERAL, "Can't launch worker thread.");
  309. tor_free(thr);
  310. return NULL;
  311. //LCOV_EXCL_STOP
  312. }
  313. return thr;
  314. }
  315. /**
  316. * Queue an item of work for a thread in a thread pool. The function
  317. * <b>fn</b> will be run in a worker thread, and will receive as arguments the
  318. * thread's state object, and the provided object <b>arg</b>. It must return
  319. * one of WQ_RPL_REPLY, WQ_RPL_ERROR, or WQ_RPL_SHUTDOWN.
  320. *
  321. * Regardless of its return value, the function <b>reply_fn</b> will later be
  322. * run in the main thread when it invokes replyqueue_process(), and will
  323. * receive as its argument the same <b>arg</b> object. It's the reply
  324. * function's responsibility to free the work object.
  325. *
  326. * On success, return a workqueue_entry_t object that can be passed to
  327. * workqueue_entry_cancel(). On failure, return NULL. (Failure is not
  328. * currently possible, but callers should check anyway.)
  329. *
  330. * Items are executed in a loose priority order -- each thread will usually
  331. * take from the queued work with the highest prioirity, but will occasionally
  332. * visit lower-priority queues to keep them from starving completely.
  333. *
  334. * Note that because of priorities and thread behavior, work items may not
  335. * be executed strictly in order.
  336. */
  337. workqueue_entry_t *
  338. threadpool_queue_work_priority(threadpool_t *pool,
  339. workqueue_priority_t prio,
  340. workqueue_reply_t (*fn)(void *, void *),
  341. void (*reply_fn)(void *),
  342. void *arg)
  343. {
  344. tor_assert(((int)prio) >= WORKQUEUE_PRIORITY_FIRST &&
  345. ((int)prio) <= WORKQUEUE_PRIORITY_LAST);
  346. workqueue_entry_t *ent = workqueue_entry_new(fn, reply_fn, arg);
  347. ent->on_pool = pool;
  348. ent->pending = 1;
  349. ent->priority = prio;
  350. tor_mutex_acquire(&pool->lock);
  351. TOR_TAILQ_INSERT_TAIL(&pool->work[prio], ent, next_work);
  352. tor_cond_signal_one(&pool->condition);
  353. tor_mutex_release(&pool->lock);
  354. return ent;
  355. }
  356. /** As threadpool_queue_work_priority(), but assumes WQ_PRI_HIGH */
  357. workqueue_entry_t *
  358. threadpool_queue_work(threadpool_t *pool,
  359. workqueue_reply_t (*fn)(void *, void *),
  360. void (*reply_fn)(void *),
  361. void *arg)
  362. {
  363. return threadpool_queue_work_priority(pool, WQ_PRI_HIGH, fn, reply_fn, arg);
  364. }
  365. /**
  366. * Queue a copy of a work item for every thread in a pool. This can be used,
  367. * for example, to tell the threads to update some parameter in their states.
  368. *
  369. * Arguments are as for <b>threadpool_queue_work</b>, except that the
  370. * <b>arg</b> value is passed to <b>dup_fn</b> once per each thread to
  371. * make a copy of it.
  372. *
  373. * UPDATE FUNCTIONS MUST BE IDEMPOTENT. We do not guarantee that every update
  374. * will be run. If a new update is scheduled before the old update finishes
  375. * running, then the new will replace the old in any threads that haven't run
  376. * it yet.
  377. *
  378. * Return 0 on success, -1 on failure.
  379. */
  380. int
  381. threadpool_queue_update(threadpool_t *pool,
  382. void *(*dup_fn)(void *),
  383. workqueue_reply_t (*fn)(void *, void *),
  384. void (*free_fn)(void *),
  385. void *arg)
  386. {
  387. int i, n_threads;
  388. void (*old_args_free_fn)(void *arg);
  389. void **old_args;
  390. void **new_args;
  391. tor_mutex_acquire(&pool->lock);
  392. n_threads = pool->n_threads;
  393. old_args = pool->update_args;
  394. old_args_free_fn = pool->free_update_arg_fn;
  395. new_args = tor_calloc(n_threads, sizeof(void*));
  396. for (i = 0; i < n_threads; ++i) {
  397. if (dup_fn)
  398. new_args[i] = dup_fn(arg);
  399. else
  400. new_args[i] = arg;
  401. }
  402. pool->update_args = new_args;
  403. pool->free_update_arg_fn = free_fn;
  404. pool->update_fn = fn;
  405. ++pool->generation;
  406. tor_cond_signal_all(&pool->condition);
  407. tor_mutex_release(&pool->lock);
  408. if (old_args) {
  409. for (i = 0; i < n_threads; ++i) {
  410. if (old_args[i] && old_args_free_fn)
  411. old_args_free_fn(old_args[i]);
  412. }
  413. tor_free(old_args);
  414. }
  415. return 0;
  416. }
  417. /** Don't have more than this many threads per pool. */
  418. #define MAX_THREADS 1024
  419. /** For half of our threads, choose lower priority queues with probability
  420. * 1/N for each of these values. Both are chosen somewhat arbitrarily. If
  421. * CHANCE_PERMISSIVE is too low, then we have a risk of low-priority tasks
  422. * stalling forever. If it's too high, we have a risk of low-priority tasks
  423. * grabbing half of the threads. */
  424. #define CHANCE_PERMISSIVE 37
  425. #define CHANCE_STRICT INT32_MAX
  426. /** Launch threads until we have <b>n</b>. */
  427. static int
  428. threadpool_start_threads(threadpool_t *pool, int n)
  429. {
  430. if (BUG(n < 0))
  431. return -1; // LCOV_EXCL_LINE
  432. if (n > MAX_THREADS)
  433. n = MAX_THREADS;
  434. tor_mutex_acquire(&pool->lock);
  435. if (pool->n_threads < n)
  436. pool->threads = tor_reallocarray(pool->threads,
  437. sizeof(workerthread_t*), n);
  438. while (pool->n_threads < n) {
  439. /* For half of our threads, we'll choose lower priorities permissively;
  440. * for the other half, we'll stick more strictly to higher priorities.
  441. * This keeps slow low-priority tasks from taking over completely. */
  442. int32_t chance = (pool->n_threads & 1) ? CHANCE_STRICT : CHANCE_PERMISSIVE;
  443. void *state = pool->new_thread_state_fn(pool->new_thread_state_arg);
  444. workerthread_t *thr = workerthread_new(chance,
  445. state, pool, pool->reply_queue);
  446. if (!thr) {
  447. //LCOV_EXCL_START
  448. tor_assert_nonfatal_unreached();
  449. pool->free_thread_state_fn(state);
  450. tor_mutex_release(&pool->lock);
  451. return -1;
  452. //LCOV_EXCL_STOP
  453. }
  454. thr->index = pool->n_threads;
  455. pool->threads[pool->n_threads++] = thr;
  456. }
  457. tor_mutex_release(&pool->lock);
  458. return 0;
  459. }
  460. /**
  461. * Construct a new thread pool with <b>n</b> worker threads, configured to
  462. * send their output to <b>replyqueue</b>. The threads' states will be
  463. * constructed with the <b>new_thread_state_fn</b> call, receiving <b>arg</b>
  464. * as its argument. When the threads close, they will call
  465. * <b>free_thread_state_fn</b> on their states.
  466. */
  467. threadpool_t *
  468. threadpool_new(int n_threads,
  469. replyqueue_t *replyqueue,
  470. void *(*new_thread_state_fn)(void*),
  471. void (*free_thread_state_fn)(void*),
  472. void *arg)
  473. {
  474. threadpool_t *pool;
  475. pool = tor_malloc_zero(sizeof(threadpool_t));
  476. tor_mutex_init_nonrecursive(&pool->lock);
  477. tor_cond_init(&pool->condition);
  478. unsigned i;
  479. for (i = WORKQUEUE_PRIORITY_FIRST; i <= WORKQUEUE_PRIORITY_LAST; ++i) {
  480. TOR_TAILQ_INIT(&pool->work[i]);
  481. }
  482. {
  483. unsigned seed;
  484. crypto_rand((void*)&seed, sizeof(seed));
  485. tor_init_weak_random(&pool->weak_rng, seed);
  486. }
  487. pool->new_thread_state_fn = new_thread_state_fn;
  488. pool->new_thread_state_arg = arg;
  489. pool->free_thread_state_fn = free_thread_state_fn;
  490. pool->reply_queue = replyqueue;
  491. if (threadpool_start_threads(pool, n_threads) < 0) {
  492. //LCOV_EXCL_START
  493. tor_assert_nonfatal_unreached();
  494. tor_cond_uninit(&pool->condition);
  495. tor_mutex_uninit(&pool->lock);
  496. tor_free(pool);
  497. return NULL;
  498. //LCOV_EXCL_STOP
  499. }
  500. return pool;
  501. }
  502. /** Return the reply queue associated with a given thread pool. */
  503. replyqueue_t *
  504. threadpool_get_replyqueue(threadpool_t *tp)
  505. {
  506. return tp->reply_queue;
  507. }
  508. /** Allocate a new reply queue. Reply queues are used to pass results from
  509. * worker threads to the main thread. Since the main thread is running an
  510. * IO-centric event loop, it needs to get woken up with means other than a
  511. * condition variable. */
  512. replyqueue_t *
  513. replyqueue_new(uint32_t alertsocks_flags)
  514. {
  515. replyqueue_t *rq;
  516. rq = tor_malloc_zero(sizeof(replyqueue_t));
  517. if (alert_sockets_create(&rq->alert, alertsocks_flags) < 0) {
  518. //LCOV_EXCL_START
  519. tor_free(rq);
  520. return NULL;
  521. //LCOV_EXCL_STOP
  522. }
  523. tor_mutex_init(&rq->lock);
  524. TOR_TAILQ_INIT(&rq->answers);
  525. return rq;
  526. }
  527. /** Internal: Run from the libevent mainloop when there is work to handle in
  528. * the reply queue handler. */
  529. static void
  530. reply_event_cb(evutil_socket_t sock, short events, void *arg)
  531. {
  532. threadpool_t *tp = arg;
  533. (void) sock;
  534. (void) events;
  535. replyqueue_process(tp->reply_queue);
  536. if (tp->reply_cb)
  537. tp->reply_cb(tp);
  538. }
  539. /** Register the threadpool <b>tp</b>'s reply queue with the libevent
  540. * mainloop of <b>base</b>. If <b>tp</b> is provided, it is run after
  541. * each time there is work to process from the reply queue. Return 0 on
  542. * success, -1 on failure.
  543. */
  544. int
  545. threadpool_register_reply_event(threadpool_t *tp,
  546. void (*cb)(threadpool_t *tp))
  547. {
  548. struct event_base *base = tor_libevent_get_base();
  549. if (tp->reply_event) {
  550. tor_event_free(tp->reply_event);
  551. }
  552. tp->reply_event = tor_event_new(base,
  553. tp->reply_queue->alert.read_fd,
  554. EV_READ|EV_PERSIST,
  555. reply_event_cb,
  556. tp);
  557. tor_assert(tp->reply_event);
  558. tp->reply_cb = cb;
  559. return event_add(tp->reply_event, NULL);
  560. }
  561. /**
  562. * Process all pending replies on a reply queue. The main thread should call
  563. * this function every time the socket returned by replyqueue_get_socket() is
  564. * readable.
  565. */
  566. void
  567. replyqueue_process(replyqueue_t *queue)
  568. {
  569. int r = queue->alert.drain_fn(queue->alert.read_fd);
  570. if (r < 0) {
  571. //LCOV_EXCL_START
  572. static ratelim_t warn_limit = RATELIM_INIT(7200);
  573. log_fn_ratelim(&warn_limit, LOG_WARN, LD_GENERAL,
  574. "Failure from drain_fd: %s",
  575. tor_socket_strerror(-r));
  576. //LCOV_EXCL_STOP
  577. }
  578. tor_mutex_acquire(&queue->lock);
  579. while (!TOR_TAILQ_EMPTY(&queue->answers)) {
  580. /* lock must be held at this point.*/
  581. workqueue_entry_t *work = TOR_TAILQ_FIRST(&queue->answers);
  582. TOR_TAILQ_REMOVE(&queue->answers, work, next_work);
  583. tor_mutex_release(&queue->lock);
  584. work->on_pool = NULL;
  585. work->reply_fn(work->arg);
  586. workqueue_entry_free(work);
  587. tor_mutex_acquire(&queue->lock);
  588. }
  589. tor_mutex_release(&queue->lock);
  590. }