January 2015 - tor-commits - lists.torproject.org

[tor/master] Make pending work cancellable.
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit c7eebe237ddf0555a99b2ef10fd95def2a4bbbd4 Author: Nick Mathewson <nickm(a)torproject.org> Date: Tue Sep 24 16:57:40 2013 -0400 Make pending work cancellable. --- src/common/workqueue.c | 72 +++++++++++++++++++++++++++----------------- src/common/workqueue.h | 18 +++++------ src/test/bench_workqueue.c | 24 +++++---------- 3 files changed, 61 insertions(+), 53 deletions(-) diff --git a/src/common/workqueue.c b/src/common/workqueue.c index ea8dcb0..80e061d 100644 --- a/src/common/workqueue.c +++ b/src/common/workqueue.c @@ -31,16 +31,18 @@ keep array of threads; round-robin between them. */ -typedef struct workqueue_entry_s { - TOR_SIMPLEQ_ENTRY(workqueue_entry_s) next_work; - int (*fn)(int status, void *state, void *arg); +struct workqueue_entry_s { + TOR_TAILQ_ENTRY(workqueue_entry_s) next_work; + struct workerthread_s *on_thread; + uint8_t pending; + int (*fn)(void *state, void *arg); void (*reply_fn)(void *arg); void *arg; -} workqueue_entry_t; +}; struct replyqueue_s { tor_mutex_t lock; - TOR_SIMPLEQ_HEAD(, workqueue_entry_s) answers; + TOR_TAILQ_HEAD(, workqueue_entry_s) answers; void (*alert_fn)(struct replyqueue_s *); // lock not held on this, next 2. tor_socket_t write_sock; @@ -50,7 +52,7 @@ struct replyqueue_s { typedef struct workerthread_s { tor_mutex_t lock; tor_cond_t condition; - TOR_SIMPLEQ_HEAD(, workqueue_entry_s) work; + TOR_TAILQ_HEAD(, workqueue_entry_s) work; unsigned is_running; unsigned is_shut_down; unsigned waiting; @@ -76,7 +78,7 @@ struct threadpool_s { static void queue_reply(replyqueue_t *queue, workqueue_entry_t *work); static workqueue_entry_t * -workqueue_entry_new(int (*fn)(int, void*, void*), +workqueue_entry_new(int (*fn)(void*, void*), void (*reply_fn)(void*), void *arg) { @@ -95,6 +97,23 @@ workqueue_entry_free(workqueue_entry_t *ent) tor_free(ent); } +int +workqueue_entry_cancel(workqueue_entry_t *ent) +{ + int cancelled = 0; + tor_mutex_acquire(&ent->on_thread->lock); + if (ent->pending) { + TOR_TAILQ_REMOVE(&ent->on_thread->work, ent, next_work); + cancelled = 1; + } + tor_mutex_release(&ent->on_thread->lock); + + if (cancelled) { + tor_free(ent); + } + return cancelled; +} + static void worker_thread_main(void *thread_) { @@ -107,20 +126,17 @@ worker_thread_main(void *thread_) thread->is_running = 1; while (1) { /* lock held. */ - while (!TOR_SIMPLEQ_EMPTY(&thread->work)) { + while (!TOR_TAILQ_EMPTY(&thread->work)) { /* lock held. */ - work = TOR_SIMPLEQ_FIRST(&thread->work); - TOR_SIMPLEQ_REMOVE_HEAD(&thread->work, next_work); + work = TOR_TAILQ_FIRST(&thread->work); + TOR_TAILQ_REMOVE(&thread->work, work, next_work); + work->pending = 0; tor_mutex_release(&thread->lock); - result = work->fn(WQ_CMD_RUN, thread->state, work->arg); + result = work->fn(thread->state, work->arg); - if (result == WQ_RPL_QUEUE) { - queue_reply(thread->reply_queue, work); - } else { - workqueue_entry_free(work); - } + queue_reply(thread->reply_queue, work); tor_mutex_acquire(&thread->lock); if (result >= WQ_RPL_ERROR) { @@ -148,8 +164,8 @@ queue_reply(replyqueue_t *queue, workqueue_entry_t *work) { int was_empty; tor_mutex_acquire(&queue->lock); - was_empty = TOR_SIMPLEQ_EMPTY(&queue->answers); - TOR_SIMPLEQ_INSERT_TAIL(&queue->answers, work, next_work); + was_empty = TOR_TAILQ_EMPTY(&queue->answers); + TOR_TAILQ_INSERT_TAIL(&queue->answers, work, next_work); tor_mutex_release(&queue->lock); if (was_empty) { @@ -175,7 +191,7 @@ workerthread_new(void *state, replyqueue_t *replyqueue) workerthread_t *thr = tor_malloc_zero(sizeof(workerthread_t)); tor_mutex_init_for_cond(&thr->lock); tor_cond_init(&thr->condition); - TOR_SIMPLEQ_INIT(&thr->work); + TOR_TAILQ_INIT(&thr->work); thr->state = state; thr->reply_queue = replyqueue; @@ -187,9 +203,9 @@ workerthread_new(void *state, replyqueue_t *replyqueue) return thr; } -void * +workqueue_entry_t * threadpool_queue_work(threadpool_t *pool, - int (*fn)(int, void *, void *), + int (*fn)(void *, void *), void (*reply_fn)(void *), void *arg) { @@ -206,11 +222,12 @@ threadpool_queue_work(threadpool_t *pool, pool->next_for_work = 0; tor_mutex_release(&pool->lock); - ent = workqueue_entry_new(fn, reply_fn, arg); tor_mutex_acquire(&worker->lock); - TOR_SIMPLEQ_INSERT_TAIL(&worker->work, ent, next_work); + ent->on_thread = worker; + ent->pending = 1; + TOR_TAILQ_INSERT_TAIL(&worker->work, ent, next_work); if (worker->waiting) /* XXXX inside or outside of lock?? */ tor_cond_signal_one(&worker->condition); @@ -298,7 +315,7 @@ replyqueue_new(void) rq = tor_malloc_zero(sizeof(replyqueue_t)); tor_mutex_init(&rq->lock); - TOR_SIMPLEQ_INIT(&rq->answers); + TOR_TAILQ_INIT(&rq->answers); rq->read_sock = pair[0]; rq->write_sock = pair[1]; @@ -331,10 +348,10 @@ replyqueue_process(replyqueue_t *queue) /* XXXX freak out on r == 0, or r == "error, not retryable". */ tor_mutex_acquire(&queue->lock); - while (!TOR_SIMPLEQ_EMPTY(&queue->answers)) { + while (!TOR_TAILQ_EMPTY(&queue->answers)) { /* lock held. */ - workqueue_entry_t *work = TOR_SIMPLEQ_FIRST(&queue->answers); - TOR_SIMPLEQ_REMOVE_HEAD(&queue->answers, next_work); + workqueue_entry_t *work = TOR_TAILQ_FIRST(&queue->answers); + TOR_TAILQ_REMOVE(&queue->answers, work, next_work); tor_mutex_release(&queue->lock); work->reply_fn(work->arg); @@ -345,3 +362,4 @@ replyqueue_process(replyqueue_t *queue) tor_mutex_release(&queue->lock); } + diff --git a/src/common/workqueue.h b/src/common/workqueue.h index e502734..47753cf 100644 --- a/src/common/workqueue.h +++ b/src/common/workqueue.h @@ -8,20 +8,20 @@ typedef struct replyqueue_s replyqueue_t; typedef struct threadpool_s threadpool_t; - +typedef struct workqueue_entry_s workqueue_entry_t; #define WQ_CMD_RUN 0 #define WQ_CMD_CANCEL 1 -#define WQ_RPL_QUEUE 0 -#define WQ_RPL_NOQUEUE 1 -#define WQ_RPL_ERROR 2 -#define WQ_RPL_SHUTDOWN 3 +#define WQ_RPL_REPLY 0 +#define WQ_RPL_ERROR 1 +#define WQ_RPL_SHUTDOWN 2 -void *threadpool_queue_work(threadpool_t *pool, - int (*fn)(int, void *, void *), - void (*reply_fn)(void *), - void *arg); +workqueue_entry_t *threadpool_queue_work(threadpool_t *pool, + int (*fn)(void *, void *), + void (*reply_fn)(void *), + void *arg); +int workqueue_entry_cancel(workqueue_entry_t *pending_work); int threadpool_start_threads(threadpool_t *pool, int n); threadpool_t *threadpool_new(int n_threads, replyqueue_t *replyqueue, diff --git a/src/test/bench_workqueue.c b/src/test/bench_workqueue.c index 1bdfbef..f190c61 100644 --- a/src/test/bench_workqueue.c +++ b/src/test/bench_workqueue.c @@ -64,7 +64,7 @@ mark_handled(int serial) } static int -workqueue_do_rsa(int cmd, void *state, void *work) +workqueue_do_rsa(void *state, void *work) { rsa_work_t *rw = work; state_t *st = state; @@ -74,16 +74,11 @@ workqueue_do_rsa(int cmd, void *state, void *work) tor_assert(st->magic == 13371337); - if (cmd == WQ_CMD_CANCEL) { - tor_free(work); - return WQ_RPL_NOQUEUE; - } - len = crypto_pk_private_sign(rsa, (char*)sig, 256, (char*)rw->msg, rw->msglen); if (len < 0) { - tor_free(work); - return WQ_RPL_NOQUEUE; + rw->msglen = 0; + return WQ_RPL_ERROR; } memset(rw->msg, 0, sizeof(rw->msg)); @@ -93,12 +88,12 @@ workqueue_do_rsa(int cmd, void *state, void *work) mark_handled(rw->serial); - return WQ_RPL_QUEUE; + return WQ_RPL_REPLY; } #if 0 static int -workqueue_do_shutdown(int cmd, void *state, void *work) +workqueue_do_shutdown(void *state, void *work) { (void)state; (void)work; @@ -110,7 +105,7 @@ workqueue_do_shutdown(int cmd, void *state, void *work) #endif static int -workqueue_do_ecdh(int cmd, void *state, void *work) +workqueue_do_ecdh(void *state, void *work) { ecdh_work_t *ew = work; uint8_t output[CURVE25519_OUTPUT_LEN]; @@ -118,16 +113,11 @@ workqueue_do_ecdh(int cmd, void *state, void *work) tor_assert(st->magic == 13371337); - if (cmd == WQ_CMD_CANCEL) { - tor_free(work); - return WQ_RPL_NOQUEUE; - } - curve25519_handshake(output, &st->ecdh, &ew->u.pk); memcpy(ew->u.msg, output, CURVE25519_OUTPUT_LEN); ++st->n_handled; mark_handled(ew->serial); - return WQ_RPL_QUEUE; + return WQ_RPL_REPLY; } static void *

1 0

[tor/master] Fix: change copyright year in workqueue and thread tests
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit f52ac5be74b3cb6f657a6d7a1fa7db2c9595728d Author: David Goulet <dgoulet(a)ev0ke.net> Date: Wed Jan 21 13:58:18 2015 -0500 Fix: change copyright year in workqueue and thread tests Signed-off-by: David Goulet <dgoulet(a)ev0ke.net> --- src/test/test_threads.c | 2 +- src/test/test_workqueue.c | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/test_threads.c b/src/test/test_threads.c index c029304..2ac08d4 100644 --- a/src/test/test_threads.c +++ b/src/test/test_threads.c @@ -1,6 +1,6 @@ /* Copyright (c) 2001-2004, Roger Dingledine. * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson. - * Copyright (c) 2007-2013, The Tor Project, Inc. */ + * Copyright (c) 2007-2015, The Tor Project, Inc. */ /* See LICENSE for licensing information */ #include "orconfig.h" diff --git a/src/test/test_workqueue.c b/src/test/test_workqueue.c index 8ce4405..aaff506 100644 --- a/src/test/test_workqueue.c +++ b/src/test/test_workqueue.c @@ -1,6 +1,6 @@ /* Copyright (c) 2001-2004, Roger Dingledine. * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson. - * Copyright (c) 2007-2013, The Tor Project, Inc. */ + * Copyright (c) 2007-2015, The Tor Project, Inc. */ /* See LICENSE for licensing information */ #include "or.h"

1 0

[tor/master] "Recursive" locks, not "reentrant" locks. Duh.
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit e47a90a976f883571bea6e58620aa13f058873e3 Author: Nick Mathewson <nickm(a)torproject.org> Date: Fri Sep 27 12:32:19 2013 -0400 "Recursive" locks, not "reentrant" locks. Duh. --- src/common/compat_pthreads.c | 16 ++++++++-------- src/common/compat_threads.c | 6 +++--- src/common/compat_threads.h | 8 ++++---- src/common/compat_winthreads.c | 2 +- 4 files changed, 16 insertions(+), 16 deletions(-) diff --git a/src/common/compat_pthreads.c b/src/common/compat_pthreads.c index 5983427..69f7bac 100644 --- a/src/common/compat_pthreads.c +++ b/src/common/compat_pthreads.c @@ -45,7 +45,6 @@ static pthread_attr_t attr_detached; /** True iff we've called tor_threads_init() */ static int threads_initialized = 0; - /** Minimalist interface to run a void function in the background. On * Unix calls fork, on win32 calls beginthread. Returns -1 on failure. * func should not return, but rather should call spawn_exit. @@ -79,9 +78,10 @@ spawn_exit(void) } /** A mutex attribute that we're going to use to tell pthreads that we want - * "reentrant" mutexes (i.e., once we can re-lock if we're already holding + * "recursive" mutexes (i.e., once we can re-lock if we're already holding * them.) */ -static pthread_mutexattr_t attr_reentrant; +static pthread_mutexattr_t attr_recursive; + /** Initialize <b>mutex</b> so it can be locked. Every mutex must be set * up with tor_mutex_init() or tor_mutex_new(); not both. */ void @@ -90,7 +90,7 @@ tor_mutex_init(tor_mutex_t *mutex) int err; if (PREDICT_UNLIKELY(!threads_initialized)) tor_threads_init(); - err = pthread_mutex_init(&mutex->mutex, &attr_reentrant); + err = pthread_mutex_init(&mutex->mutex, &attr_recursive); if (PREDICT_UNLIKELY(err)) { log_err(LD_GENERAL, "Error %d creating a mutex.", err); tor_fragile_assert(); @@ -98,9 +98,9 @@ tor_mutex_init(tor_mutex_t *mutex) } /** As tor_mutex_init, but initialize a mutex suitable that may be - * non-reentrant, if the OS supports that. */ + * non-recursive, if the OS supports that. */ void -tor_mutex_init_nonreentrant(tor_mutex_t *mutex) +tor_mutex_init_nonrecursive(tor_mutex_t *mutex) { int err; if (PREDICT_UNLIKELY(!threads_initialized)) @@ -232,8 +232,8 @@ void tor_threads_init(void) { if (!threads_initialized) { - pthread_mutexattr_init(&attr_reentrant); - pthread_mutexattr_settype(&attr_reentrant, PTHREAD_MUTEX_RECURSIVE); + pthread_mutexattr_init(&attr_recursive); + pthread_mutexattr_settype(&attr_recursive, PTHREAD_MUTEX_RECURSIVE); tor_assert(0==pthread_attr_init(&attr_detached)); tor_assert(0==pthread_attr_setdetachstate(&attr_detached, 1)); threads_initialized = 1; diff --git a/src/common/compat_threads.c b/src/common/compat_threads.c index ba48eb4..f018475 100644 --- a/src/common/compat_threads.c +++ b/src/common/compat_threads.c @@ -33,12 +33,12 @@ tor_mutex_new(void) return m; } /** Return a newly allocated, ready-for-use mutex. This one might be - * non-reentrant, if that's faster. */ + * non-recursive, if that's faster. */ tor_mutex_t * -tor_mutex_new_nonreentrant(void) +tor_mutex_new_nonrecursive(void) { tor_mutex_t *m = tor_malloc_zero(sizeof(tor_mutex_t)); - tor_mutex_init_nonreentrant(m); + tor_mutex_init_nonrecursive(m); return m; } /** Release all storage and system resources held by <b>m</b>. */ diff --git a/src/common/compat_threads.h b/src/common/compat_threads.h index 1b59391..a5db72d 100644 --- a/src/common/compat_threads.h +++ b/src/common/compat_threads.h @@ -46,9 +46,9 @@ typedef struct tor_mutex_t { tor_mutex_t *tor_mutex_new(void); -tor_mutex_t *tor_mutex_new_nonreentrant(void); +tor_mutex_t *tor_mutex_new_nonrecursive(void); void tor_mutex_init(tor_mutex_t *m); -void tor_mutex_init_nonreentrant(tor_mutex_t *m); +void tor_mutex_init_nonrecursive(tor_mutex_t *m); void tor_mutex_acquire(tor_mutex_t *m); void tor_mutex_release(tor_mutex_t *m); void tor_mutex_free(tor_mutex_t *m); @@ -56,8 +56,8 @@ void tor_mutex_uninit(tor_mutex_t *m); unsigned long tor_get_thread_id(void); void tor_threads_init(void); -/** Conditions need nonreentrant mutexes with pthreads. */ -#define tor_mutex_init_for_cond(m) tor_mutex_init_nonreentrant(m) +/** Conditions need nonrecursive mutexes with pthreads. */ +#define tor_mutex_init_for_cond(m) tor_mutex_init_nonrecursive(m) void set_main_thread(void); int in_main_thread(void); diff --git a/src/common/compat_winthreads.c b/src/common/compat_winthreads.c index e19b1ca..4820eb3 100644 --- a/src/common/compat_winthreads.c +++ b/src/common/compat_winthreads.c @@ -53,7 +53,7 @@ tor_mutex_init(tor_mutex_t *m) InitializeCriticalSection(&m->mutex); } void -tor_mutex_init_nonreentrant(tor_mutex_t *m) +tor_mutex_init_nonrecursive(tor_mutex_t *m) { InitializeCriticalSection(&m->mutex); }

1 0

[tor/master] Remove if (1) indentation in cpuworker.c
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit 6c9c54e7fa8841e3c4d4d24f5933d433171d1112 Author: Nick Mathewson <nickm(a)torproject.org> Date: Wed Oct 2 12:34:08 2013 -0400 Remove if (1) indentation in cpuworker.c To avoid having diffs turn out too big, I had replaced some unneeded ifs and fors with if (1), so that the indentation would still work out right. Now I might as well clean those up. --- src/or/cpuworker.c | 310 ++++++++++++++++++++++++++-------------------------- 1 file changed, 153 insertions(+), 157 deletions(-) diff --git a/src/or/cpuworker.c b/src/or/cpuworker.c index f3f275d..abf1f22 100644 --- a/src/or/cpuworker.c +++ b/src/or/cpuworker.c @@ -310,79 +310,78 @@ cpuworker_onion_handshake_replyfn(void *work_) --total_pending_tasks; - if (1) { - /* Could avoid this, but doesn't matter. */ - memcpy(&rpl, &job->u.reply, sizeof(rpl)); - - tor_assert(rpl.magic == CPUWORKER_REPLY_MAGIC); - - if (rpl.timed && rpl.success && - rpl.handshake_type <= MAX_ONION_HANDSHAKE_TYPE) { - /* Time how long this request took. The handshake_type check should be - needless, but let's leave it in to be safe. */ - struct timeval tv_end, tv_diff; - int64_t usec_roundtrip; - tor_gettimeofday(&tv_end); - timersub(&tv_end, &rpl.started_at, &tv_diff); - usec_roundtrip = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec; - if (usec_roundtrip >= 0 && - usec_roundtrip < MAX_BELIEVABLE_ONIONSKIN_DELAY) { - ++onionskins_n_processed[rpl.handshake_type]; - onionskins_usec_internal[rpl.handshake_type] += rpl.n_usec; - onionskins_usec_roundtrip[rpl.handshake_type] += usec_roundtrip; - if (onionskins_n_processed[rpl.handshake_type] >= 500000) { - /* Scale down every 500000 handshakes. On a busy server, that's - * less impressive than it sounds. */ - onionskins_n_processed[rpl.handshake_type] /= 2; - onionskins_usec_internal[rpl.handshake_type] /= 2; - onionskins_usec_roundtrip[rpl.handshake_type] /= 2; - } + /* Could avoid this, but doesn't matter. */ + memcpy(&rpl, &job->u.reply, sizeof(rpl)); + + tor_assert(rpl.magic == CPUWORKER_REPLY_MAGIC); + + if (rpl.timed && rpl.success && + rpl.handshake_type <= MAX_ONION_HANDSHAKE_TYPE) { + /* Time how long this request took. The handshake_type check should be + needless, but let's leave it in to be safe. */ + struct timeval tv_end, tv_diff; + int64_t usec_roundtrip; + tor_gettimeofday(&tv_end); + timersub(&tv_end, &rpl.started_at, &tv_diff); + usec_roundtrip = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec; + if (usec_roundtrip >= 0 && + usec_roundtrip < MAX_BELIEVABLE_ONIONSKIN_DELAY) { + ++onionskins_n_processed[rpl.handshake_type]; + onionskins_usec_internal[rpl.handshake_type] += rpl.n_usec; + onionskins_usec_roundtrip[rpl.handshake_type] += usec_roundtrip; + if (onionskins_n_processed[rpl.handshake_type] >= 500000) { + /* Scale down every 500000 handshakes. On a busy server, that's + * less impressive than it sounds. */ + onionskins_n_processed[rpl.handshake_type] /= 2; + onionskins_usec_internal[rpl.handshake_type] /= 2; + onionskins_usec_roundtrip[rpl.handshake_type] /= 2; } } - /* Find the circ it was talking about */ - chan_id = job->chan_id; - circ_id = job->circ_id; + } + /* Find the circ it was talking about */ + chan_id = job->chan_id; + circ_id = job->circ_id; + + p_chan = channel_find_by_global_id(chan_id); - p_chan = channel_find_by_global_id(chan_id); + if (p_chan) + circ = circuit_get_by_circid_channel(circ_id, p_chan); - if (p_chan) - circ = circuit_get_by_circid_channel(circ_id, p_chan); + log_debug(LD_OR, + "Unpacking cpuworker reply %p, chan_id is " U64_FORMAT + ", circ_id is %u, p_chan=%p, circ=%p, success=%d", + job, U64_PRINTF_ARG(chan_id), (unsigned)circ_id, + p_chan, circ, rpl.success); + if (rpl.success == 0) { log_debug(LD_OR, - "Unpacking cpuworker reply %p, chan_id is " U64_FORMAT - ", circ_id is %u, p_chan=%p, circ=%p, success=%d", - job, U64_PRINTF_ARG(chan_id), (unsigned)circ_id, - p_chan, circ, rpl.success); - - if (rpl.success == 0) { - log_debug(LD_OR, - "decoding onionskin failed. " - "(Old key or bad software.) Closing."); - if (circ) - circuit_mark_for_close(circ, END_CIRC_REASON_TORPROTOCOL); - goto done_processing; - } - if (!circ) { - /* This happens because somebody sends us a destroy cell and the - * circuit goes away, while the cpuworker is working. This is also - * why our tag doesn't include a pointer to the circ, because we'd - * never know if it's still valid. - */ - log_debug(LD_OR,"processed onion for a circ that's gone. Dropping."); - goto done_processing; - } - tor_assert(! CIRCUIT_IS_ORIGIN(circ)); - TO_OR_CIRCUIT(circ)->workqueue_entry = NULL; - if (onionskin_answer(TO_OR_CIRCUIT(circ), - &rpl.created_cell, - (const char*)rpl.keys, - rpl.rend_auth_material) < 0) { - log_warn(LD_OR,"onionskin_answer failed. Closing."); - circuit_mark_for_close(circ, END_CIRC_REASON_INTERNAL); - goto done_processing; - } - log_debug(LD_OR,"onionskin_answer succeeded. Yay."); + "decoding onionskin failed. " + "(Old key or bad software.) Closing."); + if (circ) + circuit_mark_for_close(circ, END_CIRC_REASON_TORPROTOCOL); + goto done_processing; + } + if (!circ) { + /* This happens because somebody sends us a destroy cell and the + * circuit goes away, while the cpuworker is working. This is also + * why our tag doesn't include a pointer to the circ, because we'd + * never know if it's still valid. + */ + log_debug(LD_OR,"processed onion for a circ that's gone. Dropping."); + goto done_processing; } + tor_assert(! CIRCUIT_IS_ORIGIN(circ)); + TO_OR_CIRCUIT(circ)->workqueue_entry = NULL; + if (onionskin_answer(TO_OR_CIRCUIT(circ), + &rpl.created_cell, + (const char*)rpl.keys, + rpl.rend_auth_material) < 0) { + log_warn(LD_OR,"onionskin_answer failed. Closing."); + circuit_mark_for_close(circ, END_CIRC_REASON_INTERNAL); + goto done_processing; + } + log_debug(LD_OR,"onionskin_answer succeeded. Yay."); + done_processing: memwipe(&rpl, 0, sizeof(rpl)); @@ -408,56 +407,54 @@ cpuworker_onion_handshake_threadfn(void *state_, void *work_) tor_assert(req.magic == CPUWORKER_REQUEST_MAGIC); memset(&rpl, 0, sizeof(rpl)); - if (1) { - const create_cell_t *cc = &req.create_cell; - created_cell_t *cell_out = &rpl.created_cell; - struct timeval tv_start = {0,0}, tv_end; - int n; - rpl.timed = req.timed; - rpl.started_at = req.started_at; - rpl.handshake_type = cc->handshake_type; - if (req.timed) - tor_gettimeofday(&tv_start); - n = onion_skin_server_handshake(cc->handshake_type, - cc->onionskin, cc->handshake_len, - onion_keys, - cell_out->reply, - rpl.keys, CPATH_KEY_MATERIAL_LEN, - rpl.rend_auth_material); - if (n < 0) { - /* failure */ - log_debug(LD_OR,"onion_skin_server_handshake failed."); - memset(&rpl, 0, sizeof(rpl)); - rpl.success = 0; - } else { - /* success */ - log_debug(LD_OR,"onion_skin_server_handshake succeeded."); - cell_out->handshake_len = n; - switch (cc->cell_type) { - case CELL_CREATE: - cell_out->cell_type = CELL_CREATED; break; - case CELL_CREATE2: - cell_out->cell_type = CELL_CREATED2; break; - case CELL_CREATE_FAST: - cell_out->cell_type = CELL_CREATED_FAST; break; - default: - tor_assert(0); - return WQ_RPL_SHUTDOWN; - } - rpl.success = 1; - } - rpl.magic = CPUWORKER_REPLY_MAGIC; - if (req.timed) { - struct timeval tv_diff; - int64_t usec; - tor_gettimeofday(&tv_end); - timersub(&tv_end, &tv_start, &tv_diff); - usec = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec; - if (usec < 0 || usec > MAX_BELIEVABLE_ONIONSKIN_DELAY) - rpl.n_usec = MAX_BELIEVABLE_ONIONSKIN_DELAY; - else - rpl.n_usec = (uint32_t) usec; - } + const create_cell_t *cc = &req.create_cell; + created_cell_t *cell_out = &rpl.created_cell; + struct timeval tv_start = {0,0}, tv_end; + int n; + rpl.timed = req.timed; + rpl.started_at = req.started_at; + rpl.handshake_type = cc->handshake_type; + if (req.timed) + tor_gettimeofday(&tv_start); + n = onion_skin_server_handshake(cc->handshake_type, + cc->onionskin, cc->handshake_len, + onion_keys, + cell_out->reply, + rpl.keys, CPATH_KEY_MATERIAL_LEN, + rpl.rend_auth_material); + if (n < 0) { + /* failure */ + log_debug(LD_OR,"onion_skin_server_handshake failed."); + memset(&rpl, 0, sizeof(rpl)); + rpl.success = 0; + } else { + /* success */ + log_debug(LD_OR,"onion_skin_server_handshake succeeded."); + cell_out->handshake_len = n; + switch (cc->cell_type) { + case CELL_CREATE: + cell_out->cell_type = CELL_CREATED; break; + case CELL_CREATE2: + cell_out->cell_type = CELL_CREATED2; break; + case CELL_CREATE_FAST: + cell_out->cell_type = CELL_CREATED_FAST; break; + default: + tor_assert(0); + return WQ_RPL_SHUTDOWN; + } + rpl.success = 1; + } + rpl.magic = CPUWORKER_REPLY_MAGIC; + if (req.timed) { + struct timeval tv_diff; + int64_t usec; + tor_gettimeofday(&tv_end); + timersub(&tv_end, &tv_start, &tv_diff); + usec = ((int64_t)tv_diff.tv_sec)*1000000 + tv_diff.tv_usec; + if (usec < 0 || usec > MAX_BELIEVABLE_ONIONSKIN_DELAY) + rpl.n_usec = MAX_BELIEVABLE_ONIONSKIN_DELAY; + else + rpl.n_usec = (uint32_t) usec; } memcpy(&job->u.reply, &rpl, sizeof(rpl)); @@ -499,58 +496,57 @@ assign_onionskin_to_cpuworker(or_circuit_t *circ, cpuworker_request_t req; int should_time; - if (1) { - if (!circ->p_chan) { - log_info(LD_OR,"circ->p_chan gone. Failing circ."); + if (!circ->p_chan) { + log_info(LD_OR,"circ->p_chan gone. Failing circ."); + tor_free(onionskin); + return -1; + } + + if (total_pending_tasks >= max_pending_tasks) { + log_debug(LD_OR,"No idle cpuworkers. Queuing."); + if (onion_pending_add(circ, onionskin) < 0) { tor_free(onionskin); return -1; } + return 0; + } - if (total_pending_tasks >= max_pending_tasks) { - log_debug(LD_OR,"No idle cpuworkers. Queuing."); - if (onion_pending_add(circ, onionskin) < 0) { - tor_free(onionskin); - return -1; - } - return 0; - } + if (connection_or_digest_is_known_relay(circ->p_chan->identity_digest)) + rep_hist_note_circuit_handshake_assigned(onionskin->handshake_type); - if (connection_or_digest_is_known_relay(circ->p_chan->identity_digest)) - rep_hist_note_circuit_handshake_assigned(onionskin->handshake_type); + should_time = should_time_request(onionskin->handshake_type); + memset(&req, 0, sizeof(req)); + req.magic = CPUWORKER_REQUEST_MAGIC; + req.timed = should_time; - should_time = should_time_request(onionskin->handshake_type); - memset(&req, 0, sizeof(req)); - req.magic = CPUWORKER_REQUEST_MAGIC; - req.timed = should_time; + memcpy(&req.create_cell, onionskin, sizeof(create_cell_t)); - memcpy(&req.create_cell, onionskin, sizeof(create_cell_t)); + tor_free(onionskin); - tor_free(onionskin); + if (should_time) + tor_gettimeofday(&req.started_at); - if (should_time) - tor_gettimeofday(&req.started_at); - - job = tor_malloc_zero(sizeof(cpuworker_job_t)); - job->chan_id = circ->p_chan->global_identifier; - job->circ_id = circ->p_circ_id; - memcpy(&job->u.request, &req, sizeof(req)); - memwipe(&req, 0, sizeof(req)); - - ++total_pending_tasks; - queue_entry = threadpool_queue_work(threadpool, - cpuworker_onion_handshake_threadfn, - cpuworker_onion_handshake_replyfn, - job); - if (!queue_entry) { - log_warn(LD_BUG, "Couldn't queue work on threadpool"); - tor_free(job); - return -1; - } - log_debug(LD_OR, "Queued task %p (qe=%p, chanid="U64_FORMAT", circid=%u)", - job, queue_entry, U64_PRINTF_ARG(job->chan_id), job->circ_id); + job = tor_malloc_zero(sizeof(cpuworker_job_t)); + job->chan_id = circ->p_chan->global_identifier; + job->circ_id = circ->p_circ_id; + memcpy(&job->u.request, &req, sizeof(req)); + memwipe(&req, 0, sizeof(req)); - circ->workqueue_entry = queue_entry; + ++total_pending_tasks; + queue_entry = threadpool_queue_work(threadpool, + cpuworker_onion_handshake_threadfn, + cpuworker_onion_handshake_replyfn, + job); + if (!queue_entry) { + log_warn(LD_BUG, "Couldn't queue work on threadpool"); + tor_free(job); + return -1; } + log_debug(LD_OR, "Queued task %p (qe=%p, chanid="U64_FORMAT", circid=%u)", + job, queue_entry, U64_PRINTF_ARG(job->chan_id), job->circ_id); + + circ->workqueue_entry = queue_entry; + return 0; }

1 0

[tor/master] Add shutdown and broadcast support to test_workqueue.
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit ebbc177005eaf9bd949daba657b2c703a7bd1769 Author: Nick Mathewson <nickm(a)torproject.org> Date: Sat Sep 28 00:09:20 2013 -0400 Add shutdown and broadcast support to test_workqueue. --- src/test/test_workqueue.c | 26 ++++++++++++++++++++------ 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/src/test/test_workqueue.c b/src/test/test_workqueue.c index 7ef54ef..cbf9d81 100644 --- a/src/test/test_workqueue.c +++ b/src/test/test_workqueue.c @@ -36,6 +36,7 @@ typedef struct state_s { int n_handled; crypto_pk_t *rsa; curve25519_secret_key_t ecdh; + int is_shutdown; } state_t; typedef struct rsa_work_s { @@ -94,18 +95,15 @@ workqueue_do_rsa(void *state, void *work) return WQ_RPL_REPLY; } -#if 0 static int workqueue_do_shutdown(void *state, void *work) { (void)state; (void)work; - (void)cmd; crypto_pk_free(((state_t*)state)->rsa); tor_free(state); return WQ_RPL_SHUTDOWN; } -#endif static int workqueue_do_ecdh(void *state, void *work) @@ -197,6 +195,20 @@ add_work(threadpool_t *tp) } } +static int shutting_down = 0; +static int n_shutdowns_done = 0; + +static void +shutdown_reply(void *arg) +{ + (void)arg; + tor_assert(shutting_down); + ++n_shutdowns_done; + if (n_shutdowns_done == opt_n_threads) { + tor_event_base_loopexit(tor_libevent_get_base(), NULL); + } +} + static void replysock_readable_cb(tor_socket_t sock, short what, void *arg) { @@ -236,8 +248,9 @@ replysock_readable_cb(tor_socket_t sock, short what, void *arg) } } - if (n_received == n_sent && n_sent >= opt_n_items) { - tor_event_base_loopexit(tor_libevent_get_base(), NULL); + if (shutting_down == 0 && n_received == n_sent && n_sent >= opt_n_items) { + shutting_down = 1; + threadpool_queue_for_all(tp, NULL, workqueue_do_shutdown, shutdown_reply, NULL); } } @@ -345,7 +358,8 @@ main(int argc, char **argv) event_base_loop(tor_libevent_get_base(), 0); - if (n_sent != opt_n_items || n_received != n_sent) { + if (n_sent != opt_n_items || n_received != n_sent || + n_shutdowns_done != opt_n_threads) { puts("FAIL"); return 1; } else {

1 0

[tor/master] Test and fix workqueue_entry_cancel().
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit e5f8c772f4c468a20da8b9176c2b276ac76bbe78 Author: Nick Mathewson <nickm(a)torproject.org> Date: Sat Sep 28 00:33:10 2013 -0400 Test and fix workqueue_entry_cancel(). --- src/common/workqueue.c | 12 ++++--- src/common/workqueue.h | 2 +- src/test/test_workqueue.c | 77 ++++++++++++++++++++++++++++++++++++--------- 3 files changed, 71 insertions(+), 20 deletions(-) diff --git a/src/common/workqueue.c b/src/common/workqueue.c index 9293e1f..44cf98d 100644 --- a/src/common/workqueue.c +++ b/src/common/workqueue.c @@ -119,27 +119,29 @@ workqueue_entry_free(workqueue_entry_t *ent) * executed in the main thread; that will cause undefined behavior (probably, * a crash). * - * If the work is cancelled, this function return 1. It is the caller's - * responsibility to free any storage in the work function's arguments. + * 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 0. + * or begun to execute the work item. In that case, it will return NULL. */ -int +void * workqueue_entry_cancel(workqueue_entry_t *ent) { int cancelled = 0; + void *result = NULL; tor_mutex_acquire(&ent->on_thread->lock); if (ent->pending) { TOR_TAILQ_REMOVE(&ent->on_thread->work, ent, next_work); cancelled = 1; + result = ent->arg; } tor_mutex_release(&ent->on_thread->lock); if (cancelled) { tor_free(ent); } - return cancelled; + return result; } /** diff --git a/src/common/workqueue.h b/src/common/workqueue.h index 5a6cd80..ec1f7c9 100644 --- a/src/common/workqueue.h +++ b/src/common/workqueue.h @@ -32,7 +32,7 @@ int threadpool_queue_for_all(threadpool_t *pool, int (*fn)(void *, void *), void (*reply_fn)(void *), void *arg); -int workqueue_entry_cancel(workqueue_entry_t *pending_work); +void *workqueue_entry_cancel(workqueue_entry_t *pending_work); threadpool_t *threadpool_new(int n_threads, replyqueue_t *replyqueue, void *(*new_thread_state_fn)(void*), diff --git a/src/test/test_workqueue.c b/src/test/test_workqueue.c index cbf9d81..6de6f03 100644 --- a/src/test/test_workqueue.c +++ b/src/test/test_workqueue.c @@ -23,6 +23,7 @@ static int opt_n_threads = 8; static int opt_n_items = 10000; static int opt_n_inflight = 1000; static int opt_n_lowwater = 250; +static int opt_n_cancel = 0; static int opt_ratio_rsa = 5; #ifdef TRACK_RESPONSES @@ -172,27 +173,68 @@ handle_reply(void *arg) ++n_received; } -static int +static workqueue_entry_t * add_work(threadpool_t *tp) { int add_rsa = opt_ratio_rsa == 0 || tor_weak_random_range(&weak_rng, opt_ratio_rsa) == 0; + if (add_rsa) { rsa_work_t *w = tor_malloc_zero(sizeof(*w)); w->serial = n_sent++; crypto_rand((char*)w->msg, 20); w->msglen = 20; ++rsa_sent; - return threadpool_queue_work(tp, workqueue_do_rsa, handle_reply, w) != NULL; + return threadpool_queue_work(tp, workqueue_do_rsa, handle_reply, w); } else { ecdh_work_t *w = tor_malloc_zero(sizeof(*w)); w->serial = n_sent++; /* Not strictly right, but this is just for benchmarks. */ crypto_rand((char*)w->u.pk.public_key, 32); ++ecdh_sent; - return threadpool_queue_work(tp, workqueue_do_ecdh, handle_reply, w) != NULL; + return threadpool_queue_work(tp, workqueue_do_ecdh, handle_reply, w); + } +} + +static int n_failed_cancel = 0; +static int n_successful_cancel = 0; + +static int +add_n_work_items(threadpool_t *tp, int n) +{ + int n_queued = 0; + int n_try_cancel = 0, i; + workqueue_entry_t **to_cancel; + workqueue_entry_t *ent; + + to_cancel = tor_malloc(sizeof(workqueue_entry_t*) * opt_n_cancel); + + while (n_queued++ < n) { + ent = add_work(tp); + if (! ent) { + puts("Couldn't add work."); + tor_event_base_loopexit(tor_libevent_get_base(), NULL); + return -1; + } + if (n_try_cancel < opt_n_cancel && + tor_weak_random_range(&weak_rng, n) < opt_n_cancel) { + to_cancel[n_try_cancel++] = ent; + } + } + + for (i = 0; i < n_try_cancel; ++i) { + void *work = workqueue_entry_cancel(to_cancel[i]); + if (! work) { + n_failed_cancel++; + } else { + n_successful_cancel++; + tor_free(work); + } } + + tor_free(to_cancel); + return 0; } static int shutting_down = 0; @@ -223,8 +265,13 @@ replysock_readable_cb(tor_socket_t sock, short what, void *arg) if (old_r == n_received) return; - if (opt_verbose) - printf("%d / %d\n", n_received, n_sent); + if (opt_verbose) { + printf("%d / %d", n_received, n_sent); + if (opt_n_cancel) + printf(" (%d cancelled, %d uncancellable)", + n_successful_cancel, n_failed_cancel); + puts(""); + } #ifdef TRACK_RESPONSES tor_mutex_acquire(&bitmap_mutex); for (i = 0; i < opt_n_items; ++i) { @@ -239,16 +286,14 @@ replysock_readable_cb(tor_socket_t sock, short what, void *arg) tor_mutex_release(&bitmap_mutex); #endif - if (n_sent - n_received < opt_n_lowwater) { - while (n_sent < n_received + opt_n_inflight && n_sent < opt_n_items) { - if (! add_work(tp)) { - puts("Couldn't add work."); - tor_event_base_loopexit(tor_libevent_get_base(), NULL); - } - } + if (n_sent - (n_received+n_successful_cancel) < opt_n_lowwater) { + int n_to_send = n_received + opt_n_inflight - n_sent; + if (n_to_send > opt_n_items - n_sent) + n_to_send = opt_n_items - n_sent; + add_n_work_items(tp, n_to_send); } - if (shutting_down == 0 && n_received == n_sent && n_sent >= opt_n_items) { + if (shutting_down == 0 && n_received+n_successful_cancel == n_sent && n_sent >= opt_n_items) { shutting_down = 1; threadpool_queue_for_all(tp, NULL, workqueue_do_shutdown, shutdown_reply, NULL); } @@ -263,6 +308,7 @@ help(void) " -T <threads> Use this many threads\n" " -I <inflight> Have no more than this many requests queued at once\n" " -L <lowwater> Add items whenever fewer than this many are pending\n" + " -C <cancel> Try to cancel N items of every batch that we add\n" " -R <ratio> Make one out of this many items be a slow (RSA) one\n" " --no-{eventfd2,eventfd,pipe2,pipe,socketpair}\n" " Disable one of the alert_socket backends."); @@ -291,6 +337,8 @@ main(int argc, char **argv) opt_n_lowwater = atoi(argv[++i]); } else if (!strcmp(argv[i], "-R") && i+1<argc) { opt_ratio_rsa = atoi(argv[++i]); + } else if (!strcmp(argv[i], "-C") && i+1<argc) { + opt_n_cancel = atoi(argv[++i]); } else if (!strcmp(argv[i], "--no-eventfd2")) { as_flags |= ASOCKS_NOEVENTFD2; } else if (!strcmp(argv[i], "--no-eventfd")) { @@ -311,6 +359,7 @@ main(int argc, char **argv) } if (opt_n_threads < 1 || opt_n_items < 1 || opt_n_inflight < 1 || opt_n_lowwater < 0 || + opt_n_cancel > opt_n_inflight || opt_ratio_rsa < 0) { help(); return 1; @@ -358,7 +407,7 @@ main(int argc, char **argv) event_base_loop(tor_libevent_get_base(), 0); - if (n_sent != opt_n_items || n_received != n_sent || + if (n_sent != opt_n_items || n_received+n_successful_cancel != n_sent || n_shutdowns_done != opt_n_threads) { puts("FAIL"); return 1;

1 0

[tor/master] Add unit test for thread IDs.
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit 81354b081b7bb9deabd6c53e48623190b01aab1c Author: Nick Mathewson <nickm(a)torproject.org> Date: Fri Sep 27 23:20:22 2013 -0400 Add unit test for thread IDs. --- src/test/test_threads.c | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/src/test/test_threads.c b/src/test/test_threads.c index d2a61a1..2bc24e1 100644 --- a/src/test/test_threads.c +++ b/src/test/test_threads.c @@ -21,6 +21,10 @@ static char *thread1_name_ = NULL; /** The name of thread2 for the thread test */ static char *thread2_name_ = NULL; +static int thread_fns_failed = 0; + +static unsigned long thread_fn_tid1, thread_fn_tid2; + static void thread_test_func_(void* _s) ATTR_NORETURN; /** How many iterations have the threads in the unit test run? */ @@ -42,10 +46,12 @@ thread_test_func_(void* _s) m = thread_test_start1_; cp = &thread1_name_; count = &t1_count; + thread_fn_tid1 = tor_get_thread_id(); } else { m = thread_test_start2_; cp = &thread2_name_; count = &t2_count; + thread_fn_tid2 = tor_get_thread_id(); } tor_snprintf(buf, sizeof(buf), "%lu", tor_get_thread_id()); @@ -61,6 +67,8 @@ thread_test_func_(void* _s) } tor_mutex_acquire(thread_test_mutex_); strmap_set(thread_test_strmap_, s, *cp); + if (in_main_thread()) + ++thread_fns_failed; tor_mutex_release(thread_test_mutex_); tor_mutex_release(m); @@ -80,7 +88,10 @@ test_threads_basic(void *arg) tv.tv_sec=0; tv.tv_usec=100*1000; #endif - (void)arg; + (void) arg; + + set_main_thread(); + thread_test_mutex_ = tor_mutex_new(); thread_test_start1_ = tor_mutex_new(); thread_test_start2_ = tor_mutex_new(); @@ -131,6 +142,9 @@ test_threads_basic(void *arg) !strcmp(strmap_get(thread_test_strmap_, "thread 2"), strmap_get(thread_test_strmap_, "last to run"))); + tt_int_op(thread_fns_failed, ==, 0); + tt_int_op(thread_fn_tid1, !=, thread_fn_tid2); + done: tor_free(s1); tor_free(s2); @@ -188,7 +202,7 @@ cv_test_thr_fn_(void *arg) tid = i->n_threads++; tor_mutex_release(i->mutex); (void) tid; - + tor_mutex_acquire(i->mutex); while (1) { if (i->addend) { @@ -299,4 +313,3 @@ struct testcase_t thread_tests[] = { &passthrough_setup, (void*)"tv" }, END_OF_TESTCASES }; -

1 0

[tor/master] Fix check-spaces
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit cc6529e9bb7d7e01a25b5632d6d6c2424c6fc2b4 Author: Nick Mathewson <nickm(a)torproject.org> Date: Sat Sep 28 00:52:28 2013 -0400 Fix check-spaces --- src/common/compat_pthreads.c | 1 + src/common/compat_threads.c | 1 + src/common/compat_threads.h | 2 +- src/common/compat_winthreads.c | 3 +-- src/common/workqueue.h | 1 + src/test/test.c | 1 - src/test/test_threads.c | 1 + src/test/test_workqueue.c | 8 ++++++-- 8 files changed, 12 insertions(+), 6 deletions(-) diff --git a/src/common/compat_pthreads.c b/src/common/compat_pthreads.c index 69f7bac..848bfe0 100644 --- a/src/common/compat_pthreads.c +++ b/src/common/compat_pthreads.c @@ -240,3 +240,4 @@ tor_threads_init(void) set_main_thread(); } } + diff --git a/src/common/compat_threads.c b/src/common/compat_threads.c index f018475..7944007 100644 --- a/src/common/compat_threads.c +++ b/src/common/compat_threads.c @@ -253,3 +253,4 @@ alert_sockets_close(alert_sockets_t *socks) } socks->read_fd = socks->write_fd = -1; } + diff --git a/src/common/compat_threads.h b/src/common/compat_threads.h index a5db72d..acf3083 100644 --- a/src/common/compat_threads.h +++ b/src/common/compat_threads.h @@ -44,7 +44,6 @@ typedef struct tor_mutex_t { #endif } tor_mutex_t; - tor_mutex_t *tor_mutex_new(void); tor_mutex_t *tor_mutex_new_nonrecursive(void); void tor_mutex_init(tor_mutex_t *m); @@ -113,3 +112,4 @@ int alert_sockets_create(alert_sockets_t *socks_out, uint32_t flags); void alert_sockets_close(alert_sockets_t *socks); #endif + diff --git a/src/common/compat_winthreads.c b/src/common/compat_winthreads.c index 4820eb3..71b994c 100644 --- a/src/common/compat_winthreads.c +++ b/src/common/compat_winthreads.c @@ -33,7 +33,6 @@ spawn_func(void (*func)(void *), void *data) return 0; } - /** End the current thread/process. */ void @@ -46,7 +45,6 @@ spawn_exit(void) _exit(0); } - void tor_mutex_init(tor_mutex_t *m) { @@ -195,3 +193,4 @@ tor_threads_init(void) { set_main_thread(); } + diff --git a/src/common/workqueue.h b/src/common/workqueue.h index ec1f7c9..aa8620d 100644 --- a/src/common/workqueue.h +++ b/src/common/workqueue.h @@ -45,3 +45,4 @@ tor_socket_t replyqueue_get_socket(replyqueue_t *rq); void replyqueue_process(replyqueue_t *queue); #endif + diff --git a/src/test/test.c b/src/test/test.c index 9171306..2c2328c 100644 --- a/src/test/test.c +++ b/src/test/test.c @@ -1258,7 +1258,6 @@ test_stats(void *arg) tor_free(s); } - static void * passthrough_test_setup(const struct testcase_t *testcase) { diff --git a/src/test/test_threads.c b/src/test/test_threads.c index 2bc24e1..c029304 100644 --- a/src/test/test_threads.c +++ b/src/test/test_threads.c @@ -313,3 +313,4 @@ struct testcase_t thread_tests[] = { &passthrough_setup, (void*)"tv" }, END_OF_TESTCASES }; + diff --git a/src/test/test_workqueue.c b/src/test/test_workqueue.c index 6de6f03..410f43c 100644 --- a/src/test/test_workqueue.c +++ b/src/test/test_workqueue.c @@ -293,9 +293,12 @@ replysock_readable_cb(tor_socket_t sock, short what, void *arg) add_n_work_items(tp, n_to_send); } - if (shutting_down == 0 && n_received+n_successful_cancel == n_sent && n_sent >= opt_n_items) { + if (shutting_down == 0 && + n_received+n_successful_cancel == n_sent && + n_sent >= opt_n_items) { shutting_down = 1; - threadpool_queue_for_all(tp, NULL, workqueue_do_shutdown, shutdown_reply, NULL); + threadpool_queue_for_all(tp, NULL, + workqueue_do_shutdown, shutdown_reply, NULL); } } @@ -416,3 +419,4 @@ main(int argc, char **argv) return 0; } } +

1 0

[tor/master] Support monotonic time for pthread_cond_timedwait
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit d684dbb0c76c4de0b21fabee99fc08833249bd87 Author: David Goulet <dgoulet(a)ev0ke.net> Date: Wed Jan 21 13:18:56 2015 -0500 Support monotonic time for pthread_cond_timedwait This is to avoid that the pthread_cond_timedwait() is not affected by time adjustment which could make the waiting period very long or very short which is not what we want in any cases. Signed-off-by: David Goulet <dgoulet(a)ev0ke.net> --- src/common/compat_pthreads.c | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/src/common/compat_pthreads.c b/src/common/compat_pthreads.c index c217c51..e1d4b0e 100644 --- a/src/common/compat_pthreads.c +++ b/src/common/compat_pthreads.c @@ -6,6 +6,7 @@ #include "orconfig.h" #include <pthread.h> #include <signal.h> +#include <time.h> #include "compat.h" #include "torlog.h" @@ -169,8 +170,23 @@ tor_get_thread_id(void) int tor_cond_init(tor_cond_t *cond) { + pthread_condattr_t condattr; + memset(cond, 0, sizeof(tor_cond_t)); - if (pthread_cond_init(&cond->cond, NULL)) { + /* Default condition attribute. Might be used if clock monotonic is + * available else this won't affect anything. */ + if (pthread_condattr_init(&condattr)) { + return -1; + } + +#if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC) + /* Use monotonic time so when we timedwait() on it, any clock adjustment + * won't affect the timeout value. */ + if (pthread_condattr_setclock(&condattr, CLOCK_MONOTONIC)) { + return -1; + } +#endif + if (pthread_cond_init(&cond->cond, &condattr)) { return -1; } return 0; @@ -209,12 +225,22 @@ tor_cond_wait(tor_cond_t *cond, tor_mutex_t *mutex, const struct timeval *tv) return r ? -1 : 0; } } else { - struct timespec ts; struct timeval tvnow, tvsum; + struct timespec ts; while (1) { +#if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC) + if (clock_gettime(CLOCK_MONOTONIC, &ts) < 0) { + return -1; + } + tvnow.tv_sec = ts.tv_sec; + tvnow.tv_usec = ts.tv_nsec / 1000; + timeradd(tv, &tvnow, &tvsum); +#else if (gettimeofday(&tvnow, NULL) < 0) return -1; timeradd(tv, &tvnow, &tvsum); +#endif /* HAVE_CLOCK_GETTIME, CLOCK_MONOTONIC */ + ts.tv_sec = tvsum.tv_sec; ts.tv_nsec = tvsum.tv_usec * 1000;

1 0

[tor/master] Basic unit test for condition variables.
by nickm＠torproject.org 21 Jan '15

21 Jan '15

commit 7a63005220938b30df41b51334942d7d79c14cf9 Author: Nick Mathewson <nickm(a)torproject.org> Date: Fri Sep 27 23:15:53 2013 -0400 Basic unit test for condition variables. --- configure.ac | 2 +- src/test/test.c | 18 ++++++ src/test/test.h | 2 + src/test/test_crypto.c | 36 ++++-------- src/test/test_threads.c | 150 ++++++++++++++++++++++++++++++++++++++++++++++- src/test/test_util.c | 21 +------ 6 files changed, 183 insertions(+), 46 deletions(-) diff --git a/configure.ac b/configure.ac index 69a266a..929b701 100644 --- a/configure.ac +++ b/configure.ac @@ -437,7 +437,7 @@ AC_CHECK_FUNCS( sysconf \ sysctl \ uname \ - usleep \ + usleep \ vasprintf \ _vscprintf ) diff --git a/src/test/test.c b/src/test/test.c index edc28cd..9171306 100644 --- a/src/test/test.c +++ b/src/test/test.c @@ -1258,6 +1258,24 @@ test_stats(void *arg) tor_free(s); } + +static void * +passthrough_test_setup(const struct testcase_t *testcase) +{ + return testcase->setup_data; +} +static int +passthrough_test_cleanup(const struct testcase_t *testcase, void *ptr) +{ + (void)testcase; + (void)ptr; + return 1; +} + +const struct testcase_setup_t passthrough_setup = { + passthrough_test_setup, passthrough_test_cleanup +}; + #define ENT(name) \ { #name, test_ ## name , 0, NULL, NULL } #define FORK(name) \ diff --git a/src/test/test.h b/src/test/test.h index 48037a5..b8057c5 100644 --- a/src/test/test.h +++ b/src/test/test.h @@ -158,5 +158,7 @@ crypto_pk_t *pk_generate(int idx); #define NS_MOCK(name) MOCK(name, NS(name)) #define NS_UNMOCK(name) UNMOCK(name) +extern const struct testcase_setup_t passthrough_setup; + #endif diff --git a/src/test/test_crypto.c b/src/test/test_crypto.c index 4a5a12c..8426c71 100644 --- a/src/test/test_crypto.c +++ b/src/test/test_crypto.c @@ -1975,30 +1975,14 @@ test_crypto_siphash(void *arg) ; } -static void * -pass_data_setup_fn(const struct testcase_t *testcase) -{ - return testcase->setup_data; -} -static int -pass_data_cleanup_fn(const struct testcase_t *testcase, void *ptr) -{ - (void)ptr; - (void)testcase; - return 1; -} -static const struct testcase_setup_t pass_data = { - pass_data_setup_fn, pass_data_cleanup_fn -}; - #define CRYPTO_LEGACY(name) \ { #name, test_crypto_ ## name , 0, NULL, NULL } struct testcase_t crypto_tests[] = { CRYPTO_LEGACY(formats), CRYPTO_LEGACY(rng), - { "aes_AES", test_crypto_aes, TT_FORK, &pass_data, (void*)"aes" }, - { "aes_EVP", test_crypto_aes, TT_FORK, &pass_data, (void*)"evp" }, + { "aes_AES", test_crypto_aes, TT_FORK, &passthrough_setup, (void*)"aes" }, + { "aes_EVP", test_crypto_aes, TT_FORK, &passthrough_setup, (void*)"evp" }, CRYPTO_LEGACY(sha), CRYPTO_LEGACY(pk), { "pk_fingerprints", test_crypto_pk_fingerprints, TT_FORK, NULL, NULL }, @@ -2006,23 +1990,25 @@ struct testcase_t crypto_tests[] = { CRYPTO_LEGACY(dh), CRYPTO_LEGACY(s2k_rfc2440), #ifdef HAVE_LIBSCRYPT_H - { "s2k_scrypt", test_crypto_s2k_general, 0, &pass_data, + { "s2k_scrypt", test_crypto_s2k_general, 0, &passthrough_setup, (void*)"scrypt" }, - { "s2k_scrypt_low", test_crypto_s2k_general, 0, &pass_data, + { "s2k_scrypt_low", test_crypto_s2k_general, 0, &passthrough_setup, (void*)"scrypt-low" }, #endif - { "s2k_pbkdf2", test_crypto_s2k_general, 0, &pass_data, + { "s2k_pbkdf2", test_crypto_s2k_general, 0, &passthrough_setup, (void*)"pbkdf2" }, - { "s2k_rfc2440_general", test_crypto_s2k_general, 0, &pass_data, + { "s2k_rfc2440_general", test_crypto_s2k_general, 0, &passthrough_setup, (void*)"rfc2440" }, - { "s2k_rfc2440_legacy", test_crypto_s2k_general, 0, &pass_data, + { "s2k_rfc2440_legacy", test_crypto_s2k_general, 0, &passthrough_setup, (void*)"rfc2440-legacy" }, { "s2k_errors", test_crypto_s2k_errors, 0, NULL, NULL }, { "scrypt_vectors", test_crypto_scrypt_vectors, 0, NULL, NULL }, { "pbkdf2_vectors", test_crypto_pbkdf2_vectors, 0, NULL, NULL }, { "pwbox", test_crypto_pwbox, 0, NULL, NULL }, - { "aes_iv_AES", test_crypto_aes_iv, TT_FORK, &pass_data, (void*)"aes" }, - { "aes_iv_EVP", test_crypto_aes_iv, TT_FORK, &pass_data, (void*)"evp" }, + { "aes_iv_AES", test_crypto_aes_iv, TT_FORK, &passthrough_setup, + (void*)"aes" }, + { "aes_iv_EVP", test_crypto_aes_iv, TT_FORK, &passthrough_setup, + (void*)"evp" }, CRYPTO_LEGACY(base32_decode), { "kdf_TAP", test_crypto_kdf_TAP, 0, NULL, NULL }, { "hkdf_sha256", test_crypto_hkdf_sha256, 0, NULL, NULL }, diff --git a/src/test/test_threads.c b/src/test/test_threads.c index 2b4c933..d2a61a1 100644 --- a/src/test/test_threads.c +++ b/src/test/test_threads.c @@ -144,11 +144,159 @@ test_threads_basic(void *arg) tor_mutex_free(thread_test_start2_); } -#define THREAD_TEST(name) \ +typedef struct cv_testinfo_s { + tor_cond_t *cond; + tor_mutex_t *mutex; + int value; + int addend; + int shutdown; + int n_shutdown; + int n_wakeups; + int n_timeouts; + int n_threads; + const struct timeval *tv; +} cv_testinfo_t; + +static cv_testinfo_t * +cv_testinfo_new(void) +{ + cv_testinfo_t *i = tor_malloc_zero(sizeof(*i)); + i->cond = tor_cond_new(); + i->mutex = tor_mutex_new_nonrecursive(); + return i; +} + +static void +cv_testinfo_free(cv_testinfo_t *i) +{ + if (!i) + return; + tor_cond_free(i->cond); + tor_mutex_free(i->mutex); + tor_free(i); +} + +static void cv_test_thr_fn_(void *arg) ATTR_NORETURN; + +static void +cv_test_thr_fn_(void *arg) +{ + cv_testinfo_t *i = arg; + int tid, r; + + tor_mutex_acquire(i->mutex); + tid = i->n_threads++; + tor_mutex_release(i->mutex); + (void) tid; + + tor_mutex_acquire(i->mutex); + while (1) { + if (i->addend) { + i->value += i->addend; + i->addend = 0; + } + + if (i->shutdown) { + ++i->n_shutdown; + i->shutdown = 0; + tor_mutex_release(i->mutex); + spawn_exit(); + } + r = tor_cond_wait(i->cond, i->mutex, i->tv); + ++i->n_wakeups; + if (r == 1) { + ++i->n_timeouts; + tor_mutex_release(i->mutex); + spawn_exit(); + } + } +} + +static void +test_threads_conditionvar(void *arg) +{ + cv_testinfo_t *ti=NULL; + const struct timeval msec100 = { 0, 100*1000 }; + const int timeout = !strcmp(arg, "tv"); + + ti = cv_testinfo_new(); + if (timeout) { + ti->tv = &msec100; + } + spawn_func(cv_test_thr_fn_, ti); + spawn_func(cv_test_thr_fn_, ti); + spawn_func(cv_test_thr_fn_, ti); + spawn_func(cv_test_thr_fn_, ti); + + tor_mutex_acquire(ti->mutex); + ti->addend = 7; + ti->shutdown = 1; + tor_cond_signal_one(ti->cond); + tor_mutex_release(ti->mutex); + +#define SPIN() \ + while (1) { \ + tor_mutex_acquire(ti->mutex); \ + if (ti->addend == 0) { \ + break; \ + } \ + tor_mutex_release(ti->mutex); \ + } + + SPIN(); + + ti->addend = 30; + ti->shutdown = 1; + tor_cond_signal_all(ti->cond); + tor_mutex_release(ti->mutex); + SPIN(); + + ti->addend = 1000; + if (! timeout) ti->shutdown = 1; + tor_cond_signal_one(ti->cond); + tor_mutex_release(ti->mutex); + SPIN(); + ti->addend = 300; + if (! timeout) ti->shutdown = 1; + tor_cond_signal_all(ti->cond); + tor_mutex_release(ti->mutex); + + SPIN(); + tor_mutex_release(ti->mutex); + + tt_int_op(ti->value, ==, 1337); + if (!timeout) { + tt_int_op(ti->n_shutdown, ==, 4); + } else { +#ifdef _WIN32 + Sleep(500); /* msec */ +#elif defined(HAVE_USLEEP) + usleep(500*1000); /* usec */ +#else + { + struct tv = { 0, 500*1000 }; + select(0, NULL, NULL, NULL, &tv); + } +#endif + tor_mutex_acquire(ti->mutex); + tt_int_op(ti->n_shutdown, ==, 2); + tt_int_op(ti->n_timeouts, ==, 2); + tor_mutex_release(ti->mutex); + } + + done: + cv_testinfo_free(ti); +} + +#define THREAD_TEST(name) \ { #name, test_threads_##name, TT_FORK, NULL, NULL } struct testcase_t thread_tests[] = { THREAD_TEST(basic), + { "conditionvar", test_threads_conditionvar, TT_FORK, + &passthrough_setup, (void*)"no-tv" }, + { "conditionvar_timeout", test_threads_conditionvar, TT_FORK, + &passthrough_setup, (void*)"tv" }, END_OF_TESTCASES }; diff --git a/src/test/test_util.c b/src/test/test_util.c index b4ee934..97cf387 100644 --- a/src/test/test_util.c +++ b/src/test/test_util.c @@ -4646,23 +4646,6 @@ test_util_socket(void *arg) tor_close_socket(fd4); } -static void * -socketpair_test_setup(const struct testcase_t *testcase) -{ - return testcase->setup_data; -} -static int -socketpair_test_cleanup(const struct testcase_t *testcase, void *ptr) -{ - (void)testcase; - (void)ptr; - return 1; -} - -static const struct testcase_setup_t socketpair_setup = { - socketpair_test_setup, socketpair_test_cleanup -}; - /* Test for socketpair and ersatz_socketpair(). We test them both, since * the latter is a tolerably good way to exersize tor_accept_socket(). */ static void @@ -4837,10 +4820,10 @@ struct testcase_t util_tests[] = { UTIL_TEST(mathlog, 0), UTIL_TEST(weak_random, 0), UTIL_TEST(socket, TT_FORK), - { "socketpair", test_util_socketpair, TT_FORK, &socketpair_setup, + { "socketpair", test_util_socketpair, TT_FORK, &passthrough_setup, (void*)"0" }, { "socketpair_ersatz", test_util_socketpair, TT_FORK, - &socketpair_setup, (void*)"1" }, + &passthrough_setup, (void*)"1" }, UTIL_TEST(max_mem, 0), UTIL_TEST(hostname_validation, 0), UTIL_TEST(ipv4_validation, 0),

1 0