Merge branch 'multi-priority_squashed' into maint-0.3.1

3 files changed, 28 insertions, 11 deletions

diff --git a/src/or/consdiffmgr.c b/src/or/consdiffmgr.c
index a3ffed10db..67a5d0b3c5 100644
--- a/src/or/consdiffmgr.c
+++ b/src/or/consdiffmgr.c
@@ -1616,7 +1616,8 @@ consensus_diff_queue_diff_work(consensus_cache_entry_t *diff_from,
     goto err;
 
   workqueue_entry_t *work;
-  work = cpuworker_queue_work(consensus_diff_worker_threadfn,
+  work = cpuworker_queue_work(WQ_PRI_LOW,
+                              consensus_diff_worker_threadfn,
                               consensus_diff_worker_replyfn,
                               job);
   if (!work)
@@ -1779,7 +1780,8 @@ consensus_queue_compression_work(const char *consensus,
 
   if (background_compression) {
     workqueue_entry_t *work;
-    work = cpuworker_queue_work(consensus_compress_worker_threadfn,
+    work = cpuworker_queue_work(WQ_PRI_LOW,
+                                consensus_compress_worker_threadfn,
                                 consensus_compress_worker_replyfn,
                                 job);
     if (!work) {
diff --git a/src/or/cpuworker.c b/src/or/cpuworker.c
index 1013fa555e..f5fff2b331 100644
--- a/src/or/cpuworker.c
+++ b/src/or/cpuworker.c
@@ -11,8 +11,11 @@
  * The multithreading backend for this module is in workqueue.c; this module
  * specializes workqueue.c.
  *
- * Right now, we only use this for processing onionskins, and invoke it mostly
- * from onion.c.
+ * Right now, we use this infrastructure
+ *  <ul><li>for processing onionskins in onion.c
+ *      <li>for compressing consensuses in consdiffmgr.c,
+ *      <li>and for calculating diffs and compressing them in consdiffmgr.c.
+ *  </ul>
  **/
 #include "or.h"
 #include "channel.h"
@@ -89,7 +92,14 @@ cpu_init(void)
     event_add(reply_event, NULL);
   }
   if (!threadpool) {
-    threadpool = threadpool_new(get_num_cpus(get_options()),
+    /*
+      In our threadpool implementation, half the threads are permissive and
+      half are strict (when it comes to running lower-priority tasks). So we
+      always make sure we have at least two threads, so that there will be at
+      least one thread of each kind.
+    */
+    const int n_threads = get_num_cpus(get_options()) + 1;
+    threadpool = threadpool_new(n_threads,
                                 replyqueue,
                                 worker_state_new,
                                 worker_state_free,
@@ -481,16 +491,18 @@ queue_pending_tasks(void)
 
 /** DOCDOC */
 MOCK_IMPL(workqueue_entry_t *,
-cpuworker_queue_work,(workqueue_reply_t (*fn)(void *, void *),
+cpuworker_queue_work,(workqueue_priority_t priority,
+                      workqueue_reply_t (*fn)(void *, void *),
                       void (*reply_fn)(void *),
                       void *arg))
 {
   tor_assert(threadpool);
 
-  return threadpool_queue_work(threadpool,
-                               fn,
-                               reply_fn,
-                               arg);
+  return threadpool_queue_work_priority(threadpool,
+                                        priority,
+                                        fn,
+                                        reply_fn,
+                                        arg);
 }
 
 /** Try to tell a cpuworker to perform the public key operations necessary to
@@ -545,7 +557,8 @@ assign_onionskin_to_cpuworker(or_circuit_t *circ,
   memwipe(&req, 0, sizeof(req));
 
   ++total_pending_tasks;
-  queue_entry = threadpool_queue_work(threadpool,
+  queue_entry = threadpool_queue_work_priority(threadpool,
+                                      WQ_PRI_HIGH,
                                       cpuworker_onion_handshake_threadfn,
                                       cpuworker_onion_handshake_replyfn,
                                       job);
diff --git a/src/or/cpuworker.h b/src/or/cpuworker.h
index aedf2fae32..320de9532f 100644
--- a/src/or/cpuworker.h
+++ b/src/or/cpuworker.h
@@ -16,7 +16,9 @@ void cpu_init(void);
 void cpuworkers_rotate_keyinfo(void);
 struct workqueue_entry_s;
 enum workqueue_reply_t;
+enum workqueue_priority_t;
 MOCK_DECL(struct workqueue_entry_s *, cpuworker_queue_work, (
+                    enum workqueue_priority_t priority,
                     enum workqueue_reply_t (*fn)(void *, void *),
                     void (*reply_fn)(void *),
                     void *arg));