Fix: Memory leaks in cpuworker on shutdown

Resources allocated by cpuworker weren't being freed on clean shutdown.
This applies for worker threads, worker thread pool, reply queue, reply
event, ...

7 files changed, 104 insertions, 31 deletions

diff --git a/changes/thread-memleak b/changes/thread-memleak
new file mode 100644
index 0000000000..a90792c01e
--- /dev/null
+++ b/changes/thread-memleak
@@ -0,0 +1,3 @@
+  o Minor bugfixes (memory):
+    - Fix memory leaks of the CPU worker code during shutdown. Fixes bug 833;
+      bugfix on 0.3.5.1-alpha.
diff --git a/src/app/main/shutdown.c b/src/app/main/shutdown.c
index b3f1c6d058..325cb27134 100644
--- a/src/app/main/shutdown.c
+++ b/src/app/main/shutdown.c
@@ -1,7 +1,7 @@
 /* Copyright (c) 2001 Matej Pfajfar.
  * Copyright (c) 2001-2004, Roger Dingledine.
  * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
- * Copyright (c) 2007-2021, The Tor Project, Inc. */
+ * Copyright (c) 2007-2024, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
 /**
@@ -19,6 +19,7 @@
 #include "app/main/subsysmgr.h"
 #include "core/mainloop/connection.h"
 #include "core/mainloop/mainloop_pubsub.h"
+#include "core/mainloop/cpuworker.h"
 #include "core/or/channeltls.h"
 #include "core/or/circuitlist.h"
 #include "core/or/circuitmux_ewma.h"
@@ -112,6 +113,7 @@ tor_free_all(int postfork)
   if (!postfork) {
     evdns_shutdown(1);
   }
+  cpuworker_free_all();
   geoip_free_all();
   geoip_stats_free_all();
   routerlist_free_all();
diff --git a/src/core/mainloop/cpuworker.c b/src/core/mainloop/cpuworker.c
index a42dbb528d..b12879a178 100644
--- a/src/core/mainloop/cpuworker.c
+++ b/src/core/mainloop/cpuworker.c
@@ -1,6 +1,6 @@
 /* Copyright (c) 2003-2004, Roger Dingledine.
  * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
- * Copyright (c) 2007-2021, The Tor Project, Inc. */
+ * Copyright (c) 2007-2024, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
 /**
@@ -74,7 +74,6 @@ worker_state_free_void(void *arg)
   worker_state_free_(arg);
 }
 
-static replyqueue_t *replyqueue = NULL;
 static threadpool_t *threadpool = NULL;
 
 static uint32_t total_pending_tasks = 0;
@@ -120,31 +119,33 @@ cpuworker_consensus_has_changed(const networkstatus_t *ns)
 void
 cpuworker_init(void)
 {
-  if (!replyqueue) {
-    replyqueue = replyqueue_new(0);
-  }
-  if (!threadpool) {
-    /*
-      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 = MAX(get_num_cpus(get_options()), 2);
-    threadpool = threadpool_new(n_threads,
-                                replyqueue,
-                                worker_state_new,
-                                worker_state_free_void,
-                                NULL);
-
-    int r = threadpool_register_reply_event(threadpool, NULL);
-
-    tor_assert(r == 0);
-  }
+  /*
+    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 = MAX(get_num_cpus(get_options()), 2);
+  threadpool = threadpool_new(n_threads,
+                              replyqueue_new(0),
+                              worker_state_new,
+                              worker_state_free_void,
+                              NULL);
+
+  int r = threadpool_register_reply_event(threadpool, NULL);
+
+  tor_assert(r == 0);
 
   set_max_pending_tasks(NULL);
 }
 
+/** Free all resources allocated by cpuworker. */
+void
+cpuworker_free_all(void)
+{
+  threadpool_free(threadpool);
+}
+
 /** Return the number of threads configured for our CPU worker. */
 unsigned int
 cpuworker_get_n_threads(void)
diff --git a/src/core/mainloop/cpuworker.h b/src/core/mainloop/cpuworker.h
index 7821f5612f..7f00f363fe 100644
--- a/src/core/mainloop/cpuworker.h
+++ b/src/core/mainloop/cpuworker.h
@@ -1,7 +1,7 @@
 /* Copyright (c) 2001 Matej Pfajfar.
  * Copyright (c) 2001-2004, Roger Dingledine.
  * Copyright (c) 2004-2006, Roger Dingledine, Nick Mathewson.
- * Copyright (c) 2007-2021, The Tor Project, Inc. */
+ * Copyright (c) 2007-2024, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
 /**
@@ -13,6 +13,7 @@
 #define TOR_CPUWORKER_H
 
 void cpuworker_init(void);
+void cpuworker_free_all(void);
 void cpuworkers_rotate_keyinfo(void);
 
 void cpuworker_consensus_has_changed(const networkstatus_t *ns);
diff --git a/src/lib/evloop/compat_libevent.h b/src/lib/evloop/compat_libevent.h
index 485f85529f..4ffcc0ae93 100644
--- a/src/lib/evloop/compat_libevent.h
+++ b/src/lib/evloop/compat_libevent.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2009-2021, The Tor Project, Inc. */
+/* Copyright (c) 2009-2024, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
 /**
@@ -19,6 +19,7 @@ void configure_libevent_logging(void);
 void suppress_libevent_log_msg(const char *msg);
 
 #define tor_event_new     event_new
+#define tor_event_del     event_del
 #define tor_evtimer_new   evtimer_new
 #define tor_evsignal_new  evsignal_new
 #define tor_evdns_add_server_port(sock, tcp, cb, data) \
diff --git a/src/lib/evloop/workqueue.c b/src/lib/evloop/workqueue.c
index 9a0c02fbd0..20b611f7cb 100644
--- a/src/lib/evloop/workqueue.c
+++ b/src/lib/evloop/workqueue.c
@@ -1,5 +1,5 @@
 
-/* copyright (c) 2013-2015, The Tor Project, Inc. */
+/* copyright (c) 2013-2024, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
 /**
@@ -143,6 +143,8 @@ typedef struct workerthread_t {
 } workerthread_t;
 
 static void queue_reply(replyqueue_t *queue, workqueue_entry_t *work);
+static void workerthread_free(workerthread_t *thread);
+static void replyqueue_free(replyqueue_t *queue);
 
 /** Allocate and return a new workqueue_entry_t, set up to run the function
  * <b>fn</b> in the worker thread, and <b>reply_fn</b> in the main
@@ -355,7 +357,7 @@ workerthread_new(int32_t lower_priority_chance,
     //LCOV_EXCL_START
     tor_assert_nonfatal_unreached();
     log_err(LD_GENERAL, "Can't launch worker thread.");
-    tor_free(thr);
+    workerthread_free(thr);
     return NULL;
     //LCOV_EXCL_STOP
   }
@@ -364,6 +366,15 @@ workerthread_new(int32_t lower_priority_chance,
 }
 
 /**
+ * Free up the resources allocated by a worker thread.
+ */
+static void
+workerthread_free(workerthread_t *thread)
+{
+  tor_free(thread);
+}
+
+/**
  * Queue an item of work for a thread in a thread pool.  The function
  * <b>fn</b> will be run in a worker thread, and will receive as arguments the
  * thread's state object, and the provided object <b>arg</b>. It must return
@@ -566,7 +577,7 @@ threadpool_new(int n_threads,
     tor_assert_nonfatal_unreached();
     tor_cond_uninit(&pool->condition);
     tor_mutex_uninit(&pool->lock);
-    tor_free(pool);
+    threadpool_free(pool);
     return NULL;
     //LCOV_EXCL_STOP
   }
@@ -574,6 +585,39 @@ threadpool_new(int n_threads,
   return pool;
 }
 
+/**
+ * Free up the resources allocated by worker threads, worker thread pool, ...
+ */
+void
+threadpool_free(threadpool_t *pool)
+{
+  if (!pool)
+    return;
+
+  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)
+    pool->free_update_arg_fn(pool->update_args);
+
+  if (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)
+    pool->free_thread_state_fn(pool->new_thread_state_arg);
+
+  tor_free(pool);
+}
+
 /** Return the reply queue associated with a given thread pool. */
 replyqueue_t *
 threadpool_get_replyqueue(threadpool_t *tp)
@@ -593,7 +637,7 @@ replyqueue_new(uint32_t alertsocks_flags)
   rq = tor_malloc_zero(sizeof(replyqueue_t));
   if (alert_sockets_create(&rq->alert, alertsocks_flags) < 0) {
     //LCOV_EXCL_START
-    tor_free(rq);
+    replyqueue_free(rq);
     return NULL;
     //LCOV_EXCL_STOP
   }
@@ -604,6 +648,26 @@ replyqueue_new(uint32_t alertsocks_flags)
   return rq;
 }
 
+/**
+ * Free up the resources allocated by a reply queue.
+ */
+static void
+replyqueue_free(replyqueue_t *queue)
+{
+  if (!queue)
+    return;
+
+  workqueue_entry_t *work;
+
+  while (!TOR_TAILQ_EMPTY(&queue->answers)) {
+    work = TOR_TAILQ_FIRST(&queue->answers);
+    TOR_TAILQ_REMOVE(&queue->answers, work, next_work);
+    workqueue_entry_free(work);
+  }
+
+  tor_free(queue);
+}
+
 /** Internal: Run from the libevent mainloop when there is work to handle in
  * the reply queue handler. */
 static void
diff --git a/src/lib/evloop/workqueue.h b/src/lib/evloop/workqueue.h
index 134fe7434f..9ed504249a 100644
--- a/src/lib/evloop/workqueue.h
+++ b/src/lib/evloop/workqueue.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2013-2021, The Tor Project, Inc. */
+/* Copyright (c) 2013-2024, The Tor Project, Inc. */
 /* See LICENSE for licensing information */
 
 /**
@@ -58,6 +58,7 @@ threadpool_t *threadpool_new(int n_threads,
                              void *(*new_thread_state_fn)(void*),
                              void (*free_thread_state_fn)(void*),
                              void *arg);
+void threadpool_free(threadpool_t *tp);
 replyqueue_t *threadpool_get_replyqueue(threadpool_t *tp);
 
 replyqueue_t *replyqueue_new(uint32_t alertsocks_flags);