summaryrefslogtreecommitdiff
path: root/libguile/whippet/src/parallel-tracer.h
diff options
context:
space:
mode:
Diffstat (limited to 'libguile/whippet/src/parallel-tracer.h')
-rw-r--r--libguile/whippet/src/parallel-tracer.h433
1 files changed, 433 insertions, 0 deletions
diff --git a/libguile/whippet/src/parallel-tracer.h b/libguile/whippet/src/parallel-tracer.h
new file mode 100644
index 000000000..db8afae8a
--- /dev/null
+++ b/libguile/whippet/src/parallel-tracer.h
@@ -0,0 +1,433 @@
+#ifndef PARALLEL_TRACER_H
+#define PARALLEL_TRACER_H
+
+#include <pthread.h>
+#include <stdatomic.h>
+#include <sys/mman.h>
+#include <unistd.h>
+
+#include "assert.h"
+#include "debug.h"
+#include "gc-inline.h"
+#include "gc-tracepoint.h"
+#include "local-worklist.h"
+#include "root-worklist.h"
+#include "shared-worklist.h"
+#include "spin.h"
+#include "tracer.h"
+
+#ifdef VERBOSE_LOGGING
+#define LOG(...) fprintf (stderr, "LOG: " __VA_ARGS__)
+#else
+#define LOG(...) do { } while (0)
+#endif
+
+enum trace_worker_state {
+ TRACE_WORKER_STOPPED,
+ TRACE_WORKER_IDLE,
+ TRACE_WORKER_TRACING,
+ TRACE_WORKER_STOPPING,
+ TRACE_WORKER_DEAD
+};
+
+struct gc_heap;
+struct gc_trace_worker {
+ struct gc_heap *heap;
+ struct gc_tracer *tracer;
+ size_t id;
+ size_t steal_id;
+ pthread_t thread;
+ enum trace_worker_state state;
+ pthread_mutex_t lock;
+ struct shared_worklist shared;
+ struct local_worklist local;
+ struct gc_trace_worker_data *data;
+};
+
+static inline struct gc_trace_worker_data*
+gc_trace_worker_data(struct gc_trace_worker *worker) {
+ return worker->data;
+}
+
+#define TRACE_WORKERS_MAX_COUNT 8
+
+struct gc_tracer {
+ struct gc_heap *heap;
+ atomic_size_t active_tracers;
+ size_t worker_count;
+ long epoch;
+ pthread_mutex_t lock;
+ pthread_cond_t cond;
+ int trace_roots_only;
+ struct root_worklist roots;
+ struct gc_trace_worker workers[TRACE_WORKERS_MAX_COUNT];
+};
+
+static int
+trace_worker_init(struct gc_trace_worker *worker, struct gc_heap *heap,
+ struct gc_tracer *tracer, size_t id) {
+ worker->heap = heap;
+ worker->tracer = tracer;
+ worker->id = id;
+ worker->steal_id = 0;
+ worker->thread = 0;
+ worker->state = TRACE_WORKER_STOPPED;
+ pthread_mutex_init(&worker->lock, NULL);
+ worker->data = NULL;
+ local_worklist_init(&worker->local);
+ return shared_worklist_init(&worker->shared);
+}
+
+static void trace_worker_trace(struct gc_trace_worker *worker);
+
+static void*
+trace_worker_thread(void *data) {
+ struct gc_trace_worker *worker = data;
+ struct gc_tracer *tracer = worker->tracer;
+ long trace_epoch = 0;
+
+ pthread_mutex_lock(&worker->lock);
+ while (1) {
+ long epoch = atomic_load_explicit(&tracer->epoch, memory_order_acquire);
+ if (trace_epoch != epoch) {
+ trace_epoch = epoch;
+ trace_worker_trace(worker);
+ }
+ pthread_cond_wait(&tracer->cond, &worker->lock);
+ }
+ return NULL;
+}
+
+static int
+trace_worker_spawn(struct gc_trace_worker *worker) {
+ if (pthread_create(&worker->thread, NULL, trace_worker_thread, worker)) {
+ perror("spawning tracer thread failed");
+ return 0;
+ }
+
+ return 1;
+}
+
+static int
+gc_tracer_init(struct gc_tracer *tracer, struct gc_heap *heap,
+ size_t parallelism) {
+ tracer->heap = heap;
+ atomic_init(&tracer->active_tracers, 0);
+ tracer->epoch = 0;
+ tracer->trace_roots_only = 0;
+ pthread_mutex_init(&tracer->lock, NULL);
+ pthread_cond_init(&tracer->cond, NULL);
+ root_worklist_init(&tracer->roots);
+ size_t desired_worker_count = parallelism;
+ ASSERT(desired_worker_count);
+ if (desired_worker_count > TRACE_WORKERS_MAX_COUNT)
+ desired_worker_count = TRACE_WORKERS_MAX_COUNT;
+ if (!trace_worker_init(&tracer->workers[0], heap, tracer, 0))
+ return 0;
+ tracer->worker_count++;
+ for (size_t i = 1; i < desired_worker_count; i++) {
+ if (!trace_worker_init(&tracer->workers[i], heap, tracer, i))
+ break;
+ pthread_mutex_lock(&tracer->workers[i].lock);
+ if (trace_worker_spawn(&tracer->workers[i]))
+ tracer->worker_count++;
+ else
+ break;
+ }
+ return 1;
+}
+
+static void gc_tracer_prepare(struct gc_tracer *tracer) {
+ for (size_t i = 0; i < tracer->worker_count; i++)
+ tracer->workers[i].steal_id = (i + 1) % tracer->worker_count;
+}
+static void gc_tracer_release(struct gc_tracer *tracer) {
+ for (size_t i = 0; i < tracer->worker_count; i++)
+ shared_worklist_release(&tracer->workers[i].shared);
+}
+
+static inline void
+gc_tracer_add_root(struct gc_tracer *tracer, struct gc_root root) {
+ root_worklist_push(&tracer->roots, root);
+}
+
+static inline void
+tracer_unpark_all_workers(struct gc_tracer *tracer) {
+ long old_epoch =
+ atomic_fetch_add_explicit(&tracer->epoch, 1, memory_order_acq_rel);
+ long epoch = old_epoch + 1;
+ DEBUG("starting trace; %zu workers; epoch=%ld\n", tracer->worker_count,
+ epoch);
+ GC_TRACEPOINT(trace_unpark_all);
+ pthread_cond_broadcast(&tracer->cond);
+}
+
+static inline void
+tracer_maybe_unpark_workers(struct gc_tracer *tracer) {
+ size_t active =
+ atomic_load_explicit(&tracer->active_tracers, memory_order_acquire);
+ if (active < tracer->worker_count)
+ tracer_unpark_all_workers(tracer);
+}
+
+static inline void
+tracer_share(struct gc_trace_worker *worker) {
+ LOG("tracer #%zu: sharing\n", worker->id);
+ GC_TRACEPOINT(trace_share);
+ size_t to_share = LOCAL_WORKLIST_SHARE_AMOUNT;
+ while (to_share) {
+ struct gc_ref *objv;
+ size_t count = local_worklist_pop_many(&worker->local, &objv, to_share);
+ shared_worklist_push_many(&worker->shared, objv, count);
+ to_share -= count;
+ }
+ tracer_maybe_unpark_workers(worker->tracer);
+}
+
+static inline void
+gc_trace_worker_enqueue(struct gc_trace_worker *worker, struct gc_ref ref) {
+ ASSERT(gc_ref_is_heap_object(ref));
+ if (local_worklist_full(&worker->local))
+ tracer_share(worker);
+ local_worklist_push(&worker->local, ref);
+}
+
+static struct gc_ref
+tracer_steal_from_worker(struct gc_tracer *tracer, size_t id) {
+ ASSERT(id < tracer->worker_count);
+ return shared_worklist_steal(&tracer->workers[id].shared);
+}
+
+static int
+tracer_can_steal_from_worker(struct gc_tracer *tracer, size_t id) {
+ ASSERT(id < tracer->worker_count);
+ return shared_worklist_can_steal(&tracer->workers[id].shared);
+}
+
+static struct gc_ref
+trace_worker_steal_from_any(struct gc_trace_worker *worker,
+ struct gc_tracer *tracer) {
+ for (size_t i = 0; i < tracer->worker_count; i++) {
+ LOG("tracer #%zu: stealing from #%zu\n", worker->id, worker->steal_id);
+ struct gc_ref obj = tracer_steal_from_worker(tracer, worker->steal_id);
+ if (!gc_ref_is_null(obj)) {
+ LOG("tracer #%zu: stealing got %p\n", worker->id,
+ gc_ref_heap_object(obj));
+ return obj;
+ }
+ worker->steal_id = (worker->steal_id + 1) % tracer->worker_count;
+ }
+ LOG("tracer #%zu: failed to steal\n", worker->id);
+ return gc_ref_null();
+}
+
+static int
+trace_worker_can_steal_from_any(struct gc_trace_worker *worker,
+ struct gc_tracer *tracer) {
+ LOG("tracer #%zu: checking if any worker has tasks\n", worker->id);
+ for (size_t i = 0; i < tracer->worker_count; i++) {
+ int res = tracer_can_steal_from_worker(tracer, worker->steal_id);
+ if (res) {
+ LOG("tracer #%zu: worker #%zu has tasks!\n", worker->id,
+ worker->steal_id);
+ return 1;
+ }
+ worker->steal_id = (worker->steal_id + 1) % tracer->worker_count;
+ }
+ LOG("tracer #%zu: nothing to steal\n", worker->id);
+ return 0;
+}
+
+static size_t
+trace_worker_should_continue(struct gc_trace_worker *worker, size_t spin_count) {
+ // Helper workers should park themselves immediately if they have no work.
+ if (worker->id != 0)
+ return 0;
+
+ struct gc_tracer *tracer = worker->tracer;
+
+ if (atomic_load_explicit(&tracer->active_tracers, memory_order_acquire) != 1) {
+ LOG("checking for termination: tracers active, spinning #%zu\n", spin_count);
+ yield_for_spin(spin_count);
+ return 1;
+ }
+
+ // All trace workers have exited except us, the main worker. We are
+ // probably done, but we need to synchronize to be sure that there is no
+ // work pending, for example if a worker had a spurious wakeup. Skip
+ // worker 0 (the main worker).
+
+ GC_TRACEPOINT(trace_check_termination_begin);
+ size_t locked = 1;
+ while (locked < tracer->worker_count) {
+ if (pthread_mutex_trylock(&tracer->workers[locked].lock) == 0)
+ locked++;
+ else
+ break;
+ }
+ int done = (locked == tracer->worker_count) &&
+ !trace_worker_can_steal_from_any(worker, tracer);
+ GC_TRACEPOINT(trace_check_termination_end);
+
+ if (done)
+ return 0;
+ while (locked > 1)
+ pthread_mutex_unlock(&tracer->workers[--locked].lock);
+
+ LOG("checking for termination: failed to lock, spinning #%zu\n", spin_count);
+ yield_for_spin(spin_count);
+ return 1;
+}
+
+static struct gc_ref
+trace_worker_steal(struct gc_trace_worker *worker) {
+ struct gc_tracer *tracer = worker->tracer;
+
+ // It could be that the worker's local trace queue has simply
+ // overflowed. In that case avoid contention by trying to pop
+ // something from the worker's own queue.
+ {
+ LOG("tracer #%zu: trying to pop worker's own deque\n", worker->id);
+ struct gc_ref obj = shared_worklist_try_pop(&worker->shared);
+ if (!gc_ref_is_null(obj))
+ return obj;
+ }
+
+ GC_TRACEPOINT(trace_steal);
+ LOG("tracer #%zu: trying to steal\n", worker->id);
+ struct gc_ref obj = trace_worker_steal_from_any(worker, tracer);
+ if (!gc_ref_is_null(obj))
+ return obj;
+
+ return gc_ref_null();
+}
+
+static void
+trace_with_data(struct gc_tracer *tracer,
+ struct gc_heap *heap,
+ struct gc_trace_worker *worker,
+ struct gc_trace_worker_data *data) {
+ atomic_fetch_add_explicit(&tracer->active_tracers, 1, memory_order_acq_rel);
+ worker->data = data;
+
+ LOG("tracer #%zu: running trace loop\n", worker->id);
+
+ {
+ LOG("tracer #%zu: tracing roots\n", worker->id);
+ size_t n = 0;
+ do {
+ struct gc_root root = root_worklist_pop(&tracer->roots);
+ if (root.kind == GC_ROOT_KIND_NONE)
+ break;
+ trace_root(root, heap, worker);
+ n++;
+ } while (1);
+
+ LOG("tracer #%zu: done tracing roots, %zu roots traced\n", worker->id, n);
+ }
+
+ if (tracer->trace_roots_only) {
+ // Unlike the full trace where work is generated during the trace, a
+ // roots-only trace consumes work monotonically; any object enqueued as a
+ // result of marking roots isn't ours to deal with. However we do need to
+ // synchronize with remote workers to ensure they have completed their
+ // work items.
+ if (worker->id == 0) {
+ for (size_t i = 1; i < tracer->worker_count; i++)
+ pthread_mutex_lock(&tracer->workers[i].lock);
+ }
+ } else {
+ LOG("tracer #%zu: tracing objects\n", worker->id);
+ GC_TRACEPOINT(trace_objects_begin);
+ size_t n = 0;
+ size_t spin_count = 0;
+ do {
+ while (1) {
+ struct gc_ref ref;
+ if (!local_worklist_empty(&worker->local)) {
+ ref = local_worklist_pop(&worker->local);
+ } else {
+ ref = trace_worker_steal(worker);
+ if (gc_ref_is_null(ref))
+ break;
+ }
+ trace_one(ref, heap, worker);
+ n++;
+ }
+ } while (trace_worker_should_continue(worker, spin_count++));
+ GC_TRACEPOINT(trace_objects_end);
+
+ LOG("tracer #%zu: done tracing, %zu objects traced\n", worker->id, n);
+ }
+
+ worker->data = NULL;
+ atomic_fetch_sub_explicit(&tracer->active_tracers, 1, memory_order_acq_rel);
+}
+
+static void
+trace_worker_trace(struct gc_trace_worker *worker) {
+ GC_TRACEPOINT(trace_worker_begin);
+ gc_trace_worker_call_with_data(trace_with_data, worker->tracer,
+ worker->heap, worker);
+ GC_TRACEPOINT(trace_worker_end);
+}
+
+static inline int
+gc_tracer_should_parallelize(struct gc_tracer *tracer) {
+ if (root_worklist_size(&tracer->roots) > 1)
+ return 1;
+
+ if (tracer->trace_roots_only)
+ return 0;
+
+ size_t nonempty_worklists = 0;
+ ssize_t parallel_threshold =
+ LOCAL_WORKLIST_SIZE - LOCAL_WORKLIST_SHARE_AMOUNT;
+ for (size_t i = 0; i < tracer->worker_count; i++) {
+ ssize_t size = shared_worklist_size(&tracer->workers[i].shared);
+ if (!size)
+ continue;
+ nonempty_worklists++;
+ if (nonempty_worklists > 1)
+ return 1;
+ if (size >= parallel_threshold)
+ return 1;
+ }
+ return 0;
+}
+
+static inline void
+gc_tracer_trace(struct gc_tracer *tracer) {
+ LOG("starting trace; %zu workers\n", tracer->worker_count);
+
+ for (int i = 1; i < tracer->worker_count; i++)
+ pthread_mutex_unlock(&tracer->workers[i].lock);
+
+ if (gc_tracer_should_parallelize(tracer)) {
+ LOG("waking workers\n");
+ tracer_unpark_all_workers(tracer);
+ } else {
+ LOG("starting in local-only mode\n");
+ }
+
+ trace_worker_trace(&tracer->workers[0]);
+ root_worklist_reset(&tracer->roots);
+
+ LOG("trace finished\n");
+}
+
+static inline void
+gc_tracer_trace_roots(struct gc_tracer *tracer) {
+ LOG("starting roots-only trace\n");
+
+ GC_TRACEPOINT(trace_roots_begin);
+ tracer->trace_roots_only = 1;
+ gc_tracer_trace(tracer);
+ tracer->trace_roots_only = 0;
+ GC_TRACEPOINT(trace_roots_end);
+
+ GC_ASSERT_EQ(atomic_load(&tracer->active_tracers), 0);
+ LOG("roots-only trace finished\n");
+}
+
+#endif // PARALLEL_TRACER_H