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