From e86e07e26f8998237408d31216195e5cae79c246 Mon Sep 17 00:00:00 2001 From: Thomas Lively Date: Fri, 24 Feb 2023 10:00:09 -0800 Subject: [PATCH] [Proxying][NFC] Move deferred cleanup logic to `em_task_queue` We previously had a scheme to defer freeing of `em_proyxing_queue`s until there were no more outstanding references to their `em_task_queues` sitting in message queues. Since the notification messages only ever contain references to `em_task_queue`s and not the `em_proxying_queue`s, move the deferred cleanup logic to the `em_task_queue` layer from the `em_proxying_queue` layer. This slightly simplifies the code and is a cleaner separation of concerns. This is NFC as far as users are concerned, and the code is moved with only two changes to internal behavior. First, we now use a trylock when culling zombies to avoid blocking when multiple threads are creating `em_task_queue`s at the same time. Second, we now enqueue zombie queues at the tail of the zombie list instead of the head because FIFO behavior seems fairer. The test for the zombie culling behavior is necessarily less precise now because it the objects with deferred cleanup are no longer directly user-visible. The test still does its job, though. --- system/lib/pthread/em_task_queue.c | 97 +++++++++++- system/lib/pthread/em_task_queue.h | 11 +- system/lib/pthread/proxying.c | 139 +++--------------- test/pthread/test_pthread_proxying_refcount.c | 49 +++--- 4 files changed, 138 insertions(+), 158 deletions(-) diff --git a/system/lib/pthread/em_task_queue.c b/system/lib/pthread/em_task_queue.c index f5620fc60cdda..7254089b7178a 100644 --- a/system/lib/pthread/em_task_queue.c +++ b/system/lib/pthread/em_task_queue.c @@ -5,6 +5,7 @@ * found in the LICENSE file. */ +#include #include #include #include @@ -15,7 +16,82 @@ #define EM_TASK_QUEUE_INITIAL_CAPACITY 128 +// Task Queue Lifetime Management +// ------------------------------- +// +// When tasks are added to a task queue, the Worker running the target thread +// receives an event that will cause it to execute the queue when it next +// returns to its event loop. In some cases the queue will already have been +// executed before then, but the event is still received and the queue is still +// executed. These events contain references to the queue so that the target +// thread will know which queue to execute. +// +// To avoid use-after-free bugs, we cannot free a task queue immediately when +// `em_task_queue_destroy` is called; instead, we must defer freeing the queue +// until all of its outstanding notifications have been processed. We defer +// freeing the queue using an atomic flag. Each time a notification containing a +// reference to a task queue is generated, we set the flag on that task queue. +// Each time that task queue is processed, we clear the flag as long as another +// notification for the queue has not been generated in the mean time. The +// proxying queue can only be freed once `em_task_queue_destroy` has been called +// and its notification flag has been cleared. +// +// But an extra complication is that the target thread may have died by the time +// it gets back to its event loop to process its notifications. In that case the +// thread's Worker will still receive a notification and have to clear the +// notification flag without a live runtime. Without a live runtime, there is no +// stack, so the worker cannot safely free the queue at this point even if the +// notification flag is cleared. We need a separate thread with a live runtime +// to perform the free. +// +// To ensure that queues are eventually freed, we place destroyed queues in a +// global "zombie list" where they wait for their notification flags to be +// cleared. The zombie list is scanned and zombie queues without outstanding +// notifications are freed whenever a new queue is constructed. In principle the +// zombie list could be scanned at any time, but the queue constructor is a nice +// place to do it because scanning there is sufficient to keep the number of +// zombie queues from growing without bound; creating a new zombie ultimately +// requires creating a new queue. +// +// ------------------------------- + +// The head of the zombie list. Its mutex protects access to the list and its +// other fields are not used. +static em_task_queue zombie_list_head = {.mutex = PTHREAD_MUTEX_INITIALIZER, + .zombie_prev = &zombie_list_head, + .zombie_next = &zombie_list_head}; + +static void em_task_queue_free(em_task_queue* queue) { + pthread_mutex_destroy(&queue->mutex); + free(queue->tasks); + free(queue); +} + +static void cull_zombies() { + if (pthread_mutex_trylock(&zombie_list_head.mutex) != 0) { + // Some other thread is already culling. In principle there may be new + // cullable zombies after it finishes, but it's not worth waiting to find + // out. + return; + } + em_task_queue* curr = zombie_list_head.zombie_next; + while (curr != &zombie_list_head) { + em_task_queue* next = curr->zombie_next; + if (curr->notification == NOTIFICATION_NONE) { + // Remove the zombie from the list and free it. + curr->zombie_prev->zombie_next = curr->zombie_next; + curr->zombie_next->zombie_prev = curr->zombie_prev; + em_task_queue_free(curr); + } + curr = next; + } + pthread_mutex_unlock(&zombie_list_head.mutex); +} + em_task_queue* em_task_queue_create(pthread_t thread) { + // Free any queue that has been destroyed and is safe to free. + cull_zombies(); + em_task_queue* queue = malloc(sizeof(em_task_queue)); if (queue == NULL) { return NULL; @@ -32,14 +108,27 @@ em_task_queue* em_task_queue_create(pthread_t thread) { .tasks = tasks, .capacity = EM_TASK_QUEUE_INITIAL_CAPACITY, .head = 0, - .tail = 0}; + .tail = 0, + .zombie_prev = NULL, + .zombie_next = NULL}; return queue; } void em_task_queue_destroy(em_task_queue* queue) { - pthread_mutex_destroy(&queue->mutex); - free(queue->tasks); - free(queue); + assert(queue->zombie_next == NULL && queue->zombie_prev == NULL); + if (queue->notification == NOTIFICATION_NONE) { + // No outstanding references to the queue, so we can go ahead and free it. + em_task_queue_free(queue); + return; + } + // Otherwise add the queue to the zombie list so that it will eventually be + // freed safely. + pthread_mutex_lock(&zombie_list_head.mutex); + queue->zombie_next = &zombie_list_head; + queue->zombie_prev = zombie_list_head.zombie_prev; + queue->zombie_next->zombie_prev = queue; + queue->zombie_prev->zombie_next = queue; + pthread_mutex_unlock(&zombie_list_head.mutex); } // Not thread safe. Returns 1 on success and 0 on failure. diff --git a/system/lib/pthread/em_task_queue.h b/system/lib/pthread/em_task_queue.h index f23bda7e74291..bb1b6d5fc605e 100644 --- a/system/lib/pthread/em_task_queue.h +++ b/system/lib/pthread/em_task_queue.h @@ -17,7 +17,9 @@ typedef struct task { void* arg; } task; -// A task queue holding tasks to be processed by a particular thread. +// A task queue holding tasks to be processed by a particular thread. The only +// "public" field is `notification`. All other fields should be considered +// private implementation details. typedef struct em_task_queue { // Flag encoding the state of postMessage notifications for this task queue. // Accessed directly from JS, so must be the first member. @@ -30,8 +32,7 @@ typedef struct em_task_queue { // Recursion guard. Only accessed on the target thread, so there's no need to // hold the lock when accessing it. TODO: We disallow recursive processing // because that's what the old proxying API does, so it is safer to start with - // the same behavior. Experiment with relaxing this restriction once the old - // API uses these queues as well. + // the same behavior. Experiment with relaxing this restriction. int processing; // Ring buffer of tasks of size `capacity`. New tasks are enqueued at // `tail` and dequeued at `head`. @@ -39,6 +40,10 @@ typedef struct em_task_queue { int capacity; int head; int tail; + // Doubly linked list pointers for the zombie list. See em_task_queue.c for + // details. + struct em_task_queue* zombie_prev; + struct em_task_queue* zombie_next; } em_task_queue; em_task_queue* em_task_queue_create(pthread_t thread); diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 6fe29ee7e64e7..b1e9e234429bd 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -15,51 +15,6 @@ #include "em_task_queue.h" #include "proxying_notification_state.h" -// Proxy Queue Lifetime Management -// ------------------------------- -// -// Proxied tasks are executed either when the user manually calls -// `emscripten_proxy_execute_queue` on the target thread or when the target -// thread returns to the event loop. The queue does not know which execution -// path will be used ahead of time when the work is proxied, so it must -// conservatively send a message to the target thread's event loop in case the -// user expects the event loop to drive the execution. These notifications -// contain references to the queue that will be dereferenced when the target -// thread returns to its event loop and receives the notification, even if the -// user manages the execution of the queue themselves. -// -// To avoid use-after-free bugs, we cannot free a queue immediately when a user -// calls `em_proxying_queue_destroy`; instead, we have to defer freeing the -// queue until all of its outstanding notifications have been processed. We -// defer freeing the queue using a reference counting scheme. Each time a -// notification containing a reference to the a thread-local task queue is -// generated, we set a flag on that task queue. Each time that task queue is -// processed, we clear the flag. The proxying queue can only be freed once -// `em_proxying_queue_destroy` has been called and the notification flags on -// each of its task queues have been cleared. -// -// But an extra complication is that the target thread may have died by the time -// it gets back to its event loop to process its notifications. This can happen -// when a user proxies some work to a thread, then calls -// `emscripten_proxy_execute_queue` on that thread, then destroys the queue and -// exits the thread. In that situation no work will be dropped, but the thread's -// worker will still receive a notification and have to clear the notification -// flag without a live runtime. Without a live runtime, there is no stack, so -// the worker cannot safely free the queue at this point even if the refcount -// goes to zero. We need a separate thread with a live runtime to perform the -// free. -// -// To ensure that queues are eventually freed, we place destroyed queues in a -// global "zombie list" where they wait for their notification flags to be -// cleared. The zombie list is scanned whenever a new queue is constructed and -// any of the zombie queues without outstanding notifications are freed. In -// principle the zombie list could be scanned at any time, but the queue -// constructor is a nice place to do it because scanning there is sufficient to -// keep the number of zombie queues from growing without bound; creating a new -// zombie ultimately requires creating a new queue. -// -// ------------------------------- - struct em_proxying_queue { // Protects all accesses to em_task_queues, size, and capacity. pthread_mutex_t mutex; @@ -67,103 +22,45 @@ struct em_proxying_queue { em_task_queue** task_queues; int size; int capacity; - // Doubly linked list pointers for the zombie list. - em_proxying_queue* zombie_prev; - em_proxying_queue* zombie_next; }; // The system proxying queue. -static em_proxying_queue system_proxying_queue = {.mutex = - PTHREAD_MUTEX_INITIALIZER, - .task_queues = NULL, - .size = 0, - .capacity = 0, - .zombie_prev = NULL, - .zombie_next = NULL}; +static em_proxying_queue system_proxying_queue = { + .mutex = PTHREAD_MUTEX_INITIALIZER, + .task_queues = NULL, + .size = 0, + .capacity = 0, +}; em_proxying_queue* emscripten_proxy_get_system_queue(void) { return &system_proxying_queue; } -// The head of the zombie list. Its mutex protects access to the list and its -// other fields are not used. -static em_proxying_queue zombie_list_head = {.mutex = PTHREAD_MUTEX_INITIALIZER, - .zombie_prev = &zombie_list_head, - .zombie_next = &zombie_list_head}; - -static void em_proxying_queue_free(em_proxying_queue* q) { - pthread_mutex_destroy(&q->mutex); - for (int i = 0; i < q->size; i++) { - em_task_queue_destroy(q->task_queues[i]); - } - free(q->task_queues); - free(q); -} - -// Does not lock `q` because it should only be called after `q` has been -// destroyed when it would be UB for new work to come in and race to generate a -// new notification. -static int has_notification(em_proxying_queue* q) { - for (int i = 0; i < q->size; i++) { - if (q->task_queues[i]->notification != NOTIFICATION_NONE) { - return 1; - } - } - return 0; -} - -static void cull_zombies() { - pthread_mutex_lock(&zombie_list_head.mutex); - em_proxying_queue* curr = zombie_list_head.zombie_next; - while (curr != &zombie_list_head) { - em_proxying_queue* next = curr->zombie_next; - if (!has_notification(curr)) { - // Remove the zombie from the list and free it. - curr->zombie_prev->zombie_next = curr->zombie_next; - curr->zombie_next->zombie_prev = curr->zombie_prev; - em_proxying_queue_free(curr); - } - curr = next; - } - pthread_mutex_unlock(&zombie_list_head.mutex); -} - em_proxying_queue* em_proxying_queue_create(void) { - // Free any queue that has been destroyed and is safe to free. - cull_zombies(); - // Allocate the new queue. em_proxying_queue* q = malloc(sizeof(em_proxying_queue)); if (q == NULL) { return NULL; } - *q = (em_proxying_queue){.mutex = PTHREAD_MUTEX_INITIALIZER, - .task_queues = NULL, - .size = 0, - .capacity = 0, - .zombie_prev = NULL, - .zombie_next = NULL}; + *q = (em_proxying_queue){ + .mutex = PTHREAD_MUTEX_INITIALIZER, + .task_queues = NULL, + .size = 0, + .capacity = 0, + }; return q; } void em_proxying_queue_destroy(em_proxying_queue* q) { assert(q != NULL); assert(q != &system_proxying_queue && "cannot destroy system proxying queue"); - assert(!q->zombie_next && !q->zombie_prev && - "double freeing em_proxying_queue!"); - if (!has_notification(q)) { - // No outstanding references to the queue, so we can go ahead and free it. - em_proxying_queue_free(q); - return; + + pthread_mutex_destroy(&q->mutex); + for (int i = 0; i < q->size; i++) { + em_task_queue_destroy(q->task_queues[i]); } - // Otherwise add the queue to the zombie list so that it will eventually be - // freed safely. - pthread_mutex_lock(&zombie_list_head.mutex); - q->zombie_next = zombie_list_head.zombie_next; - q->zombie_prev = &zombie_list_head; - q->zombie_next->zombie_prev = q; - q->zombie_prev->zombie_next = q; - pthread_mutex_unlock(&zombie_list_head.mutex); + free(q->task_queues); + free(q); } // Not thread safe. Returns NULL if there are no tasks for the thread. diff --git a/test/pthread/test_pthread_proxying_refcount.c b/test/pthread/test_pthread_proxying_refcount.c index 51b35925b2123..454628868158a 100644 --- a/test/pthread/test_pthread_proxying_refcount.c +++ b/test/pthread/test_pthread_proxying_refcount.c @@ -11,25 +11,18 @@ #define SANITIZER #endif -// The first two queues will be zombies and the next two will be created just to -// cull the zombies. -em_proxying_queue* queues[4]; +// Proxying queues accessed from the worker thread. +em_proxying_queue* queues[2]; #ifndef SANITIZER // If we are not using sanitizers (which need to use their own allocators), // override free so we can track when queues are actually freed. -int queues_freed[4] = {}; +_Atomic int frees = 0; void __attribute__((noinline)) free(void* ptr) { - for (int i = 0; i < 4; i++) { - if (ptr && queues[i] == ptr) { - queues_freed[i] = 1; - queues[i] = NULL; - break; - } - } + frees++; emscripten_builtin_free(ptr); } @@ -76,6 +69,8 @@ void* execute_and_free_queue(void* arg) { emscripten_exit_with_live_runtime(); } +void nop(void* arg) {} + int main() { emscripten_console_log("start"); for (int i = 0; i < 2; i++) { @@ -101,35 +96,29 @@ int main() { #ifndef SANITIZER // Our zombies should not have been freed yet. - assert(!queues_freed[0]); - assert(!queues_freed[1]); + int frees_before_cull = frees; #endif // SANITIZER - // Cull the zombies! - queues[2] = em_proxying_queue_create(); + // Cull the zombies! (by forcing a new task queue to be allocated) + em_proxying_queue* culler = em_proxying_queue_create(); + emscripten_proxy_async(culler, pthread_self(), nop, NULL); #ifndef SANITIZER // Now they should be free. - assert(queues_freed[0]); - assert(queues_freed[1]); - assert(!queues_freed[2]); + int frees_after_cull = frees; + assert(frees_after_cull > frees_before_cull); #endif // SANITIZER - em_proxying_queue_destroy(queues[2]); + // If we try again, there should be nothing left to cull. + em_proxying_queue* non_culler = em_proxying_queue_create(); + emscripten_proxy_async(non_culler, pthread_self(), nop, NULL); #ifndef SANITIZER - // The new queue should have been immediately freed. - assert(queues_freed[2]); -#endif // SANITIZER - - // Cull again, but this time there should be nothing to cull. - queues[3] = em_proxying_queue_create(); - em_proxying_queue_destroy(queues[3]); + assert(frees == frees_after_cull); +#endif -#ifndef SANITIZER - // The new queue should have been immediately freed. - assert(queues_freed[3]); -#endif // SANITIZER + em_proxying_queue_destroy(culler); + em_proxying_queue_destroy(non_culler); emscripten_console_log("done"); }