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, ...

3 files changed, 72 insertions, 6 deletions

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);