Skip to content

[Proxying][NFC] Move deferred cleanup logic to em_task_queue #18851

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tlively merged 1 commit into from
Feb 28, 2023
Merged

[Proxying][NFC] Move deferred cleanup logic to em_task_queue #18851

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
97 changes: 93 additions & 4 deletions system/lib/pthread/em_task_queue.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
* found in the LICENSE file.
*/

#include <assert.h>
#include <emscripten/threading.h>
#include <stdatomic.h>
#include <stdlib.h>
Expand All @@ -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;
Expand All @@ -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.
Expand Down
11 changes: 8 additions & 3 deletions system/lib/pthread/em_task_queue.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -30,15 +32,18 @@ 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`.
task* tasks;
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);
Expand Down
139 changes: 18 additions & 121 deletions system/lib/pthread/proxying.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,155 +15,52 @@
#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;
// `size` task queue pointers stored in an array of size `capacity`.
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.
Expand Down
Loading