/** Number of elements in threads. */
int n_threads;
- /** Number of elements to be created in threads. */
- int n_threads_max;
/** Mutex to protect all the above fields. */
tor_mutex_t lock;
void *(*new_thread_state_fn)(void*);
void (*free_thread_state_fn)(void*);
void *new_thread_state_arg;
-
- /** Used for signalling the worker threads to exit. */
- int exit;
- /** Mutex for controlling worker threads' startup and exit. */
- tor_mutex_t control_lock;
};
/** Used to put a workqueue_priority_t value into a bitfield. */
static void
worker_thread_main(void *thread_)
{
- static int n_worker_threads_running = 0;
workerthread_t *thread = thread_;
threadpool_t *pool = thread->in_pool;
workqueue_entry_t *work;
workqueue_reply_t result;
tor_mutex_acquire(&pool->lock);
- log_debug(LD_GENERAL, "Worker thread (TID: %lu) has started.",
- tor_get_thread_id());
-
- if (++n_worker_threads_running == pool->n_threads_max)
- /* Let the main thread know, the last worker thread has started. */
- tor_cond_signal_one(&pool->condition);
-
while (1) {
- /* Exit thread when signaled to exit */
- if (pool->exit)
- break;
-
/* lock must be held at this point. */
while (worker_thread_has_work(thread)) {
/* lock must be held at this point. */
workqueue_reply_t r = update_fn(thread->state, arg);
if (r != WQ_RPL_REPLY) {
- break;
+ return;
}
tor_mutex_acquire(&pool->lock);
/* We may need to exit the thread. */
if (result != WQ_RPL_REPLY) {
- break;
+ return;
}
tor_mutex_acquire(&pool->lock);
}
log_warn(LD_GENERAL, "Fail tor_cond_wait.");
}
}
-
- log_debug(LD_GENERAL, "Worker thread (TID: %lu) has exited.",
- tor_get_thread_id());
-
- if (--n_worker_threads_running == 0)
- tor_mutex_release(&pool->control_lock);
-
- tor_mutex_release(&pool->lock);
}
/** Put a reply on the reply queue. The reply must not currently be on
pool->threads = tor_reallocarray(pool->threads,
sizeof(workerthread_t*), n);
- pool->n_threads_max = n;
- log_debug(LD_GENERAL, "Starting worker threads...");
-
while (pool->n_threads < n) {
/* For half of our threads, we'll choose lower priorities permissively;
* for the other half, we'll stick more strictly to higher priorities.
thr->index = pool->n_threads;
pool->threads[pool->n_threads++] = thr;
}
-
tor_mutex_release(&pool->lock);
- tor_mutex_acquire(&pool->control_lock);
-
- struct timeval tv = {.tv_sec = 30, .tv_usec = 0};
-
- /* Wait for the last launched thread to confirm us, it has started.
- * Wait max 30 seconds */
- switch (tor_cond_wait(&pool->condition, &pool->control_lock, &tv)) {
- case -1:
- log_warn(LD_GENERAL, "Failed to confirm worker threads' start up.");
- goto error;
- case 1:
- log_warn(LD_GENERAL, "Failed to confirm worker threads' "
- "start up after timeout.");
- goto error;
- case 0:
- log_debug(LD_GENERAL, "Starting worker threads finished.");
- break;
- default:;
- }
-
- /* On success control lock stays locked. This is required for the
- * main thread to wait for the worker threads to exit on shutdown. */
return 0;
-
-error:
- tor_mutex_release(&pool->control_lock);
- return -1;
}
/**
pool = tor_malloc_zero(sizeof(threadpool_t));
tor_mutex_init_nonrecursive(&pool->lock);
tor_cond_init(&pool->condition);
- tor_mutex_init_nonrecursive(&pool->control_lock);
- pool->exit = 0;
-
unsigned i;
for (i = WORKQUEUE_PRIORITY_FIRST; i <= WORKQUEUE_PRIORITY_LAST; ++i) {
TOR_TAILQ_INIT(&pool->work[i]);
if (threadpool_start_threads(pool, n_threads) < 0) {
//LCOV_EXCL_START
tor_assert_nonfatal_unreached();
+ tor_cond_uninit(&pool->condition);
+ tor_mutex_uninit(&pool->lock);
threadpool_free(pool);
return NULL;
//LCOV_EXCL_STOP
if (!pool)
return;
- tor_mutex_acquire(&pool->lock);
- /* Signal the worker threads to exit */
- pool->exit = 1;
- /* If worker threads are waiting for work, let them continue to exit */
- tor_cond_signal_all(&pool->condition);
-
- log_debug(LD_GENERAL, "Signaled worker threads to exit. "
- "Waiting for them to exit...");
- tor_mutex_release(&pool->lock);
-
- /* Wait until all worker threads have exited.
- * pool->control_lock must be prelocked here. */
- tor_mutex_acquire(&pool->control_lock);
- /* Unlock required, else main thread hangs on
- * tor_mutex_uninit(&pool->control_lock) */
- tor_mutex_release(&pool->control_lock);
-
- /* If this message appears in the log before all threads have confirmed
- * their exit, then pool->control_lock wasn't prelocked for some reason. */
- log_debug(LD_GENERAL, "All worker threads have exited. "
- "Beginning to clean up...");
-
- tor_cond_uninit(&pool->condition);
- tor_mutex_uninit(&pool->lock);
- tor_mutex_uninit(&pool->control_lock);
-
if (pool->threads) {
for (int i = 0; i != pool->n_threads; ++i)
workerthread_free(pool->threads[i]);
tor_free(pool->threads);
}
- if (pool->update_args) {
- if (!pool->free_update_arg_fn)
- log_warn(LD_GENERAL, "Freeing pool->update_args not possible. "
- "pool->free_update_arg_fn is not set.");
- else
- pool->free_update_arg_fn(pool->update_args);
- }
+ if (pool->update_args)
+ pool->free_update_arg_fn(pool->update_args);
if (pool->reply_event) {
- if (tor_event_del(pool->reply_event) == -1)
- log_warn(LD_GENERAL, "libevent error: deleting reply event failed.");
- else
- tor_event_free(pool->reply_event);
+ tor_event_del(pool->reply_event);
+ tor_event_free(pool->reply_event);
}
if (pool->reply_queue)
replyqueue_free(pool->reply_queue);
- if (pool->new_thread_state_arg) {
- if (!pool->free_thread_state_fn)
- log_warn(LD_GENERAL, "Freeing pool->new_thread_state_arg not possible. "
- "pool->free_thread_state_fn is not set.");
- else
- pool->free_thread_state_fn(pool->new_thread_state_arg);
- }
+ if (pool->new_thread_state_arg)
+ pool->free_thread_state_fn(pool->new_thread_state_arg);
tor_free(pool);
-
- log_debug(LD_GENERAL, "Cleanup finished.");
}
/** Return the reply queue associated with a given thread pool. */
projects / thirdparty / tor.git / commitdiff
