// Ractor implementation #include "ruby/ruby.h" #include "ruby/thread.h" #include "ruby/ractor.h" #include "ruby/re.h" #include "ruby/thread_native.h" #include "vm_core.h" #include "vm_sync.h" #include "ractor_core.h" #include "internal/array.h" #include "internal/class.h" #include "internal/complex.h" #include "internal/cont.h" #include "internal/error.h" #include "internal/gc.h" #include "internal/hash.h" #include "internal/object.h" #include "internal/array.h" #include "internal/string.h" #include "internal/variable.h" #include "eval_intern.h" #include "internal/io.h" #include "internal/marshal.h" #include "internal/ractor.h" #include "internal/rational.h" #include "internal/re.h" #include "internal/struct.h" #include "internal/st.h" #include "internal/thread.h" #include "internal/vm.h" #include "ruby/encoding.h" #include "variable.h" #include "shape.h" #include "yjit.h" #include "zjit.h" VALUE rb_cRactor; static VALUE rb_cRactorSelector; VALUE rb_eRactorUnsafeError; VALUE rb_eRactorIsolationError; static VALUE rb_eRactorError; static VALUE rb_eRactorRemoteError; static VALUE rb_eRactorMovedError; static VALUE rb_eRactorClosedError; static VALUE rb_cRactorMovedObject; static ID id_marshal_dump, id_marshal_load; static ID id_dump, id_load, id_dump_data, id_load_data; static void vm_ractor_blocking_cnt_inc(rb_vm_t *vm, rb_ractor_t *r, const char *file, int line); #if RACTOR_CHECK_MODE > 0 bool rb_ractor_ignore_belonging_flag = false; #endif // Ractor locking static void ASSERT_ractor_unlocking(rb_ractor_t *r) { #if RACTOR_CHECK_MODE > 0 const rb_execution_context_t *ec = rb_current_ec_noinline(); if (ec != NULL && r->sync.locked_by == rb_ractor_self(rb_ec_ractor_ptr(ec))) { rb_bug("recursive ractor locking"); } #endif } static void ASSERT_ractor_locking(rb_ractor_t *r) { #if RACTOR_CHECK_MODE > 0 const rb_execution_context_t *ec = rb_current_ec_noinline(); if (ec != NULL && r->sync.locked_by != rb_ractor_self(rb_ec_ractor_ptr(ec))) { rp(r->sync.locked_by); rb_bug("ractor lock is not acquired."); } #endif } static void ractor_lock(rb_ractor_t *r, const char *file, int line) { RUBY_DEBUG_LOG2(file, line, "locking r:%"PRI_SERIALT_PREFIX"u%s", r->pub.id, rb_current_ractor_raw(false) == r ? " (self)" : ""); ASSERT_ractor_unlocking(r); rb_native_mutex_lock(&r->sync.lock); const rb_execution_context_t *ec = rb_current_ec_noinline(); if (ec) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); VM_ASSERT(!cr->malloc_gc_disabled); cr->malloc_gc_disabled = true; } #if RACTOR_CHECK_MODE > 0 if (ec != NULL) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); r->sync.locked_by = rb_ractor_self(cr); } #endif RUBY_DEBUG_LOG2(file, line, "locked r:%"PRI_SERIALT_PREFIX"u%s", r->pub.id, rb_current_ractor_raw(false) == r ? " (self)" : ""); } static void ractor_lock_self(rb_ractor_t *cr, const char *file, int line) { VM_ASSERT(cr == rb_ec_ractor_ptr(rb_current_ec_noinline())); #if RACTOR_CHECK_MODE > 0 VM_ASSERT(cr->sync.locked_by != cr->pub.self); #endif ractor_lock(cr, file, line); } static void ractor_unlock(rb_ractor_t *r, const char *file, int line) { ASSERT_ractor_locking(r); #if RACTOR_CHECK_MODE > 0 r->sync.locked_by = Qnil; #endif const rb_execution_context_t *ec = rb_current_ec_noinline(); if (ec) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); VM_ASSERT(cr->malloc_gc_disabled); cr->malloc_gc_disabled = false; } rb_native_mutex_unlock(&r->sync.lock); RUBY_DEBUG_LOG2(file, line, "r:%"PRI_SERIALT_PREFIX"u%s", r->pub.id, rb_current_ractor_raw(false) == r ? " (self)" : ""); } static void ractor_unlock_self(rb_ractor_t *cr, const char *file, int line) { VM_ASSERT(cr == rb_ec_ractor_ptr(rb_current_ec_noinline())); #if RACTOR_CHECK_MODE > 0 VM_ASSERT(cr->sync.locked_by == cr->pub.self); #endif ractor_unlock(cr, file, line); } #define RACTOR_LOCK(r) ractor_lock(r, __FILE__, __LINE__) #define RACTOR_UNLOCK(r) ractor_unlock(r, __FILE__, __LINE__) #define RACTOR_LOCK_SELF(r) ractor_lock_self(r, __FILE__, __LINE__) #define RACTOR_UNLOCK_SELF(r) ractor_unlock_self(r, __FILE__, __LINE__) void rb_ractor_lock_self(rb_ractor_t *r) { RACTOR_LOCK_SELF(r); } void rb_ractor_unlock_self(rb_ractor_t *r) { RACTOR_UNLOCK_SELF(r); } // Ractor status static const char * ractor_status_str(enum ractor_status status) { switch (status) { case ractor_created: return "created"; case ractor_running: return "running"; case ractor_blocking: return "blocking"; case ractor_terminated: return "terminated"; } rb_bug("unreachable"); } static void ractor_status_set(rb_ractor_t *r, enum ractor_status status) { RUBY_DEBUG_LOG("r:%"PRI_SERIALT_PREFIX"u [%s]->[%s]", r->pub.id, ractor_status_str(r->status_), ractor_status_str(status)); // check 1 if (r->status_ != ractor_created) { VM_ASSERT(r == GET_RACTOR()); // only self-modification is allowed. ASSERT_vm_locking(); } // check2: transition check. assume it will be vanished on non-debug build. switch (r->status_) { case ractor_created: VM_ASSERT(status == ractor_blocking); break; case ractor_running: VM_ASSERT(status == ractor_blocking|| status == ractor_terminated); break; case ractor_blocking: VM_ASSERT(status == ractor_running); break; case ractor_terminated: rb_bug("unreachable"); break; } r->status_ = status; } static bool ractor_status_p(rb_ractor_t *r, enum ractor_status status) { return rb_ractor_status_p(r, status); } // Ractor data/mark/free static void ractor_local_storage_mark(rb_ractor_t *r); static void ractor_local_storage_free(rb_ractor_t *r); static void ractor_sync_mark(const struct rb_gc_mark_ctx *ctx, rb_ractor_t *r); static void ractor_sync_free(rb_ractor_t *r); static size_t ractor_sync_memsize(const rb_ractor_t *r); static void ractor_sync_init(rb_ractor_t *r); static int mark_targeted_hook_list(st_data_t key, st_data_t value, st_data_t _arg) { rb_hook_list_t *hook_list = (rb_hook_list_t*)value; if (hook_list->type == hook_list_type_targeted_iseq) { rb_gc_mark((VALUE)key); } else { rb_method_definition_t *def = (rb_method_definition_t*)key; RUBY_ASSERT(hook_list->type == hook_list_type_targeted_def); rb_gc_mark(def->body.bmethod.proc); } rb_hook_list_mark(hook_list); return ST_CONTINUE; } static void ractor_mark_thread(const struct rb_gc_mark_ctx *ctx, rb_thread_t *th) { rb_gc_mark_ctx(ctx, th->self); /* A thread's ec lives inside the root fiber struct and is freed with that * fiber's wrapper object, so keep the fiber wrappers alive from here too. */ if (th->root_fiber) { VALUE root_fiber_self = rb_fiberptr_self(th->root_fiber); if (root_fiber_self) rb_gc_mark_ctx(ctx, root_fiber_self); } /* The ec sits inside its fiber, so marking that fiber's wrapper scans the ec * as well. Only when there is no wrapper yet (mid-creation, teardown) does * the ec need marking of its own. */ VALUE ec_fiber_self = (th->ec && th->ec->fiber_ptr) ? rb_fiberptr_self(th->ec->fiber_ptr) : 0; if (ec_fiber_self) { rb_gc_mark_ctx(ctx, ec_fiber_self); } else if (th->ec) { rb_execution_context_mark_ctx(ctx, th->ec); } /* Root the thread's remaining possessions directly as well; thgroup in * particular has no other root. */ rb_thread_mark_owned_roots(ctx, th); } static void ractor_mark_unshareable_parts(const struct rb_gc_mark_ctx *ctx, rb_ractor_t *r) { /* A single VALUE slot written by the owner in one word, so any GC reads it safely. * Its target belongs to another Ractor, so containment makes a foreign marker skip * it. */ rb_gc_mark_ctx(ctx, r->r_stdin); rb_gc_mark_ctx(ctx, r->r_stdout); rb_gc_mark_ctx(ctx, r->r_stderr); rb_gc_mark_ctx(ctx, r->verbose); rb_gc_mark_ctx(ctx, r->debug); // mark the received messages (the structures the owner mutates guard themselves) ractor_sync_mark(ctx, r); /* Structures the owner mutates while running follow. */ rb_hook_list_mark(&r->pub.hooks); if (r->pub.targeted_hooks.num_entries) { st_foreach(&r->pub.targeted_hooks, mark_targeted_hook_list, 0); } if (r->threads.cnt > 0) { rb_thread_t *th = 0; ccan_list_for_each(&r->threads.set, th, lt_node) { VM_ASSERT(th != NULL); ractor_mark_thread(ctx, th); } } /* A thread in the MN termination epilogue has left the set but is still * running on its coroutine stack; it stays a root until its last use. * Read once: the epilogue clears the slot concurrently. The thread is * past rb_fiber_close/thread_cleanup_func by then -- the same state * thread_mark walks whenever a terminated Thread's wrapper is still * referenced, and ractor_mark_thread performs the same marks. */ rb_thread_t *dying_th = RUBY_ATOMIC_PTR_LOAD(r->threads.dying_th); if (dying_th) ractor_mark_thread(ctx, dying_th); ractor_local_storage_mark(r); } static void ractor_mark(void *ptr) { rb_ractor_t *r = (rb_ractor_t *)ptr; const struct rb_gc_mark_ctx ctx = rb_gc_current_mark_ctx(); /* Only the wrapper's direct references: following an unshareable object from the * shareable wrapper would break the shref rule. Unshareable roots are marked by the * root scan (rb_ractor_mark_local_roots); zombie_objspaces covers the terminated. */ rb_gc_mark_ctx(&ctx, r->loc); rb_gc_mark_ctx(&ctx, r->name); /* The default port is shareable, so following it breaks no rule. Other Ractors * still send/value through it after termination, and once a terminated Ractor left * both the set and zombie_objspaces (orphan-merged) this marker is its only cover. */ rb_gc_mark_ctx(&ctx, r->sync.default_port_value); /* A single-objspace impl (mmtk) has no zombie_objspaces and no pin/shref bits, so * the root scan cannot reach a terminated Ractor's queue, in-flight payloads or * join value; and no shref rule forbids following them from the wrapper. */ if (!rb_gc_multi_objspace_p()) { ractor_mark_unshareable_parts(&ctx, r); rb_ractor_mark_terminated_join_value(r); } else if (rb_gc_during_global_gc_p()) { /* The join value is only of use to whoever can still call Ractor#value, which * means holding this wrapper, so mark it as the wrapper's child rather than as a * root. A global GC stops the world and marks every objspace together, which is * what lets the shareable wrapper reach an unshareable value at all. */ rb_ractor_mark_terminated_join_value(r); } } /* Mark the GC roots reachable from Ractor r's C structs. A local GC cannot rely on the * heap Ractor and Thread wrapper objects, which may live in another objspace, so this * Ractor's own possessions are rooted directly from here. */ void rb_ractor_mark_local_roots(const struct rb_gc_mark_ctx *ctx, rb_ractor_t *r) { if (r->postmortem) { /* The final self collection: everything else -- the Thread and Fiber * wrappers, stdio, stack leftovers -- is what it exists to reclaim. * Skipping the walk below cannot drop dying_th: postmortem runs on the * Ractor's last thread, which can only run after any predecessor's * epilogue cleared the slot (under the same scheduler lock). */ VM_ASSERT(RUBY_ATOMIC_PTR_LOAD(r->threads.dying_th) == NULL); rb_ractor_mark_terminated_join_value(r); rb_gc_mark_vm_stack_values((long)r->registered_marks_cnt, r->registered_marks); rb_gc_mark_registered_addrs(r, true); return; } rb_gc_mark_ctx(ctx, r->loc); rb_gc_mark_ctx(ctx, r->name); /* Only the root scan calls this: a local GC for itself, a global GC for the whole * set under the barrier. A terminated Ractor has left the set; zombie_objspaces * covers it instead. */ VM_ASSERT(r == rb_current_ractor_raw(false) || rb_gc_during_global_gc_p()); VM_ASSERT(!rb_ractor_status_p(r, ractor_terminated)); ractor_mark_unshareable_parts(ctx, r); /* This Ractor's rb_gc_register_mark_object pins, treated conservatively: a local GC * marks only its own residents and leaves foreign or shareable entries to their * owner or to the global GC. */ rb_gc_mark_vm_stack_values((long)r->registered_marks_cnt, r->registered_marks); } /* Mark and pin a terminated, unfreed Ractor's return value (legacy); the global GC * calls this via zombie_objspaces. Pinned because compaction does not update C-struct * slots. The default port is covered by the mutual wrapper/port marking instead. */ void rb_ractor_mark_terminated_join_value(rb_ractor_t *r) { VALUE slots[] = { r->sync.legacy, }; rb_gc_mark_vm_stack_values((long)numberof(slots), slots); } /* Move src's rb_gc_register_mark_object pins to dst. Called before merging src's * objspace into dst, so a pinned object never loses its root in between. An absorb can * run during a GC sweep, so plain realloc keeps it from re-entering GC. */ void rb_ractor_absorb_registered_marks(rb_ractor_t *dst, rb_ractor_t *src) { if (src->registered_marks_cnt == 0) return; size_t need = dst->registered_marks_cnt + src->registered_marks_cnt; if (need > dst->registered_marks_capa) { size_t nc = dst->registered_marks_capa ? dst->registered_marks_capa : 64; while (nc < need) nc *= 2; VALUE *p = realloc(dst->registered_marks, nc * sizeof(VALUE)); if (!p) rb_bug("rb_ractor_absorb_registered_marks: out of memory"); dst->registered_marks = p; dst->registered_marks_capa = nc; } MEMCPY(dst->registered_marks + dst->registered_marks_cnt, src->registered_marks, VALUE, src->registered_marks_cnt); dst->registered_marks_cnt = need; src->registered_marks_cnt = 0; } void rb_ractor_absorb_registered_addrs_without_gc(rb_ractor_t *dst, rb_ractor_t *src) { rb_vm_t *vm = GET_VM(); rb_native_mutex_lock(&vm->gc.registered_addrs.lock); if (src->registered_addrs_cnt > 0) { size_t need = dst->registered_addrs_cnt + src->registered_addrs_cnt; if (need > dst->registered_addrs_capa) { size_t nc = dst->registered_addrs_capa ? dst->registered_addrs_capa : 64; while (nc < need) nc *= 2; struct rb_ractor_registered_addr *p = realloc(dst->registered_addrs, nc * sizeof(struct rb_ractor_registered_addr)); if (!p) rb_bug("rb_ractor_absorb_registered_addrs_without_gc: out of memory"); dst->registered_addrs = p; dst->registered_addrs_capa = nc; } MEMCPY(dst->registered_addrs + dst->registered_addrs_cnt, src->registered_addrs, struct rb_ractor_registered_addr, src->registered_addrs_cnt); dst->registered_addrs_cnt = need; src->registered_addrs_cnt = 0; rb_gc_registered_addrs_enroll_without_gc(vm, dst); } rb_gc_registered_addrs_unenroll_without_gc(vm, src); rb_native_mutex_unlock(&vm->gc.registered_addrs.lock); } static int free_targeted_hook_lists(st_data_t key, st_data_t val, st_data_t _arg) { rb_hook_list_t *hook_list = (rb_hook_list_t*)val; rb_hook_list_free(hook_list); return ST_DELETE; } static void free_targeted_hooks(st_table *hooks_tbl) { st_foreach(hooks_tbl, free_targeted_hook_lists, 0); } void rb_thread_sched_destroy(struct rb_thread_sched *); static void ractor_free(void *ptr) { rb_ractor_t *r = (rb_ractor_t *)ptr; RUBY_DEBUG_LOG("free r:%"PRI_SERIALT_PREFIX"u", rb_ractor_id(r)); free_targeted_hooks(&r->pub.targeted_hooks); rb_thread_sched_destroy(&r->threads.sched); rb_native_mutex_destroy(&r->sync.lock); ractor_local_storage_free(r); rb_hook_list_free(&r->pub.hooks); rb_st_free_embedded_table(&r->pub.targeted_hooks); if (r->newobj_cache) { RUBY_ASSERT(r == ruby_single_main_ractor); rb_gc_ractor_cache_free(r->newobj_cache); r->newobj_cache = NULL; } /* Died unjoined and the handle is collected: nobody can inherit it. We are in a * sweep under the global GC barrier, so disown the zombie_objspaces entry and post * the merge to main. main itself only gets here in the free-at-exit walk: leave it * and its objspace to VM destruct. */ if (r->objspace && !r->main_ractor) { rb_gc_objspace_disown(r->objspace); r->objspace = NULL; } ractor_sync_free(r); if (r->in_terminated_set) { rb_native_mutex_lock(&GET_VM()->gc.registered_addrs.lock); ccan_list_del(&r->vmlr_node); r->in_terminated_set = false; rb_native_mutex_unlock(&GET_VM()->gc.registered_addrs.lock); } /* An orphan (unjoined) Ractor hands its rb_gc_register_mark_object pins to main * before its objspace is absorbed; the join path does the same for the joiner. * Both happen before the objspace merge, so no window has unmoved registrations. */ if (!r->main_ractor) { rb_ractor_absorb_registered_marks(GET_VM()->ractor.main_ractor, r); rb_ractor_absorb_registered_addrs_without_gc(GET_VM()->ractor.main_ractor, r); } else { rb_native_mutex_lock(&GET_VM()->gc.registered_addrs.lock); rb_gc_registered_addrs_unenroll_without_gc(GET_VM(), r); rb_native_mutex_unlock(&GET_VM()->gc.registered_addrs.lock); } free(r->registered_marks); r->registered_marks = NULL; r->registered_marks_cnt = r->registered_marks_capa = 0; free(r->registered_addrs); r->registered_addrs = NULL; r->registered_addrs_cnt = r->registered_addrs_capa = 0; if (!r->main_ractor) { SIZED_FREE(r); } } static int targeted_hook_list_memsize_i(st_data_t key, st_data_t val, st_data_t arg) { size_t *size = (size_t *)arg; rb_hook_list_t *hook_list = (rb_hook_list_t *)val; *size += sizeof(rb_hook_list_t) + rb_hook_list_memsize(hook_list); return ST_CONTINUE; } static size_t ractor_memsize(const void *ptr) { rb_ractor_t *r = (rb_ractor_t *)ptr; size_t size = sizeof(rb_ractor_t); size += ractor_sync_memsize(r); size += rb_st_memsize(&r->pub.targeted_hooks) - sizeof(struct st_table); st_foreach(&r->pub.targeted_hooks, targeted_hook_list_memsize_i, (st_data_t)&size); size += rb_hook_list_memsize(&r->pub.hooks); if (r->local_storage) { size += st_memsize(r->local_storage); } if (r->idkey_local_storage) { size += rb_id_table_memsize(r->idkey_local_storage); } size += r->registered_marks_capa * sizeof(VALUE); size += r->registered_addrs_capa * sizeof(struct rb_ractor_registered_addr); return size; } static void ractor_update_references(void *ptr) { /* registered_marks are pinned (marked by rb_gc_mark_vm_stack_values), so * compaction does not need to update them. */ } static const rb_data_type_t ractor_data_type = { "ractor", { ractor_mark, ractor_free, ractor_memsize, ractor_update_references, }, 0, 0, RUBY_TYPED_FREE_IMMEDIATELY /* | RUBY_TYPED_WB_PROTECTED */ }; bool rb_ractor_p(VALUE gv) { if (rb_typeddata_is_kind_of(gv, &ractor_data_type)) { return true; } else { return false; } } static inline rb_ractor_t * RACTOR_PTR(VALUE self) { VM_ASSERT(rb_ractor_p(self)); rb_ractor_t *r = DATA_PTR(self); return r; } #define MAIN_RACTOR_ID 1 static rb_serial_t ractor_last_id = MAIN_RACTOR_ID; #include "ractor_sync.c" // creation/termination /* Ids are never reused, so they must not wrap either: 64 bits, which rules out an * atomic (there is no portable 64-bit one). Serialized by the VM lock, or by the * GVL before there is a second Ractor to take it against -- the same condition * vm_insert_ractor0 asserts. */ static rb_serial_t ractor_next_id(void) { VM_ASSERT(RB_VM_LOCKED_P() || !rb_multi_ractor_p()); return ++ractor_last_id; } static void vm_insert_ractor0(rb_vm_t *vm, rb_ractor_t *r, bool single_ractor_mode) { RUBY_DEBUG_LOG("r:%"PRI_SERIALT_PREFIX"u ractor.cnt:%u++", r->pub.id, vm->ractor.cnt); VM_ASSERT(single_ractor_mode || RB_VM_LOCKED_P()); /* End main's cycle before a second Ractor becomes visible: the collection was * planned for a single-objspace world (a local GC is a whole-world GC there and may * free shareable objects), so it must not straddle the transition. */ if (vm->ractor.cnt == 1) { rb_gc_rest(); } ccan_list_add_tail(&vm->ractor.set, &r->vmlr_node); vm->ractor.cnt++; if (r->newobj_cache) { VM_ASSERT(r == ruby_single_main_ractor); } else { r->newobj_cache = rb_gc_ractor_cache_alloc(r); } } static void cancel_single_ractor_mode(void) { // enable multi-ractor mode RUBY_DEBUG_LOG("enable multi-ractor mode"); ruby_single_main_ractor = NULL; rb_yjit_invalidate_single_ractor(); rb_zjit_invalidate_single_ractor(); ASSERT_vm_unlocking(); rb_funcall(rb_cRactor, rb_intern("_activated"), 0); } static void vm_insert_ractor(rb_vm_t *vm, rb_ractor_t *r) { VM_ASSERT(ractor_status_p(r, ractor_created)); if (rb_multi_ractor_p()) { RB_VM_LOCK(); { vm_insert_ractor0(vm, r, false); vm_ractor_blocking_cnt_inc(vm, r, __FILE__, __LINE__); /* The child is in the set and enumerated on its own now, so drop the cover * through its creator and avoid enumerating it twice. Cleared under the * same VM lock that added it, so no whole-VM walk sees both. */ rb_ractor_t *cur = rb_current_ractor_raw(false); if (cur && cur->creating_child_objspace == r->objspace) { cur->creating_child_objspace = NULL; } } RB_VM_UNLOCK(); } else { if (vm->ractor.cnt == 0) { // main ractor vm_insert_ractor0(vm, r, true); ractor_status_set(r, ractor_blocking); ractor_status_set(r, ractor_running); } else { cancel_single_ractor_mode(); vm_insert_ractor0(vm, r, true); vm_ractor_blocking_cnt_inc(vm, r, __FILE__, __LINE__); /* As in the multi-Ractor branch: the child joined the set, so drop the * creator's cover, or a global GC enumerates the child's objspace twice and * sweeps its live main Thread and root Fiber. */ rb_ractor_t *cur = rb_current_ractor_raw(false); if (cur && cur->creating_child_objspace == r->objspace) { cur->creating_child_objspace = NULL; } } } } static void vm_remove_ractor(rb_vm_t *vm, rb_ractor_t *cr) { VM_ASSERT(ractor_status_p(cr, ractor_running)); VM_ASSERT(vm->ractor.cnt > 1); VM_ASSERT(cr->threads.cnt == 1); RB_VM_LOCK(); { RUBY_DEBUG_LOG("ractor.cnt:%u-- terminate_waiting:%d", vm->ractor.cnt, vm->ractor.sync.terminate_waiting); VM_ASSERT(vm->ractor.cnt > 0); ccan_list_del(&cr->vmlr_node); /* A single-objspace impl has no zombie_objspaces, so nothing roots the * registered_marks of a Ractor that left the set; track it in a separate list * until ractor_free. */ if (!rb_gc_multi_objspace_p()) { rb_native_mutex_lock(&vm->gc.registered_addrs.lock); ccan_list_add(&vm->ractor.terminated_set, &cr->vmlr_node); cr->in_terminated_set = true; rb_native_mutex_unlock(&vm->gc.registered_addrs.lock); } if (vm->ractor.cnt <= 2 && vm->ractor.sync.terminate_waiting) { rb_native_cond_signal(&vm->ractor.sync.terminate_cond); } rb_gc_ractor_cache_free(cr->newobj_cache); cr->newobj_cache = NULL; /* The objspace loses its owning thread: keep it enumerable until inheritance * merges it. Register in zombie_objspaces BEFORE decrementing cnt: other * Ractors read rb_gc_single_objspace_p lock-free, and the other order opens a * cnt==1-no-zombie window where a GC skips shareable pinning and collects * objects (a cc, say) reachable only through this objspace. */ if (cr->objspace) { /* The final self collection already ran (ractor_postmortem_collect), * so the entry measures exactly the pages the joiner will inherit. */ rb_gc_objspace_retire(&cr->objspace); } vm->ractor.cnt--; ractor_status_set(cr, ractor_terminated); } RB_VM_UNLOCK(); } /* The dying thread's final collection of its own objspace, and the capture of what it * must free itself. Called with the GVL still held: a concurrent global GC waits for * this thread's safepoint, so the collection is lock-free like any local GC. */ static void ractor_postmortem_collect(rb_thread_t *th, struct rb_ractor_postmortem_frees *pf) { rb_ractor_t *const cr = th->ractor; VM_ASSERT(cr != GET_VM()->ractor.main_ractor); VM_ASSERT(th->ec != NULL); struct rb_fiber_struct *const fiber = th->ec->fiber_ptr; const bool fiber_wrapped = fiber && rb_fiberptr_self(fiber) != 0; /* With a thread-event hook registered the callbacks may retain any Ractor's Thread * object (rb_internal_thread_event_hook_t), and such references are invisible to * the reduced roots: keep the ordinary retire roots, wrappers survive to absorb. */ cr->postmortem = rb_gc_multi_objspace_p() && !rb_thread_event_hooks_registered_p(); rb_gc_objspace_postmortem_self(); /* Only what this collection provably swept (wrapper gone, struct deferred with an * in-band mark) may be touched after vm_remove_ractor unlinks the Ractor. */ pf->th = (th->self == 0) ? th : NULL; pf->fiber = (fiber_wrapped && rb_fiberptr_self(fiber) == 0) ? fiber : NULL; } void rb_ractor_postmortem_free(const struct rb_ractor_postmortem_frees *pf) { if (pf->fiber == NULL && pf->th == NULL) return; /* The frees below resolve their objspace through the TLS ec, which sits inside * pf->fiber and leads to the retired Ractor: cut the resolution off so they fall * back to the main objspace. (The MN epilogue has already done this.) */ #ifdef RB_THREAD_LOCAL_SPECIFIER rb_current_ec_set(NULL); #else native_tls_set(ruby_current_ec_key, NULL); #endif /* the fiber struct embeds the thread's final ec, so free it first */ if (pf->fiber) rb_fiber_free_body(pf->fiber); if (pf->th) rb_thread_free_body(pf->th); } static VALUE ractor_alloc(VALUE klass) { rb_ractor_t *r; VALUE rv = TypedData_Make_Struct(klass, rb_ractor_t, &ractor_data_type, r); RB_OBJ_SET_SHAREABLE(rv); r->pub.self = rv; r->next_ec_serial = 1; VM_ASSERT(ractor_status_p(r, ractor_created)); return rv; } static rb_ractor_t _main_ractor = { .loc = Qnil, .name = Qnil, .pub.id = MAIN_RACTOR_ID, .pub.self = Qnil, .next_ec_serial = 1, .main_ractor = true, }; rb_ractor_t * rb_ractor_main_alloc(void) { rb_ractor_t *r = &_main_ractor; /* The main Ractor is allocated before its objspace exists, so its newobj cache is * created later in Init_BareVM, once rb_gc_init_objspaces has set r->objspace. */ ruby_single_main_ractor = r; return r; } #if defined(HAVE_WORKING_FORK) // Set up the main Ractor for the VM after fork. // Puts us in "single Ractor mode" void rb_ractor_atfork(rb_vm_t *vm, rb_thread_t *th) { // initialize as a main ractor vm->ractor.cnt = 0; vm->ractor.blocking_cnt = 0; /* Only main survives a fork: the holds of dead Ractors and of critical sections are * gone, leaving main's own disable. */ rb_gc_disable_holders_atfork(); /* Only the main Ractor survives a fork, so drop the creation cover. The set was * just emptied by rb_vm_living_threads_init, and zombie_objspaces still holds the * non-main objspaces that terminate_atfork parked there for the orphan merge. */ th->ractor->creating_child_objspace = NULL; ruby_single_main_ractor = th->ractor; th->ractor->status_ = ractor_created; rb_ractor_living_threads_init(th->ractor); rb_ractor_living_threads_insert(th->ractor, th); VM_ASSERT(vm->ractor.blocking_cnt == 0); VM_ASSERT(vm->ractor.cnt == 1); } void rb_ractor_terminate_atfork(rb_vm_t *vm, rb_ractor_t *r) { rb_gc_ractor_cache_free(r->newobj_cache); r->newobj_cache = NULL; r->status_ = ractor_terminated; // a termination epilogue in the parent did not survive the fork r->threads.dying_th = NULL; if (!rb_gc_multi_objspace_p()) { ccan_list_del(&r->vmlr_node); ccan_list_add(&vm->ractor.terminated_set, &r->vmlr_node); r->in_terminated_set = true; } /* In a forked child every other Ractor is terminated-unjoined, so keep its objspace * enumerable until a join or a global GC merges it. */ if (r->objspace) { rb_gc_objspace_retire(&r->objspace); } ractor_sync_terminate_atfork(vm, r); } #endif void rb_thread_sched_init(struct rb_thread_sched *, bool atfork); void rb_ractor_living_threads_init(rb_ractor_t *r) { ccan_list_head_init(&r->threads.set); r->threads.cnt = 0; r->threads.blocking_cnt = 0; r->threads.terminating = false; // atfork: a sibling's termination epilogue did not survive the fork r->threads.dying_th = NULL; } static void ractor_init(rb_ractor_t *r, VALUE name, VALUE loc) { ractor_sync_init(r); st_init_existing_numtable_with_size(&r->pub.targeted_hooks, 0); r->pub.hooks.type = hook_list_type_ractor_local; // thread management rb_thread_sched_init(&r->threads.sched, false); rb_ractor_living_threads_init(r); // naming if (!NIL_P(name)) { rb_encoding *enc; StringValueCStr(name); enc = rb_enc_get(name); if (!rb_enc_asciicompat(enc)) { rb_raise(rb_eArgError, "ASCII incompatible encoding (%s)", rb_enc_name(enc)); } name = RB_OBJ_SET_SHAREABLE(rb_str_new_frozen(name)); } if (!SPECIAL_CONST_P(loc)) RB_OBJ_SET_SHAREABLE(loc); r->loc = loc; r->name = name; } void rb_ractor_main_setup(rb_vm_t *vm, rb_ractor_t *r, rb_thread_t *th) { VALUE rv = r->pub.self = TypedData_Wrap_Struct(rb_cRactor, &ractor_data_type, r); RB_OBJ_SET_SHAREABLE(r->pub.self); ractor_init(r, Qnil, Qnil); r->threads.main = th; rb_ractor_living_threads_insert(r, th); rb_ractor_setup_default_port(r); RB_GC_GUARD(rv); } static VALUE ractor_create(rb_execution_context_t *ec, VALUE self, VALUE loc, VALUE name, VALUE args, VALUE block) { VALUE rv = ractor_alloc(self); rb_ractor_t *r = RACTOR_PTR(rv); ractor_init(r, name, loc); RB_VM_LOCKING() { r->pub.id = ractor_next_id(); } RUBY_DEBUG_LOG("r:%"PRI_SERIALT_PREFIX"u", r->pub.id); rb_ractor_t *cr = rb_ec_ractor_ptr(ec); r->verbose = cr->verbose; r->debug = cr->debug; /* Every Ractor has an objspace, and it must exist before its thread runs: the * first allocation goes there through rb_gc_get_objspace. */ r->objspace = rb_gc_objspace_alloc(); rb_thread_create_ractor(r, args, block); RB_GC_GUARD(rv); return rv; } #if 0 static VALUE ractor_create_func(VALUE klass, VALUE loc, VALUE name, VALUE args, rb_block_call_func_t func) { VALUE block = rb_proc_new(func, Qnil); return ractor_create(rb_current_ec_noinline(), klass, loc, name, args, block); } #endif static void ractor_atexit(rb_execution_context_t *ec, rb_ractor_t *cr, VALUE result, bool exc) { ractor_notify_exit(ec, cr, result, exc); } /* The dying thread's last work inside its Ractor. The order is the point: a joiner woken * before the collection spins in ractor_value for the whole of it. */ void rb_ractor_postmortem(rb_thread_t *th, struct rb_ractor_postmortem_frees *pf) { ractor_postmortem_collect(th, pf); ractor_send_exit_tokens(th->ec, th->ractor); } void rb_ractor_atexit(rb_execution_context_t *ec, VALUE result) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); ractor_atexit(ec, cr, result, false); } void rb_ractor_atexit_exception(rb_execution_context_t *ec) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); ractor_atexit(ec, cr, ec->errinfo, true); } void rb_ractor_teardown(rb_execution_context_t *ec) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); // sync with rb_ractor_terminate_interrupt_main_thread() RB_VM_LOCKING() { VM_ASSERT(cr->threads.main != NULL); cr->threads.main = NULL; } } void rb_ractor_receive_parameters(rb_execution_context_t *ec, rb_ractor_t *r, int len, VALUE *ptr) { for (int i=0; iractor.main_ractor; } int rb_ractor_living_thread_num(const rb_ractor_t *r) { return r->threads.cnt; } // only for current ractor VALUE rb_ractor_thread_list(void) { rb_ractor_t *r = GET_RACTOR(); rb_thread_t *th = 0; VALUE ary = rb_ary_new(); ccan_list_for_each(&r->threads.set, th, lt_node) { switch (th->status) { case THREAD_RUNNABLE: case THREAD_STOPPED: case THREAD_STOPPED_FOREVER: rb_ary_push(ary, th->self); default: break; } } return ary; } void rb_ractor_living_threads_insert(rb_ractor_t *r, rb_thread_t *th) { VM_ASSERT(th != NULL); RACTOR_LOCK(r); { RUBY_DEBUG_LOG("r(%"PRI_SERIALT_PREFIX"u)->threads.cnt:%d++", r->pub.id, r->threads.cnt); ccan_list_add_tail(&r->threads.set, &th->lt_node); r->threads.cnt++; } RACTOR_UNLOCK(r); // first thread for a ractor if (r->threads.cnt == 1) { VM_ASSERT(ractor_status_p(r, ractor_created)); vm_insert_ractor(th->vm, r); } } static void vm_ractor_blocking_cnt_inc(rb_vm_t *vm, rb_ractor_t *r, const char *file, int line) { ractor_status_set(r, ractor_blocking); RUBY_DEBUG_LOG2(file, line, "vm->ractor.blocking_cnt:%d++", vm->ractor.blocking_cnt); vm->ractor.blocking_cnt++; VM_ASSERT(vm->ractor.blocking_cnt <= vm->ractor.cnt); } void rb_vm_ractor_blocking_cnt_inc(rb_vm_t *vm, rb_ractor_t *cr, const char *file, int line) { ASSERT_vm_locking(); VM_ASSERT(GET_RACTOR() == cr); vm_ractor_blocking_cnt_inc(vm, cr, file, line); } void rb_vm_ractor_blocking_cnt_dec(rb_vm_t *vm, rb_ractor_t *cr, const char *file, int line) { ASSERT_vm_locking(); VM_ASSERT(GET_RACTOR() == cr); RUBY_DEBUG_LOG2(file, line, "vm->ractor.blocking_cnt:%d--", vm->ractor.blocking_cnt); VM_ASSERT(vm->ractor.blocking_cnt > 0); vm->ractor.blocking_cnt--; ractor_status_set(cr, ractor_running); } /* Remove a child that never started (send_parameters failed during creation). The * creator calls this (rb_ractor_living_threads_remove assumes the current Ractor); * leaving the set and disowning the objspace share one VM-lock section, no window. */ void rb_ractor_cancel_creation(rb_ractor_t *r, rb_thread_t *th) { RACTOR_LOCK(r); { ccan_list_del(&th->lt_node); r->threads.cnt--; } RACTOR_UNLOCK(r); RB_VM_LOCK(); { rb_vm_t *vm = th->vm; VM_ASSERT(vm->ractor.cnt > 1); ccan_list_del(&r->vmlr_node); vm->ractor.cnt--; /* Give back the blocking count vm_insert_ractor took at insert time. A child * that never ran has no chance to decrement it, and without this the * blocking_cnt <= cnt invariant breaks on the next insert. */ VM_ASSERT(r->status_ == ractor_blocking); VM_ASSERT(vm->ractor.blocking_cnt > 0); vm->ractor.blocking_cnt--; rb_gc_ractor_cache_free(r->newobj_cache); r->newobj_cache = NULL; if (r->objspace) { rb_gc_objspace_disown(r->objspace); r->objspace = NULL; } r->status_ = ractor_terminated; } RB_VM_UNLOCK(); } void rb_ractor_living_threads_remove(rb_ractor_t *cr, rb_thread_t *th) { VM_ASSERT(cr == GET_RACTOR()); RUBY_DEBUG_LOG("r->threads.cnt:%d--", cr->threads.cnt); if (cr->threads.cnt == 1) { vm_remove_ractor(th->vm, cr); } else { RACTOR_LOCK(cr); { ccan_list_del(&th->lt_node); cr->threads.cnt--; } RACTOR_UNLOCK(cr); } } void rb_ractor_blocking_threads_inc(rb_ractor_t *cr, const char *file, int line) { RUBY_DEBUG_LOG2(file, line, "cr->threads.blocking_cnt:%d++", cr->threads.blocking_cnt); VM_ASSERT(cr->threads.cnt > 0); VM_ASSERT(cr == GET_RACTOR()); cr->threads.blocking_cnt++; } void rb_ractor_blocking_threads_dec(rb_ractor_t *cr, const char *file, int line) { RUBY_DEBUG_LOG2(file, line, "r->threads.blocking_cnt:%d--, r->threads.cnt:%u", cr->threads.blocking_cnt, cr->threads.cnt); VM_ASSERT(cr == GET_RACTOR()); cr->threads.blocking_cnt--; } void rb_ractor_vm_barrier_interrupt_running_thread(rb_ractor_t *r) { VM_ASSERT(r != GET_RACTOR()); ASSERT_ractor_unlocking(r); ASSERT_vm_locking(); RACTOR_LOCK(r); { if (ractor_status_p(r, ractor_running)) { rb_execution_context_t *ec = r->threads.running_ec; if (ec) { RUBY_VM_SET_VM_BARRIER_INTERRUPT(ec); } } } RACTOR_UNLOCK(r); } void rb_ractor_terminate_interrupt_main_thread(rb_ractor_t *r) { VM_ASSERT(r != GET_RACTOR()); ASSERT_ractor_unlocking(r); ASSERT_vm_locking(); rb_thread_t *main_th = r->threads.main; if (main_th) { if (main_th->status != THREAD_KILLED) { RUBY_VM_SET_TERMINATE_INTERRUPT(main_th->ec); rb_threadptr_interrupt(main_th); } else { RUBY_DEBUG_LOG("killed (%p)", (void *)main_th); } } } void rb_thread_terminate_all(rb_thread_t *th); // thread.c static void ractor_terminal_interrupt_all(rb_vm_t *vm) { if (vm->ractor.cnt > 1) { // send terminate notification to all ractors rb_ractor_t *r = 0; ccan_list_for_each(&vm->ractor.set, r, vmlr_node) { if (r != vm->ractor.main_ractor) { RUBY_DEBUG_LOG("r:%"PRI_SERIALT_PREFIX"u", rb_ractor_id(r)); rb_ractor_terminate_interrupt_main_thread(r); } } } } void rb_add_running_thread(rb_thread_t *th); void rb_del_running_thread(rb_thread_t *th); void rb_ractor_terminate_all(void) { rb_vm_t *vm = GET_VM(); rb_ractor_t *cr = vm->ractor.main_ractor; RUBY_DEBUG_LOG("ractor.cnt:%d", (int)vm->ractor.cnt); VM_ASSERT(cr == GET_RACTOR()); // only main-ractor's main-thread should kick it. RB_VM_LOCK(); { ractor_terminal_interrupt_all(vm); // kill all ractors } RB_VM_UNLOCK(); rb_thread_terminate_all(GET_THREAD()); // kill other threads in main-ractor and wait RB_VM_LOCK(); { while (vm->ractor.cnt > 1) { RUBY_DEBUG_LOG("terminate_waiting:%d", vm->ractor.sync.terminate_waiting); vm->ractor.sync.terminate_waiting = true; // wait for 1sec rb_vm_ractor_blocking_cnt_inc(vm, cr, __FILE__, __LINE__); rb_del_running_thread(rb_ec_thread_ptr(cr->threads.running_ec)); rb_ractor_sched_wait_terminate(vm, &vm->ractor.sync.terminate_cond, 1000 /* ms */); while (vm->ractor.sched.barrier_is_waiting) { // A barrier is waiting. Threads relinquish the VM lock before joining the barrier and // since we just acquired the VM lock back, we're blocking other threads from joining it. // We loop until the barrier is over. We can't join this barrier because our thread isn't added to // running_threads until the call below to `rb_add_running_thread`. RB_VM_UNLOCK(); unsigned int lev; RB_VM_LOCK_ENTER_LEV_NB(&lev); } rb_add_running_thread(rb_ec_thread_ptr(cr->threads.running_ec)); rb_vm_ractor_blocking_cnt_dec(vm, cr, __FILE__, __LINE__); ractor_terminal_interrupt_all(vm); } } RB_VM_UNLOCK(); /* Every other Ractor is dead. main inherits all uninherited objspaces, so the * remaining at-exit work (finalizers, IO flush, free-at-exit) sees every object. */ rb_gc_objspace_absorb_all_zombies(); } rb_execution_context_t * rb_vm_main_ractor_ec(rb_vm_t *vm) { /* This code needs to carefully work around two bugs: * - Bug #20016: When M:N threading is enabled, running_ec is NULL if no thread is * actually currently running (as opposed to without M:N threading, when * running_ec will still point to the _last_ thread which ran) * - Bug #20197: If the main thread is sleeping, setting its postponed job * interrupt flag is pointless; it won't look at the flag until it stops sleeping * for some reason. It would be better to set the flag on the running ec, which * will presumably look at it soon. * * Solution: use running_ec if it's set, otherwise fall back to the main thread ec. * This is still susceptible to some rare race conditions (what if the last thread * to run just entered a long-running sleep?), but seems like the best balance of * robustness and complexity. */ rb_execution_context_t *running_ec = vm->ractor.main_ractor->threads.running_ec; if (running_ec) { return running_ec; } return vm->ractor.main_thread->ec; } static VALUE ractor_moved_missing(int argc, VALUE *argv, VALUE self) { rb_raise(rb_eRactorMovedError, "can not send any methods to a moved object"); } /* * Document-class: Ractor::Error * * The parent class of Ractor-related error classes. */ /* * Document-class: Ractor::ClosedError * * Raised when an attempt is made to send a message to a closed port, * or to retrieve a message from a closed and empty port. * Ports may be closed explicitly with Ractor::Port#close * and are closed implicitly when a Ractor terminates. * * port = Ractor::Port.new * port.close * port << "test" # Ractor::ClosedError * port.receive # Ractor::ClosedError * * ClosedError is a descendant of StopIteration, so the closing of a port will break * out of loops without propagating the error. */ /* * Document-class: Ractor::IsolationError * * Raised on attempt to make a Ractor-unshareable object * Ractor-shareable. */ /* * Document-class: Ractor::RemoteError * * Raised on Ractor#join or Ractor#value if there was an uncaught exception in the Ractor. * Its +cause+ will contain the original exception, and +ractor+ is the original ractor * it was raised in. * * r = Ractor.new { raise "Something weird happened" } * * begin * r.value * rescue => e * p e # => # * p e.ractor == r # => true * p e.cause # => # * end * */ /* * Document-class: Ractor::MovedError * * Raised on an attempt to access an object which was moved in Ractor#send or Ractor::Port#send. * * r = Ractor.new { sleep } * * ary = [1, 2, 3] * r.send(ary, move: true) * ary.inspect * # Ractor::MovedError (can not send any methods to a moved object) * */ /* * Document-class: Ractor::MovedObject * * A special object which replaces any value that was moved to another ractor in Ractor#send * or Ractor::Port#send. Any attempt to access the object results in Ractor::MovedError. * * r = Ractor.new { receive } * * ary = [1, 2, 3] * r.send(ary, move: true) * p Ractor::MovedObject === ary * # => true * ary.inspect * # Ractor::MovedError (can not send any methods to a moved object) */ /* * Document-class: Ractor::UnsafeError * * Raised when Ractor-unsafe C-methods is invoked by a non-main Ractor. */ // Main docs are in ractor.rb, but without this clause there are weird artifacts // in their rendering. /* * Document-class: Ractor * */ void Init_Ractor(void) { rb_cRactor = rb_define_class("Ractor", rb_cObject); rb_undef_alloc_func(rb_cRactor); rb_eRactorError = rb_define_class_under(rb_cRactor, "Error", rb_eRuntimeError); rb_eRactorIsolationError = rb_define_class_under(rb_cRactor, "IsolationError", rb_eRactorError); rb_eRactorRemoteError = rb_define_class_under(rb_cRactor, "RemoteError", rb_eRactorError); rb_eRactorMovedError = rb_define_class_under(rb_cRactor, "MovedError", rb_eRactorError); rb_eRactorClosedError = rb_define_class_under(rb_cRactor, "ClosedError", rb_eStopIteration); rb_eRactorUnsafeError = rb_define_class_under(rb_cRactor, "UnsafeError", rb_eRactorError); rb_cRactorMovedObject = rb_define_class_under(rb_cRactor, "MovedObject", rb_cBasicObject); rb_undef_alloc_func(rb_cRactorMovedObject); rb_define_method(rb_cRactorMovedObject, "method_missing", ractor_moved_missing, -1); // override methods defined in BasicObject rb_define_method(rb_cRactorMovedObject, "__send__", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "!", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "==", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "!=", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "__id__", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "equal?", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "instance_eval", ractor_moved_missing, -1); rb_define_method(rb_cRactorMovedObject, "instance_exec", ractor_moved_missing, -1); id_marshal_dump = rb_intern_const("marshal_dump"); id_marshal_load = rb_intern_const("marshal_load"); id_dump = rb_intern_const("_dump"); id_load = rb_intern_const("_load"); id_dump_data = rb_intern_const("_dump_data"); id_load_data = rb_intern_const("_load_data"); Init_RactorPort(); } void rb_ractor_dump(void) { rb_vm_t *vm = GET_VM(); rb_ractor_t *r = 0; ccan_list_for_each(&vm->ractor.set, r, vmlr_node) { if (r != vm->ractor.main_ractor) { fprintf(stderr, "r:%"PRI_SERIALT_PREFIX"u (%s)\n", r->pub.id, ractor_status_str(r->status_)); } } } VALUE rb_ractor_stdin(void) { if (rb_ractor_main_p()) { return rb_stdin; } else { rb_ractor_t *cr = GET_RACTOR(); if (UNLIKELY(cr->r_stdin == 0)) { cr->r_stdin = rb_io_prep_stdin(); } return cr->r_stdin; } } VALUE rb_ractor_stdout(void) { if (rb_ractor_main_p()) { return rb_stdout; } else { rb_ractor_t *cr = GET_RACTOR(); if (UNLIKELY(cr->r_stdout == 0)) { cr->r_stdout = rb_io_prep_stdout(); } return cr->r_stdout; } } VALUE rb_ractor_stderr(void) { if (rb_ractor_main_p()) { return rb_stderr; } else { rb_ractor_t *cr = GET_RACTOR(); if (UNLIKELY(cr->r_stderr == 0)) { cr->r_stderr = rb_io_prep_stderr(); } return cr->r_stderr; } } void rb_ractor_stdin_set(VALUE in) { if (rb_ractor_main_p()) { rb_stdin = in; } else { rb_ractor_t *cr = GET_RACTOR(); RB_OBJ_WRITE(cr->pub.self, &cr->r_stdin, in); } } void rb_ractor_stdout_set(VALUE out) { if (rb_ractor_main_p()) { rb_stdout = out; } else { rb_ractor_t *cr = GET_RACTOR(); RB_OBJ_WRITE(cr->pub.self, &cr->r_stdout, out); } } void rb_ractor_stderr_set(VALUE err) { if (rb_ractor_main_p()) { rb_stderr = err; } else { rb_ractor_t *cr = GET_RACTOR(); RB_OBJ_WRITE(cr->pub.self, &cr->r_stderr, err); } } st_table * rb_ractor_targeted_hooks(rb_ractor_t *cr) { return &cr->pub.targeted_hooks; } static void rb_obj_set_shareable_no_assert(VALUE obj) { /* make_shareable_check_shareable refuses an IO, because the traversal cannot reach * the VALUE members inside its fptr. */ VM_ASSERT(!RB_TYPE_P(obj, T_FILE)); FL_SET_RAW(obj, FL_SHAREABLE); rb_gc_obj_became_shareable(obj); /* Ivars on a shareable object would be mutable shared state, so freeze them * (not obj itself). A T_IMEMO has no shape id to transition. */ bool froze_ivars = false; if (!RB_OBJ_FROZEN_RAW(obj) && !RB_TYPE_P(obj, T_IMEMO) && !RB_TYPE_P(obj, T_CLASS) && !RB_TYPE_P(obj, T_MODULE) && !RB_TYPE_P(obj, T_ICLASS)) { RBASIC_SET_SHAPE_ID(obj, rb_shape_transition_frozen(RBASIC_SHAPE_ID(obj))); froze_ivars = true; } /* A T_OBJECT can have a fields imemo too (too_complex and friends), and an imemo * born while its owner was unshareable stays unshareable * (imemo_fields_complex_from_obj), so align it here. */ if (rb_obj_gen_fields_p(obj) || BUILTIN_TYPE(obj) == T_OBJECT) { /* obj is shareable already, so rb_obj_fields_no_ractor_check finds the right * table. Make the fields imemo itself shareable and record shrefs for the * hidden field values the traversal never reaches. */ VALUE fields = rb_obj_fields_no_ractor_check(obj); if (imemo_type_p(fields, imemo_fields)) { // no recursive mark FL_SET_RAW(fields, FL_SHAREABLE); rb_gc_obj_became_shareable(fields); // the imemo carries its owner's shape id, frozen bit included if (froze_ivars) RBASIC_SET_SHAPE_ID(fields, RBASIC_SHAPE_ID(obj)); // Field values the traversal never reaches (hidden internal ivars, say) // can stay unshareable, so record their shrefs to keep the shareable // fields imemo's edges correct. rb_imemo_fields_record_shrefs(fields); } } } #ifndef STRICT_VERIFY_SHAREABLE #define STRICT_VERIFY_SHAREABLE 0 #endif bool rb_ractor_verify_shareable(VALUE obj) { #if STRICT_VERIFY_SHAREABLE rb_gc_verify_shareable(obj); #endif return true; } VALUE rb_obj_set_shareable(VALUE obj) { RUBY_ASSERT(!RB_SPECIAL_CONST_P(obj)); rb_obj_set_shareable_no_assert(obj); RUBY_ASSERT(rb_ractor_verify_shareable(obj)); return obj; } /// traverse function // 2: stop search // 1: skip child // 0: continue enum obj_traverse_iterator_result { traverse_cont, traverse_skip, traverse_stop, }; typedef enum obj_traverse_iterator_result (*rb_obj_traverse_enter_func)(VALUE obj); typedef enum obj_traverse_iterator_result (*rb_obj_traverse_leave_func)(VALUE obj); typedef enum obj_traverse_iterator_result (*rb_obj_traverse_final_func)(VALUE obj); static enum obj_traverse_iterator_result null_leave(VALUE obj); struct obj_traverse_data { rb_obj_traverse_enter_func enter_func; rb_obj_traverse_leave_func leave_func; st_table *rec; VALUE rec_hash; }; struct obj_traverse_callback_data { bool stop; struct obj_traverse_data *data; }; static int obj_traverse_i(VALUE obj, struct obj_traverse_data *data); static int obj_hash_traverse_i(VALUE key, VALUE val, VALUE ptr) { struct obj_traverse_callback_data *d = (struct obj_traverse_callback_data *)ptr; if (obj_traverse_i(key, d->data)) { d->stop = true; return ST_STOP; } if (obj_traverse_i(val, d->data)) { d->stop = true; return ST_STOP; } return ST_CONTINUE; } static void obj_traverse_reachable_i(VALUE obj, void *ptr) { struct obj_traverse_callback_data *d = (struct obj_traverse_callback_data *)ptr; // We need to assume foreign objects are properly shareable. If we check // their shareability, this could result in a crash because the page could // be unmapped. The object could also be freed concurrently. if (!rb_objspace_live_object_p(obj)) { return; } // If the object is "garbage" (unmarked during lazy sweeping), it's okay // to traverse it. if (obj_traverse_i(obj, d->data)) { d->stop = true; } } // Traverse obj's children via its GC mark function. Returns 1 to stop. static int obj_traverse_reachable(VALUE obj, struct obj_traverse_data *data) { struct obj_traverse_callback_data d = { .stop = false, .data = data, }; rb_objspace_reachable_objects_from_local(obj, obj_traverse_reachable_i, &d); return d.stop; } static struct st_table * obj_traverse_rec(struct obj_traverse_data *data) { if (UNLIKELY(!data->rec)) { data->rec_hash = rb_ident_hash_new(); rb_obj_hide(data->rec_hash); data->rec = RHASH_ST_TABLE(data->rec_hash); } return data->rec; } static int obj_traverse_ivar_foreach_i(ID key, VALUE val, st_data_t ptr) { struct obj_traverse_callback_data *d = (struct obj_traverse_callback_data *)ptr; if (obj_traverse_i(val, d->data)) { d->stop = true; return ST_STOP; } return ST_CONTINUE; } static int obj_traverse_i(VALUE obj, struct obj_traverse_data *data) { if (RB_SPECIAL_CONST_P(obj)) return 0; switch (data->enter_func(obj)) { case traverse_cont: break; case traverse_skip: return 0; // skip children case traverse_stop: return 1; // stop search } if (UNLIKELY(st_insert(obj_traverse_rec(data), obj, 1))) { // already traversed return 0; } RB_OBJ_WRITTEN(data->rec_hash, Qundef, obj); if (rb_obj_shape_has_ivars(obj)) { struct obj_traverse_callback_data d = { .stop = false, .data = data, }; rb_ivar_foreach(obj, obj_traverse_ivar_foreach_i, (st_data_t)&d); if (d.stop) return 1; } switch (BUILTIN_TYPE(obj)) { // no child node case T_STRING: case T_FLOAT: case T_BIGNUM: case T_REGEXP: case T_SYMBOL: break; case T_OBJECT: /* Instance variables already traversed. */ break; case T_ARRAY: { for (rb_len_t i = 0; i < RARRAY_LEN(obj); i++) { VALUE e = RARRAY_AREF(obj, i); if (obj_traverse_i(e, data)) return 1; } } break; case T_HASH: { if (obj_traverse_i(RHASH_IFNONE(obj), data)) return 1; struct obj_traverse_callback_data d = { .stop = false, .data = data, }; rb_hash_foreach(obj, obj_hash_traverse_i, (VALUE)&d); if (d.stop) return 1; } break; case T_STRUCT: { rb_len_t len = RSTRUCT_LEN_RAW(obj); const VALUE *ptr = RSTRUCT_CONST_PTR(obj); for (rb_len_t i=0; istr, data)) return 1; break; case T_RATIONAL: if (obj_traverse_i(RRATIONAL(obj)->num, data)) return 1; if (obj_traverse_i(RRATIONAL(obj)->den, data)) return 1; break; case T_COMPLEX: if (obj_traverse_i(RCOMPLEX(obj)->real, data)) return 1; if (obj_traverse_i(RCOMPLEX(obj)->imag, data)) return 1; break; case T_DATA: { void *const ptr = RTYPEDDATA_GET_DATA(obj); const rb_data_type_t *type = RTYPEDDATA_TYPE(obj); if (!ptr || !type->function.dmark) { // no references (the class and ivars are handled elsewhere) } else if (type->flags & RUBY_TYPED_DECL_MARKING) { const size_t *offsets = (const size_t *)(uintptr_t)type->function.dmark; for (; *offsets != RUBY_REF_END; offsets++) { VALUE ref = *(VALUE *)((char *)ptr + *offsets); if (obj_traverse_i(ref, data)) return 1; } } else { if (obj_traverse_reachable(obj, data)) return 1; } } break; case T_IMEMO: // TODO: Not sure this can actually happen; traverse rather than crash. if (obj_traverse_reachable(obj, data)) return 1; break; // unreachable case T_CLASS: case T_MODULE: case T_ICLASS: default: rp(obj); rb_bug("unreachable"); } if (data->leave_func(obj) == traverse_stop) { return 1; } else { return 0; } } struct rb_obj_traverse_final_data { rb_obj_traverse_final_func final_func; int stopped; }; static int obj_traverse_final_i(st_data_t key, st_data_t val, st_data_t arg) { struct rb_obj_traverse_final_data *data = (void *)arg; if (data->final_func(key)) { data->stopped = 1; return ST_STOP; } return ST_CONTINUE; } // 0: traverse all // 1: stopped static int rb_obj_traverse(VALUE obj, rb_obj_traverse_enter_func enter_func, rb_obj_traverse_leave_func leave_func, rb_obj_traverse_final_func final_func) { struct obj_traverse_data data = { .enter_func = enter_func, .leave_func = leave_func, .rec = NULL, }; if (obj_traverse_i(obj, &data)) return 1; if (final_func && data.rec) { struct rb_obj_traverse_final_data f = {final_func, 0}; st_foreach(data.rec, obj_traverse_final_i, (st_data_t)&f); return f.stopped; } return 0; } static int allow_frozen_shareable_p(VALUE obj) { if (RB_TYPE_P(obj, T_FILE)) { return false; } else if (!RB_TYPE_P(obj, T_DATA)) { return true; } else { const rb_data_type_t *type = RTYPEDDATA_TYPE(obj); if (type->flags & RUBY_TYPED_FROZEN_SHAREABLE) { return true; } } return false; } static void make_shareable_freeze(VALUE obj) { VALUE klass = RBASIC_CLASS(obj); if (klass == rb_cString && BASIC_OP_UNREDEFINED_P(BOP_FREEZE, STRING_REDEFINED_OP_FLAG)) { rb_str_freeze(obj); } else if (klass == rb_cArray && BASIC_OP_UNREDEFINED_P(BOP_FREEZE, ARRAY_REDEFINED_OP_FLAG)) { rb_ary_freeze(obj); } else if (klass == rb_cHash && BASIC_OP_UNREDEFINED_P(BOP_FREEZE, HASH_REDEFINED_OP_FLAG)) { rb_hash_freeze(obj); } else { rb_funcall(obj, idFreeze, 0); } } static enum obj_traverse_iterator_result make_shareable_check_shareable_freeze(VALUE obj, enum obj_traverse_iterator_result result) { if (!RB_OBJ_FROZEN_RAW(obj)) { make_shareable_freeze(obj); if (UNLIKELY(!RB_OBJ_FROZEN_RAW(obj))) { rb_raise(rb_eRactorError, "#freeze does not freeze object correctly"); } if (RB_OBJ_SHAREABLE_P(obj)) { return traverse_skip; } } return result; } static int obj_refer_only_shareables_p(VALUE obj); static enum obj_traverse_iterator_result make_shareable_check_shareable(VALUE obj) { VM_ASSERT(!SPECIAL_CONST_P(obj)); if (rb_ractor_shareable_p(obj)) { return traverse_skip; } else if (!allow_frozen_shareable_p(obj)) { if (!RB_TYPE_P(obj, T_DATA)) { rb_raise(rb_eRactorError, "can not make shareable object for %+"PRIsVALUE, obj); } else if (RTYPEDDATA_TYPE(obj)->flags & RUBY_TYPED_FROZEN_SHAREABLE_NO_REC) { if (obj_refer_only_shareables_p(obj)) { make_shareable_check_shareable_freeze(obj, traverse_skip); RB_OBJ_SET_SHAREABLE(obj); return traverse_skip; } else { rb_raise(rb_eRactorError, "can not make shareable object for %+"PRIsVALUE" because it refers unshareable objects", obj); } } else if (rb_obj_is_proc(obj)) { rb_proc_ractor_make_shareable(obj, Qundef); return traverse_cont; } else { rb_raise(rb_eRactorError, "can not make shareable object for %+"PRIsVALUE, obj); } } switch (TYPE(obj)) { case T_IMEMO: return traverse_skip; case T_OBJECT: { // If a T_OBJECT is shared and has no free capacity, we can't safely store the object_id inline, // as it would require to move the object content into an external buffer. // This is only a problem for T_OBJECT, given other types have external fields and can do RCU. // To avoid this issue, we proactively create the object_id. shape_id_t shape_id = RBASIC_SHAPE_ID(obj); attr_index_t capacity = RSHAPE_CAPACITY(shape_id); attr_index_t free_capacity = capacity - RSHAPE_LEN(shape_id); if (!rb_shape_has_object_id(shape_id) && capacity && !free_capacity) { rb_obj_id(obj); } } break; default: break; } return make_shareable_check_shareable_freeze(obj, traverse_cont); } static enum obj_traverse_iterator_result mark_shareable(VALUE obj) { if (RB_BUILTIN_TYPE(obj) == T_STRING) { rb_str_make_independent(obj); } else if (RB_BUILTIN_TYPE(obj) == T_ARRAY) { rb_ary_cancel_sharing(obj); } rb_obj_set_shareable_no_assert(obj); return traverse_cont; } VALUE rb_ractor_make_shareable(VALUE obj) { rb_obj_traverse(obj, make_shareable_check_shareable, null_leave, mark_shareable); return obj; } static VALUE ractor_copy(VALUE obj); // defined below VALUE rb_ractor_make_shareable_copy(VALUE obj) { VALUE copy = ractor_copy(obj); return rb_ractor_make_shareable(copy); } VALUE rb_ractor_ensure_shareable(VALUE obj, VALUE name) { if (!rb_ractor_shareable_p(obj)) { VALUE message = rb_sprintf("cannot assign unshareable object to %"PRIsVALUE, name); rb_exc_raise(rb_exc_new_str(rb_eRactorIsolationError, message)); } return obj; } void rb_ractor_ensure_main_ractor(const char *msg) { if (!rb_ractor_main_p()) { rb_raise(rb_eRactorIsolationError, "%s", msg); } } static enum obj_traverse_iterator_result shareable_p_enter(VALUE obj) { if (RB_OBJ_SHAREABLE_P(obj)) { return traverse_skip; } else if (RB_TYPE_P(obj, T_CLASS) || RB_TYPE_P(obj, T_MODULE) || RB_TYPE_P(obj, T_ICLASS)) { // TODO: remove it mark_shareable(obj); return traverse_skip; } else if (RB_OBJ_FROZEN_RAW(obj) && allow_frozen_shareable_p(obj)) { return traverse_cont; } else if (RB_OBJ_FROZEN_RAW(obj) && RB_TYPE_P(obj, T_DATA) && (RTYPEDDATA_TYPE(obj)->flags & RUBY_TYPED_FROZEN_SHAREABLE_NO_REC)) { // Similar to RUBY_TYPED_FROZEN_SHAREABLE, but the object is only // shareable if all reachable objects are already shareable (they // are not made shareable recursively). if (obj_refer_only_shareables_p(obj)) { mark_shareable(obj); return traverse_skip; } } return traverse_stop; // fail } bool rb_ractor_shareable_p_continue(VALUE obj) { if (rb_obj_traverse(obj, shareable_p_enter, null_leave, mark_shareable)) { return false; } else { return true; } } static enum obj_traverse_iterator_result null_leave(VALUE obj) { return traverse_cont; } /// traverse and replace function // 2: stop search // 1: skip child // 0: continue struct obj_traverse_replace_data; static int obj_traverse_replace_i(VALUE obj, struct obj_traverse_replace_data *data); typedef enum obj_traverse_iterator_result (*rb_obj_traverse_replace_enter_func)(VALUE obj, struct obj_traverse_replace_data *data); typedef enum obj_traverse_iterator_result (*rb_obj_traverse_replace_leave_func)(VALUE obj, struct obj_traverse_replace_data *data); struct obj_traverse_replace_data { rb_obj_traverse_replace_enter_func enter_func; rb_obj_traverse_replace_leave_func leave_func; /* old -> new map, a plain st_table: an OLD key may live in another Ractor's * objspace and must not become a GC edge here (marking a freed foreign key is a * UAF). Keys compare by address; replacements stay alive via rec_keepalive. */ st_table *rec; VALUE rec_keepalive; VALUE replacement; bool move; }; struct obj_traverse_replace_callback_data { bool stop; VALUE src; struct obj_traverse_replace_data *data; }; static int obj_hash_traverse_replace_foreach_i(st_data_t key, st_data_t value, st_data_t argp, int error) { return ST_REPLACE; } static int obj_hash_traverse_replace_i(st_data_t *key, st_data_t *val, st_data_t ptr, int exists) { struct obj_traverse_replace_callback_data *d = (struct obj_traverse_replace_callback_data *)ptr; struct obj_traverse_replace_data *data = d->data; if (obj_traverse_replace_i(*key, data)) { d->stop = true; return ST_STOP; } else if (*key != data->replacement) { VALUE v = *key = data->replacement; RB_OBJ_WRITTEN(d->src, Qundef, v); } if (obj_traverse_replace_i(*val, data)) { d->stop = true; return ST_STOP; } else if (*val != data->replacement) { VALUE v = *val = data->replacement; RB_OBJ_WRITTEN(d->src, Qundef, v); } return ST_CONTINUE; } static int obj_iv_hash_traverse_replace_foreach_i(st_data_t _key, st_data_t _val, st_data_t _data, int _x) { return ST_REPLACE; } static int obj_iv_hash_traverse_replace_i(st_data_t * _key, st_data_t * val, st_data_t ptr, int exists) { struct obj_traverse_replace_callback_data *d = (struct obj_traverse_replace_callback_data *)ptr; struct obj_traverse_replace_data *data = d->data; if (obj_traverse_replace_i(*(VALUE *)val, data)) { d->stop = true; return ST_STOP; } else if (*(VALUE *)val != data->replacement) { VALUE v = *(VALUE *)val = data->replacement; RB_OBJ_WRITTEN(d->src, Qundef, v); } return ST_CONTINUE; } static struct st_table * obj_traverse_replace_rec(struct obj_traverse_replace_data *data) { if (UNLIKELY(!data->rec)) { data->rec = st_init_numtable(); data->rec_keepalive = rb_ary_hidden_new(0); } return data->rec; } static void obj_refer_only_shareables_p_i(VALUE obj, void *ptr) { int *pcnt = (int *)ptr; if (!rb_objspace_live_object_p(obj)) { return; } if (!rb_ractor_shareable_p(obj)) { ++*pcnt; } } static int obj_refer_only_shareables_p(VALUE obj) { int cnt = 0; rb_objspace_reachable_objects_from_local(obj, obj_refer_only_shareables_p_i, &cnt); return cnt == 0; } static int obj_traverse_replace_i(VALUE obj, struct obj_traverse_replace_data *data) { st_data_t replacement; if (RB_SPECIAL_CONST_P(obj)) { data->replacement = obj; return 0; } /* Dedup before enter_func, so a revisited shared/cyclic node reuses its recorded * replacement; otherwise the copy path would build a wasteful temporary holding a * containment-breaking cross-objspace edge. */ if (UNLIKELY(st_lookup(obj_traverse_replace_rec(data), (st_data_t)obj, &replacement))) { data->replacement = (VALUE)replacement; return 0; } switch (data->enter_func(obj, data)) { case traverse_cont: break; case traverse_skip: return 0; // skip children case traverse_stop: return 1; // stop search } replacement = (st_data_t)data->replacement; st_insert(obj_traverse_replace_rec(data), (st_data_t)obj, replacement); if (!RB_SPECIAL_CONST_P((VALUE)replacement)) { rb_ary_push(data->rec_keepalive, (VALUE)replacement); } if (!data->move) { obj = replacement; } #define CHECK_AND_REPLACE(parent_obj, v) do { \ VALUE _val = (v); \ if (obj_traverse_replace_i(_val, data)) { return 1; } \ else if (data->replacement != _val) { RB_OBJ_WRITE(parent_obj, &v, data->replacement); } \ } while (0) if (UNLIKELY(rb_obj_gen_fields_p(obj))) { VALUE fields_obj = rb_obj_fields_no_ractor_check(obj); if (UNLIKELY(rb_obj_shape_complex_p(obj))) { struct obj_traverse_replace_callback_data d = { .stop = false, .data = data, .src = fields_obj, }; rb_st_foreach_with_replace( rb_imemo_fields_complex_tbl(fields_obj), obj_iv_hash_traverse_replace_foreach_i, obj_iv_hash_traverse_replace_i, (st_data_t)&d ); if (d.stop) return 1; } else { uint32_t fields_count = RSHAPE_LEN(RBASIC_SHAPE_ID(obj)); VALUE *fields = rb_imemo_fields_ptr(fields_obj); for (uint32_t i = 0; i < fields_count; i++) { CHECK_AND_REPLACE(fields_obj, fields[i]); } } } switch (BUILTIN_TYPE(obj)) { // no child node case T_FLOAT: case T_BIGNUM: case T_REGEXP: case T_FILE: case T_SYMBOL: break; case T_STRING: rb_str_make_independent(obj); break; case T_OBJECT: { VALUE fields_obj = ROBJECT_FIELDS_OBJ(obj); shape_id_t shape_id = RBASIC_SHAPE_ID(fields_obj); if (rb_shape_complex_p(shape_id)) { struct obj_traverse_replace_callback_data d = { .stop = false, .data = data, .src = obj, }; rb_st_foreach_with_replace( rb_imemo_fields_complex_tbl(fields_obj), obj_iv_hash_traverse_replace_foreach_i, obj_iv_hash_traverse_replace_i, (st_data_t)&d ); if (d.stop) return 1; } else { attr_index_t len = RSHAPE_LEN(shape_id); VALUE *ptr = rb_imemo_fields_ptr(fields_obj); for (attr_index_t i = 0; i < len; i++) { CHECK_AND_REPLACE(obj, ptr[i]); } } } break; case T_ARRAY: { rb_ary_cancel_sharing(obj); for (rb_len_t i = 0; i < RARRAY_LEN(obj); i++) { VALUE e = RARRAY_AREF(obj, i); if (obj_traverse_replace_i(e, data)) { return 1; } else if (e != data->replacement) { RARRAY_ASET(obj, i, data->replacement); } } RB_GC_GUARD(obj); } break; case T_HASH: { struct obj_traverse_replace_callback_data d = { .stop = false, .data = data, .src = obj, }; rb_hash_stlike_foreach_with_replace(obj, obj_hash_traverse_replace_foreach_i, obj_hash_traverse_replace_i, (VALUE)&d); if (d.stop) return 1; // TODO: rehash here? VALUE ifnone = RHASH_IFNONE(obj); if (obj_traverse_replace_i(ifnone, data)) { return 1; } else if (ifnone != data->replacement) { RHASH_SET_IFNONE(obj, data->replacement); } } break; case T_STRUCT: { rb_len_t len = RSTRUCT_LEN_RAW(obj); const VALUE *ptr = RSTRUCT_CONST_PTR(obj); for (rb_len_t i=0; istr); break; case T_RATIONAL: CHECK_AND_REPLACE(obj, RRATIONAL(obj)->num); CHECK_AND_REPLACE(obj, RRATIONAL(obj)->den); break; case T_COMPLEX: CHECK_AND_REPLACE(obj, RCOMPLEX(obj)->real); CHECK_AND_REPLACE(obj, RCOMPLEX(obj)->imag); break; case T_DATA: if (!data->move && obj_refer_only_shareables_p(obj)) { break; } else { rb_raise(rb_eRactorError, "can not %s %"PRIsVALUE" object.", data->move ? "move" : "copy", rb_class_of(obj)); } case T_IMEMO: // not supported yet return 1; // unreachable case T_CLASS: case T_MODULE: case T_ICLASS: default: rp(obj); rb_bug("unreachable"); } data->replacement = (VALUE)replacement; if (data->leave_func(obj, data) == traverse_stop) { return 1; } else { return 0; } } // 0: traverse all // 1: stopped static VALUE rb_obj_traverse_replace(VALUE obj, rb_obj_traverse_replace_enter_func enter_func, rb_obj_traverse_replace_leave_func leave_func, bool move) { struct obj_traverse_replace_data data = { .enter_func = enter_func, .leave_func = leave_func, .rec = NULL, .rec_keepalive = Qfalse, .replacement = Qundef, .move = move, }; int stopped = obj_traverse_replace_i(obj, &data); /* The enter and leave functions report failure with traverse_stop rather than by * raising, so this is the only place the table is freed. */ if (data.rec) st_free_table(data.rec); RB_GC_GUARD(data.rec_keepalive); if (stopped) { return Qundef; } else { return data.replacement; } } /* Courier: serializes a Ractor message payload -- copied or moved -- into an xmalloc'd * structure that belongs to no objspace, so no sender GC can mark, sweep, compact or * race with it. A node array with id references handles sharing and cycles, and the * receiver rebuilds it in its own objspace in two passes. Copy and move differ only in * whether the source is read or taken apart: see courier_build.copy. */ enum courier_node_kind { COURIER_KIND_REF, /* an immediate or a shareable object: carried by value */ COURIER_KIND_BACKTRACE, /* an exception's backtrace: frames copied into an off-heap blob */ COURIER_KIND_STRING, COURIER_KIND_ARRAY, COURIER_KIND_HASH, COURIER_KIND_OBJECT, COURIER_KIND_STRUCT, COURIER_KIND_MATCH, COURIER_KIND_IO, COURIER_KIND_REGEXP, /* recompiled from its source and options (copy only) */ COURIER_KIND_HOOKED, /* rebuilt from its dump hook's payload by klass._load, marshal_load or _load_data */ }; /* Marshal's protocols, but nothing is serialized: the hook's return value travels as an * ordinary child node, so sharing, cycles and shareable references all survive. */ enum courier_hook { COURIER_HOOK_NONE, COURIER_HOOK_MARSHAL_DUMP, /* marshal_dump -> alloc + marshal_load */ COURIER_HOOK_DUMP, /* _dump -> klass._load */ COURIER_HOOK_COMPAT, /* rb_marshal_define_compat dumper -> alloc + loader (Set, Process::Status) */ COURIER_HOOK_DUMP_DATA, /* _dump_data -> alloc + _load_data */ }; /* Which one obj's class implements, in Marshal's order of preference. */ static enum courier_hook courier_hook_of(VALUE obj) { if (rb_obj_respond_to(obj, id_marshal_dump, TRUE)) return COURIER_HOOK_MARSHAL_DUMP; if (rb_obj_respond_to(obj, id_dump, TRUE)) return COURIER_HOOK_DUMP; if (rb_marshal_compat_lookup(CLASS_OF(obj), NULL, NULL)) return COURIER_HOOK_COMPAT; if (BUILTIN_TYPE(obj) == T_DATA && rb_obj_respond_to(obj, id_dump_data, TRUE)) return COURIER_HOOK_DUMP_DATA; return COURIER_HOOK_NONE; } struct courier_node { enum courier_node_kind kind; bool frozen; /* The instance and generic ivars every non-REF node can have (a String or Array * can hold generic ivars too) */ uint32_t niv; ID *iv_ids; /* owned by the courier */ uint32_t *iv_vals; /* owned by the courier; node ids */ union { VALUE ref; struct { char *ptr; rb_len_t len, capa; int encidx; VALUE klass; } str; /* the courier owns ptr */ struct { rb_len_t len; uint32_t *elems; VALUE klass; } ary; /* the courier owns elems */ struct { long size; uint32_t *kv; uint32_t ifnone_id; bool compare_by_id; bool proc_default; VALUE klass; } hash; /* owns kv (2*size) */ struct { VALUE klass; } obj; struct { rb_len_t len; uint32_t *elems; VALUE klass; } strct; /* owns elems */ struct { uint32_t regexp_id, str_id; int num_regs; void *regs; VALUE klass; } match; /* owns regs */ struct { void *blob; int size; } bt; /* the courier owns blob */ struct { VALUE src; int options; VALUE klass; } re; /* src is an fstring: shareable */ struct { VALUE klass; uint32_t payload_id; enum courier_hook hook; } hooked; struct { struct rb_io *fptr; /* carried by pointer (it owns the fd) */ VALUE klass; /* The sender-side VALUE members of fptr travel as ordinary child nodes: * capture detaches them from fptr (see the T_FILE case) and rebuild writes * them back into the receiving shell with RB_OBJ_WRITE. */ uint32_t pathv_id, ecopts_id, wc_pre_ecopts_id, wc_asciicompat_id, timeout_id; } io; } u; }; /* A child slot holds a node id, or -- with this bit set -- an index into c->refs. * The courier is in-process, so a shareable payload can travel as the VALUE itself * instead of costing a whole courier_node; the basket holding the courier marks c->refs. */ #define COURIER_ID_REF_BIT 0x80000000u struct rb_ractor_courier { struct courier_node *nodes; uint32_t *order; /* node ids in capture's post-order: children before parents */ uint32_t count; uint32_t capa; VALUE *refs; /* shareable payloads, embedded by value */ uint32_t refs_count; uint32_t refs_capa; uint32_t root; /* src VALUE -> (node id + 1), only while building. Marked so that a key cannot * die (a dump hook's payload has no other owner) or move (compaction). */ st_table *seen; }; struct courier_build { struct rb_ractor_courier *c; /* Copy mode: read the sources instead of taking them apart. No husk, no buffer * hand-over, no freeing of the source's internals. */ bool copy; uint32_t ordered; /* nodes appended to c->order so far */ }; static uint32_t courier_capture(struct courier_build *b, VALUE obj); /* Off the hot path: the preflight sizes both arrays, so this only runs if its count * came out short. Swap a fresh array in rather than realloc -- the courier is a GC * root while it is being built, and a realloc leaves the old pointer live over a * window where it may already have been freed. */ NOINLINE(static void courier_grow_nodes(struct rb_ractor_courier *c)); NOINLINE(static void courier_grow_refs(struct rb_ractor_courier *c)); static void courier_grow_nodes(struct rb_ractor_courier *c) { uint32_t capa = c->capa ? c->capa * 2 : 8; struct courier_node *nodes = ALLOC_N(struct courier_node, capa); if (c->count > 0) MEMCPY(nodes, c->nodes, struct courier_node, c->count); struct courier_node *old_nodes = c->nodes; c->nodes = nodes; c->capa = capa; ruby_xfree(old_nodes); REALLOC_N(c->order, uint32_t, capa); } static void courier_grow_refs(struct rb_ractor_courier *c) { uint32_t capa = c->refs_capa ? c->refs_capa * 2 : 8; VALUE *refs = ALLOC_N(VALUE, capa); if (c->refs_count > 0) MEMCPY(refs, c->refs, VALUE, c->refs_count); VALUE *old_refs = c->refs; c->refs = refs; c->refs_capa = capa; ruby_xfree(old_refs); } static uint32_t courier_alloc_node(struct rb_ractor_courier *c) { if (RB_UNLIKELY(c->count == c->capa)) courier_grow_nodes(c); /* Fill the slot with a harmless REF/Qnil and bump the count only after, the way * courier_alloc_ref does: the courier is a GC root while it is being built, and * the mark walks nodes[0, count). A captured node overwrites this later. */ struct courier_node *n = &c->nodes[c->count]; n->kind = COURIER_KIND_REF; n->frozen = false; n->niv = 0; n->iv_ids = NULL; n->iv_vals = NULL; n->u.ref = Qnil; return c->count++; } /* Size the arrays from the preflight's count, so capture never grows them. A count * that turns out short is not a problem: the growth path below still works. */ static void courier_reserve(struct rb_ractor_courier *c, uint32_t nodes, uint32_t refs) { if (nodes > 0) { c->nodes = ALLOC_N(struct courier_node, nodes); c->order = ALLOC_N(uint32_t, nodes); c->capa = nodes; } if (refs > 0) { c->refs = ALLOC_N(VALUE, refs); c->refs_capa = refs; } } /* Embed a shareable payload by value and return its tagged child id. No dedup: a REF * is the same word however often it appears, and an array of immediates would otherwise * pay a lookup and an insert per element. */ static uint32_t courier_alloc_ref(struct rb_ractor_courier *c, VALUE v) { /* The count is bumped only after the slot holds a real VALUE: the courier is a GC * root while it is being built and must never be walkable half-written. */ if (RB_UNLIKELY(c->refs_count == c->refs_capa)) courier_grow_refs(c); c->refs[c->refs_count] = v; return COURIER_ID_REF_BIT | c->refs_count++; } /* Resolve a child slot to the object it names. */ static VALUE courier_child(const struct rb_ractor_courier *c, VALUE shells, uint32_t id) { if (id & COURIER_ID_REF_BIT) return c->refs[id & ~COURIER_ID_REF_BIT]; return RARRAY_AREF(shells, id); } /* Turn a moved source into a valid RactorMovedObject without passing through flags==0, * so a concurrent foreign marker always sees either the original object or the shell. */ static void move_neutralize_source(VALUE obj) { /* The shell stays in the original slot: keep the capacity bits, give it a frozen * field-less ROBJECT shape (read before the flags are overwritten). The old body is * then never read as ivars and compaction's slot-size check still holds. */ shape_id_t shape_id = (RBASIC_SHAPE_ID(obj) & SHAPE_ID_CAPACITY_MASK) | ROOT_SHAPE_ID | SHAPE_ID_LAYOUT_ROBJECT | SHAPE_ID_FL_FROZEN; /* A non-T_OBJECT host (a String with ivars, say) must drop its generic_fields * entry: obj stops being a host below and its fields_obj is collected, so a stale * entry would let the global GC walk a freed value. */ rb_free_generic_ivar(obj); /* A copy-on-write sharer reads its payload straight out of an embedded root's slot * (String#dup of a frozen string, Array#[] of a frozen array), and it outlives the * move, so that body has to survive as it is. */ bool wipe_body = true; switch (BUILTIN_TYPE(obj)) { case T_STRING: if (!STR_EMBED_P(obj) && !rb_str_reembeddable_p(obj)) { /* A heap (non-embedded), shared root string keeps its buffer because * other strings reference this shared root. It needs to keep T_STRING * because otherwise the GC will not free the buffer when this object * dies which will leak memory. */ RBASIC_SET_CLASS_RAW(obj, rb_cRactorMovedObject); RBASIC(obj)->flags |= FL_FREEZE; RBASIC_SET_FULL_SHAPE_ID(obj, (shape_id & ~SHAPE_ID_LAYOUT_MASK) | SHAPE_ID_LAYOUT_OTHER); RSTRING(obj)->len = 0; return; } wipe_body = !rb_str_embedded_shared_root_p(obj); break; case T_ARRAY: if (!ARY_EMBED_P(obj) && !ARY_SHARED_P(obj) && (ARY_SHARED_ROOT_P(obj) || OBJ_FROZEN(obj))) { /* A heap (non-embedded), shared root array keeps its buffer because * other arrays reference this shared root. It needs to keep T_ARRAY * because otherwise the GC will not free the buffer when this object * dies which will leak memory. */ RBASIC_SET_CLASS_RAW(obj, rb_cRactorMovedObject); RBASIC(obj)->flags |= FL_FREEZE; RBASIC_SET_FULL_SHAPE_ID(obj, (shape_id & ~SHAPE_ID_LAYOUT_MASK) | SHAPE_ID_LAYOUT_OTHER); if (!ARY_SHARED_ROOT_P(obj)) { /* Present as empty to stale readers. Not for a shared root: its * len doubles as the buffer capacity that ARY_HEAP_SIZE frees by. */ RARRAY(obj)->as.heap.len = 0; } return; } wipe_body = !rb_ary_embedded_shared_root_p(obj); break; default: break; } /* Keep FL_FINALIZE: the finalizer table entry stays keyed on this slot, and a * shell without the flag makes the two disagree (rb_gc_impl_shutdown_call_finalizer_i * asserts on it). The finalizer runs when the shell dies, in the Ractor that * defined it; the rebuilt object gets fresh flags and does not inherit it. */ VALUE flags = T_OBJECT | FL_FREEZE | (RBASIC(obj)->flags & (FL_PROMOTED | FL_FINALIZE)); /* Read the slot size before the header is rewritten. */ size_t slot_size = rb_gc_obj_slot_size(obj); RBASIC_SET_CLASS_RAW(obj, rb_cRactorMovedObject); RBASIC(obj)->flags = flags; RBASIC_SET_FULL_SHAPE_ID(obj, shape_id); /* Wipe the old body. The shell has no fields, so nothing reads it as ivars, but * C code holding the object from before the move still reads it with its old type * (a running Array iteration, the RMatch capa of a $~ entry): a zeroed body makes * those reads see an empty object instead of stale internals. */ if (wipe_body) { MEMZERO((char *)obj + sizeof(struct RBasic), char, slot_size - sizeof(struct RBasic)); } } struct courier_hash_ctx { struct courier_build *b; uint32_t *kv; long i; }; static int courier_capture_hash_i(st_data_t key, st_data_t val, st_data_t arg) { struct courier_hash_ctx *hc = (struct courier_hash_ctx *)arg; uint32_t kid = courier_capture(hc->b, (VALUE)key); uint32_t vid = courier_capture(hc->b, (VALUE)val); hc->kv[hc->i++] = kid; hc->kv[hc->i++] = vid; return ST_CONTINUE; } struct courier_obj_ctx { struct courier_build *b; ID *ids; uint32_t *vals; long n; long capa; }; static int courier_capture_ivar_i(ID name, VALUE val, st_data_t arg) { struct courier_obj_ctx *oc = (struct courier_obj_ctx *)arg; if (oc->n == oc->capa) { oc->capa = oc->capa ? oc->capa * 2 : 4; REALLOC_N(oc->ids, ID, oc->capa); REALLOC_N(oc->vals, uint32_t, oc->capa); } uint32_t vid = courier_capture(oc->b, val); oc->ids[oc->n] = name; oc->vals[oc->n] = vid; oc->n++; return ST_CONTINUE; } /* Capture obj's instance and generic ivars as node ids, recursing into the values. * Handles both a T_OBJECT's inline ivars and the generic ivars of a String, Array and * so on. */ static void courier_capture_ivars(struct courier_build *b, VALUE obj, uint32_t id) { struct courier_obj_ctx oc = { b, NULL, NULL, 0, 0 }; rb_ivar_foreach_buffered(obj, courier_capture_ivar_i, (st_data_t)&oc); b->c->nodes[id].niv = (uint32_t)oc.n; b->c->nodes[id].iv_ids = oc.ids; b->c->nodes[id].iv_vals = oc.vals; } /* Run obj's dump hook and capture what it returns as an ordinary child node. That * includes _dump's String: Marshal writes its ivars next to its bytes (Time keeps the * sub-microsecond part and the zone there), and a String node carries them the same * way. */ static void courier_capture_hooked(struct courier_build *b, VALUE obj, uint32_t id, enum courier_hook hook) { VALUE klass = rb_obj_class(obj); VALUE payload; switch (hook) { case COURIER_HOOK_DUMP: { /* _dump takes the depth limit Marshal would have applied; a copy has none. */ VALUE limit = INT2FIX(-1); payload = rb_funcallv(obj, id_dump, 1, &limit); if (!RB_TYPE_P(payload, T_STRING)) { rb_raise(rb_eTypeError, "_dump() must return string"); } break; } case COURIER_HOOK_MARSHAL_DUMP: payload = rb_funcallv(obj, id_marshal_dump, 0, 0); break; case COURIER_HOOK_DUMP_DATA: payload = rb_funcallv(obj, id_dump_data, 0, 0); break; case COURIER_HOOK_COMPAT: { VALUE (*dumper)(VALUE); rb_marshal_compat_lookup(klass, &dumper, NULL); payload = dumper(obj); break; } default: rb_bug("courier_capture_hooked: no dump protocol"); } uint32_t payload_id = courier_capture(b, payload); b->c->nodes[id].kind = COURIER_KIND_HOOKED; b->c->nodes[id].u.hooked.klass = klass; b->c->nodes[id].u.hooked.hook = hook; b->c->nodes[id].u.hooked.payload_id = payload_id; } /* A move carries the singleton class with its object; a copy drops it, like #dup. */ static inline VALUE courier_klass(struct courier_build *b, VALUE obj) { return b->copy ? rb_obj_class(obj) : RBASIC_CLASS(obj); } /* Capture obj into the courier, recurse into its children, return its node id. The id * is registered before recursing (a cycle back resolves to the same node); node fields * are written after (recursion can realloc c->nodes); a move neutralizes the source * exactly once after the switch. */ static uint32_t courier_capture(struct courier_build *b, VALUE obj) { /* An immediate is never in seen (only captured objects are inserted), so it can * skip the lookup entirely: that is the whole cost of an array of numbers. */ if (RB_SPECIAL_CONST_P(obj)) { return courier_alloc_ref(b->c, obj); } /* Seen first, and only then shareable: move husks each source as it goes, and a * husk is a frozen field-less object, which rb_ractor_shareable_p answers true for. * Testing shareable first would embed the husk instead of resolving the second * occurrence to the node the first one built. */ st_data_t existing; if (st_lookup(b->c->seen, (st_data_t)obj, &existing)) { return (uint32_t)existing - 1; } if (rb_ractor_shareable_p(obj)) { return courier_alloc_ref(b->c, obj); } uint32_t id = courier_alloc_node(b->c); st_insert(b->c->seen, (st_data_t)obj, (st_data_t)(uintptr_t)(id + 1)); /* Reject an unmovable object before anything is mutated. */ if (BUILTIN_TYPE(obj) == T_FILE && RFILE(obj)->fptr == NULL) { rb_raise(rb_eRactorError, "can not move an uninitialized IO"); } bool frozen = OBJ_FROZEN(obj); b->c->nodes[id].frozen = frozen; courier_capture_ivars(b, obj, id); /* shared: instance and generic ivars */ switch (BUILTIN_TYPE(obj)) { case T_STRING: { /* Give the source its own buffer (drop sharing, copy a static STR_NOFREE one). * Safe even when frozen: it changes ownership, not content. Afterwards a string * is embedded, owns a private heap buffer, or is a shared ROOT (a no-op). */ if (!b->copy) rb_str_make_independent(obj); rb_len_t len = RSTRING_LEN(obj); int encidx = ENCODING_GET(obj); /* The receiver adopts this buffer as a String body, which is freed by size: * capa has to describe the allocation exactly (capa + terminator bytes). */ const int termlen = rb_enc_mbminlen(rb_enc_from_index(encidx)); char *ptr; rb_len_t capa; if (!b->copy && !STR_EMBED_P(obj) && rb_str_reembeddable_p(obj)) { /* Owns a private heap buffer: carry the pointer over (zero-copy) and leave * the source as a shell that does not free it. */ ptr = RSTRING(obj)->as.heap.ptr; capa = RSTRING(obj)->as.heap.aux.capa; } else { /* Embedded or a shared root: copy the bytes into a courier-owned buffer. * Taking a root's buffer would dangle its copy-on-write children, so leave * it (the same reason T_ARRAY excludes ARY_SHARED_ROOT_P below). */ ptr = ALLOC_N(char, len + termlen); if (len) memcpy(ptr, RSTRING_PTR(obj), len); memset(ptr + len, 0, termlen); capa = len; } b->c->nodes[id].kind = COURIER_KIND_STRING; b->c->nodes[id].u.str.klass = courier_klass(b, obj); b->c->nodes[id].u.str.ptr = ptr; b->c->nodes[id].u.str.len = len; b->c->nodes[id].u.str.capa = capa; b->c->nodes[id].u.str.encidx = encidx; break; } case T_ARRAY: { rb_len_t len = RARRAY_LEN(obj); uint32_t *elems = len ? ALLOC_N(uint32_t, len) : NULL; for (rb_len_t i = 0; i < len; i++) { elems[i] = courier_capture(b, RARRAY_AREF(obj, i)); } b->c->nodes[id].kind = COURIER_KIND_ARRAY; b->c->nodes[id].u.ary.klass = courier_klass(b, obj); b->c->nodes[id].u.ary.len = len; b->c->nodes[id].u.ary.elems = elems; /* Free the source's heap buffer now that the children were read, but only when it * is private: a sharer's belongs to its root, a root's to its sharers -- and a * frozen array is a root without carrying the flag. */ if (!b->copy && !ARY_EMBED_P(obj) && !ARY_SHARED_P(obj) && !ARY_SHARED_ROOT_P(obj) && !OBJ_FROZEN(obj)) { ruby_xfree((void *)RARRAY_CONST_PTR(obj)); } break; } case T_HASH: { uint32_t ifnone_id = courier_capture(b, RHASH_IFNONE(obj)); /* An entry count, not a String or Array length, so u.hash.size stays long */ long size = RHASH_SIZE(obj); uint32_t *kv = size ? ALLOC_N(uint32_t, size * 2) : NULL; struct courier_hash_ctx hc = { b, kv, 0 }; rb_hash_stlike_foreach(obj, courier_capture_hash_i, (st_data_t)&hc); b->c->nodes[id].kind = COURIER_KIND_HASH; b->c->nodes[id].u.hash.klass = courier_klass(b, obj); b->c->nodes[id].u.hash.size = size; b->c->nodes[id].u.hash.kv = kv; b->c->nodes[id].u.hash.ifnone_id = ifnone_id; b->c->nodes[id].u.hash.compare_by_id = RTEST(rb_hash_compare_by_id_p(obj)); b->c->nodes[id].u.hash.proc_default = FL_TEST_RAW(obj, RHASH_PROC_DEFAULT) != 0; /* Free the source's st-table internals (an ar table lives in the slot) */ if (!b->copy) rb_hash_free(obj); break; } case T_OBJECT: b->c->nodes[id].kind = COURIER_KIND_OBJECT; b->c->nodes[id].u.obj.klass = courier_klass(b, obj); break; case T_STRUCT: { rb_len_t len = RSTRUCT_LEN(obj); uint32_t *elems = len ? ALLOC_N(uint32_t, len) : NULL; for (rb_len_t i = 0; i < len; i++) { elems[i] = courier_capture(b, RSTRUCT_GET(obj, (int)i)); } b->c->nodes[id].kind = COURIER_KIND_STRUCT; b->c->nodes[id].u.strct.len = len; b->c->nodes[id].u.strct.elems = elems; b->c->nodes[id].u.strct.klass = courier_klass(b, obj); /* Free the source's private heap buffer (an embedded struct has none) */ if (!b->copy && RSTRUCT_EMBED_LEN(obj) == 0) { ruby_xfree((void *)RSTRUCT_CONST_PTR(obj)); } break; } case T_MATCH: { /* The regexp and the matched string travel as ordinary children; re.c dumps the * registers (freeing the source's onig and char_offset). */ VALUE re, st; int nregs; void *regs = rb_match_blob_dump(obj, &re, &st, &nregs, !b->copy); uint32_t rid = courier_capture(b, re); uint32_t sid = courier_capture(b, st); b->c->nodes[id].kind = COURIER_KIND_MATCH; b->c->nodes[id].u.match.regexp_id = rid; b->c->nodes[id].u.match.str_id = sid; b->c->nodes[id].u.match.num_regs = nregs; b->c->nodes[id].u.match.regs = regs; b->c->nodes[id].u.match.klass = courier_klass(b, obj); break; } case T_FILE: { VM_ASSERT(!b->copy); /* copy_courier_supported_p rejects it */ /* Carry the whole fptr (fd included) by pointer; the source shell does not * close it. fptr's VALUE members lose their root once the source is T_MOVED, * so capture them as ordinary child nodes, detached; rebuild writes them back. */ struct rb_io *fptr = RFILE(obj)->fptr; VM_ASSERT(!RTEST(fptr->tied_io_for_writing) && !RTEST(fptr->wakeup_mutex)); uint32_t pathv_id = courier_capture(b, fptr->pathv); uint32_t ecopts_id = courier_capture(b, fptr->encs.ecopts); uint32_t wc_pre_id = courier_capture(b, fptr->writeconv_pre_ecopts); uint32_t wc_ac_id = courier_capture(b, fptr->writeconv_asciicompat); uint32_t timeout_id = courier_capture(b, fptr->timeout); fptr->self = Qnil; /* it points at the moved-from T_MOVED; attach rebuilds it */ fptr->pathv = Qnil; fptr->encs.ecopts = Qnil; fptr->writeconv_pre_ecopts = Qnil; fptr->writeconv_asciicompat = Qnil; fptr->timeout = Qnil; fptr->write_lock = Qnil; fptr->wakeup_mutex = Qnil; fptr->tied_io_for_writing = 0; /* io.c tests it as a C boolean, so 0 rather than Qnil */ b->c->nodes[id].kind = COURIER_KIND_IO; b->c->nodes[id].u.io.fptr = fptr; b->c->nodes[id].u.io.klass = courier_klass(b, obj); b->c->nodes[id].u.io.pathv_id = pathv_id; b->c->nodes[id].u.io.ecopts_id = ecopts_id; b->c->nodes[id].u.io.wc_pre_ecopts_id = wc_pre_id; b->c->nodes[id].u.io.wc_asciicompat_id = wc_ac_id; b->c->nodes[id].u.io.timeout_id = timeout_id; break; } case T_REGEXP: /* Copy only: the receiver compiles the source again, as Marshal does. Move * would have to take the onig pattern apart. */ if (b->copy) { /* The source is an fstring (reg_set_source), so it can be carried as is. */ VALUE src = RREGEXP_SRC(obj); VM_ASSERT(rb_ractor_shareable_p(src)); b->c->nodes[id].kind = COURIER_KIND_REGEXP; b->c->nodes[id].u.re.klass = courier_klass(b, obj); b->c->nodes[id].u.re.src = src; b->c->nodes[id].u.re.options = rb_reg_options(obj); break; } /* fall through */ case T_DATA: /* Only an exception's backtrace, and only for a copy: move still refuses every * T_DATA (its source would have to be taken apart). */ if (b->copy && rb_backtrace_p(obj)) { int size; void *blob = rb_backtrace_blob_dump(obj, &size); b->c->nodes[id].kind = COURIER_KIND_BACKTRACE; b->c->nodes[id].u.bt.blob = blob; b->c->nodes[id].u.bt.size = size; break; } /* fall through */ default: { /* Copy has one more option: the object's own dump hook, which the preflight * already found. Move has not, since it would have to take the source apart. */ enum courier_hook hook = b->copy ? courier_hook_of(obj) : COURIER_HOOK_NONE; if (hook == COURIER_HOOK_NONE) { rb_raise(rb_eRactorError, "can not %s a %"PRIsVALUE" object", b->copy ? "copy" : "move", rb_class_name(rb_obj_class(obj))); } courier_capture_hooked(b, obj, id, hook); break; } } if (!b->copy) move_neutralize_source(obj); /* Every child has returned: the post-order materialize fills in. */ b->c->order[b->ordered++] = id; return id; } /* Like the copy walk, this also sizes the courier: see copy_support_ctx. */ struct move_preflight_ctx { st_table *seen; uint32_t nodes, refs; }; static void move_preflight(VALUE obj, struct move_preflight_ctx *ctx); static int move_preflight_ivar_i(ID name, VALUE val, st_data_t arg) { move_preflight(val, (struct move_preflight_ctx *)arg); return ST_CONTINUE; } static int move_preflight_hash_i(st_data_t key, st_data_t val, st_data_t arg) { move_preflight((VALUE)key, (struct move_preflight_ctx *)arg); move_preflight((VALUE)val, (struct move_preflight_ctx *)arg); return ST_CONTINUE; } /* A read-only pre-walk of courier_capture's decision tree. Capture turns sources into * T_MOVED as it goes, so an unmovable object midway would leave the graph broken beyond * repair; every "can not move" error is raised here, before anything is mutated. */ static void move_preflight(VALUE obj, struct move_preflight_ctx *ctx) { st_table *const seen = ctx->seen; if (RB_SPECIAL_CONST_P(obj) || rb_ractor_shareable_p(obj)) { ctx->refs++; return; } if (st_lookup(seen, (st_data_t)obj, NULL)) return; /* cycle */ st_insert(seen, (st_data_t)obj, 0); ctx->nodes++; /* The receiver takes over a materialized singleton class, so its contents have to * be movable too. */ VALUE klass = RBASIC_CLASS(obj); if (RB_UNLIKELY(klass && FL_TEST_RAW(klass, FL_SINGLETON))) { rb_class_check_singleton_movable(klass); } switch (BUILTIN_TYPE(obj)) { case T_STRING: case T_OBJECT: break; /* children are ivars only (below) */ case T_MATCH: { struct RMatch *rm = RMATCH(obj); move_preflight(rm->regexp, ctx); move_preflight(rm->str, ctx); break; } case T_ARRAY: for (rb_len_t i = 0; i < RARRAY_LEN(obj); i++) { move_preflight(RARRAY_AREF(obj, i), ctx); } break; case T_HASH: rb_hash_stlike_foreach(obj, move_preflight_hash_i, (st_data_t)ctx); move_preflight(RHASH_IFNONE(obj), ctx); break; case T_STRUCT: for (rb_len_t i = 0; i < RSTRUCT_LEN(obj); i++) { move_preflight(RSTRUCT_GET(obj, (int)i), ctx); } break; case T_FILE: { struct rb_io *fptr = RFILE(obj)->fptr; if (fptr == NULL) { rb_raise(rb_eRactorError, "can not move an uninitialized IO"); } if (RTEST(fptr->tied_io_for_writing)) { /* A popen("r+") pair: moving one side would dangle the tied writer on the * sender. */ rb_raise(rb_eRactorError, "can not move an IO tied to a writer IO"); } if (RTEST(fptr->wakeup_mutex)) { /* A close is in progress: a thread is blocked on this IO. */ rb_raise(rb_eRactorError, "can not move an IO that is being closed"); } move_preflight(fptr->pathv, ctx); move_preflight(fptr->encs.ecopts, ctx); move_preflight(fptr->writeconv_pre_ecopts, ctx); move_preflight(fptr->writeconv_asciicompat, ctx); move_preflight(fptr->timeout, ctx); break; } default: rb_raise(rb_eRactorError, "can not move a %"PRIsVALUE" object", rb_class_name(rb_obj_class(obj))); } rb_ivar_foreach(obj, move_preflight_ivar_i, (st_data_t)ctx); } /* The walk also sizes the courier: one node per distinct unshareable object, one ref * per occurrence of a shareable one -- exactly what courier_capture allocates, so the * arrays never have to grow while the graph is being captured. */ struct copy_support_ctx { st_table *seen; uint32_t nodes, refs; bool ok; }; static bool copy_courier_supported_p(VALUE obj, struct copy_support_ctx *ctx); static int copy_support_val_i(st_data_t val, st_data_t arg) { struct copy_support_ctx *ctx = (struct copy_support_ctx *)arg; if (!copy_courier_supported_p((VALUE)val, ctx)) { ctx->ok = false; return ST_STOP; } return ST_CONTINUE; } static int copy_support_ivar_i(ID name, VALUE val, st_data_t arg) { return copy_support_val_i((st_data_t)val, arg); } static int copy_support_hash_i(st_data_t key, st_data_t val, st_data_t arg) { if (copy_support_val_i(key, arg) == ST_STOP) return ST_STOP; return copy_support_val_i(val, arg); } /* Read-only walk: can the copy courier carry obj's whole graph? A no is a send error. */ static bool copy_courier_supported_p(VALUE obj, struct copy_support_ctx *ctx) { st_table *const seen = ctx->seen; if (RB_SPECIAL_CONST_P(obj) || rb_ractor_shareable_p(obj)) { ctx->refs++; return true; } if (st_lookup(seen, (st_data_t)obj, NULL)) return true; /* cycle */ st_insert(seen, (st_data_t)obj, 0); ctx->nodes++; if (RBASIC_CLASS(obj) == 0) return false; switch (BUILTIN_TYPE(obj)) { case T_STRING: case T_OBJECT: case T_REGEXP: break; /* children are ivars only (below) */ case T_MATCH: { struct RMatch *rm = RMATCH(obj); if (!copy_courier_supported_p(rm->regexp, ctx)) return false; if (!copy_courier_supported_p(rm->str, ctx)) return false; break; } case T_DATA: /* An exception's backtrace is the one T_DATA the courier carries natively. */ if (!rb_backtrace_p(obj) && courier_hook_of(obj) == COURIER_HOOK_NONE) return false; break; case T_ARRAY: for (rb_len_t i = 0; i < RARRAY_LEN(obj); i++) { if (!copy_courier_supported_p(RARRAY_AREF(obj, i), ctx)) return false; } break; case T_HASH: rb_hash_stlike_foreach(obj, copy_support_hash_i, (st_data_t)ctx); if (!ctx->ok) return false; if (!copy_courier_supported_p(RHASH_IFNONE(obj), ctx)) return false; break; case T_STRUCT: for (rb_len_t i = 0; i < RSTRUCT_LEN(obj); i++) { if (!copy_courier_supported_p(RSTRUCT_GET(obj, (int)i), ctx)) return false; } break; default: /* Anything else has to dump itself. What the hook returns is not walked here: * running it twice is not an option, so capture allocates its nodes through the * growth path instead of the reservation. */ if (courier_hook_of(obj) == COURIER_HOOK_NONE) return false; break; } rb_ivar_foreach(obj, copy_support_ivar_i, (st_data_t)ctx); return ctx->ok; } /* Build a courier holding a copy of obj's graph, leaving the sources untouched. * Returns NULL when the graph has a type it cannot carry. */ struct rb_ractor_courier * rb_ractor_courier_build_copy(VALUE obj, struct rb_ractor_courier **slot) { struct copy_support_ctx scan = { st_init_numtable(), 0, 0, true }; { bool ok = copy_courier_supported_p(obj, &scan); st_free_table(scan.seen); if (!ok) return NULL; } struct rb_ractor_courier *c = ZALLOC(struct rb_ractor_courier); courier_reserve(c, scan.nodes, scan.refs); c->seen = st_init_numtable(); struct courier_build b = { c, true }; /* Publish it into the caller's basket before capturing anything: from here the * shareable payloads it collects are rooted by the basket's holder. */ *slot = c; enum ruby_tag_type state; rb_execution_context_t *ec = GET_EC(); EC_PUSH_TAG(ec); if ((state = EC_EXEC_TAG()) == TAG_NONE) { c->root = courier_capture(&b, obj); } EC_POP_TAG(); st_free_table(c->seen); c->seen = NULL; /* Published above, so the basket owns it even half-built: it frees it. */ if (state != TAG_NONE) EC_JUMP_TAG(ec, state); return c; } /* Build a courier from obj and turn every captured source into a RactorMovedObject * (move semantics). Returns the xmalloc'd courier. */ struct rb_ractor_courier * rb_ractor_courier_build_move(VALUE obj, struct rb_ractor_courier **slot) { /* Two phases, preflight then commit, so an unmovable object is raised from the * read-only walk while the graph is still intact. */ struct move_preflight_ctx scan = { st_init_numtable(), 0, 0 }; { enum ruby_tag_type state; rb_execution_context_t *ec = GET_EC(); EC_PUSH_TAG(ec); if ((state = EC_EXEC_TAG()) == TAG_NONE) { move_preflight(obj, &scan); } EC_POP_TAG(); st_free_table(scan.seen); if (state != TAG_NONE) EC_JUMP_TAG(ec, state); } struct rb_ractor_courier *c = ZALLOC(struct rb_ractor_courier); courier_reserve(c, scan.nodes, scan.refs); c->seen = st_init_numtable(); struct courier_build b = { c, false }; /* Publish it into the caller's basket before the sources become T_MOVED: from here * the basket's holder roots what the courier carries, and partial nodes are * initialized mark-safe. */ *slot = c; enum ruby_tag_type state; rb_execution_context_t *ec = GET_EC(); EC_PUSH_TAG(ec); if ((state = EC_EXEC_TAG()) == TAG_NONE) { c->root = courier_capture(&b, obj); } EC_POP_TAG(); st_free_table(c->seen); c->seen = NULL; if (state != TAG_NONE) { /* courier_capture raised (an unmovable type, an interrupt). The courier belongs * to the basket from the publish above, so leave it there and re-raise: the * basket frees it, once, on the way out. */ EC_JUMP_TAG(ec, state); } return c; } /* Shells are created with the base/real class, so re-attach the original subclass or * singleton class (classes are shareable; the reference is safe). A singleton's * attached object still points at the sender's source: re-attach it to the shell. */ static void courier_apply_klass(VALUE shell, VALUE klass) { if (klass != RBASIC_CLASS(shell)) { RBASIC_SET_CLASS(shell, klass); } if (RB_UNLIKELY(FL_TEST_RAW(klass, FL_SINGLETON))) { rb_singleton_class_attached(klass, shell); /* the singleton class follows its object, which is now this Ractor's */ rb_class_take_ownership(klass); } } /* Rebuild the courier's graph in the current Ractor's objspace and return its root. * Two passes (allocate shells, then fill) break reference cycles. */ VALUE rb_ractor_courier_materialize(struct rb_ractor_courier *c) { /* A hidden Array roots every shell, keeping them alive while the allocations that * build the rest of the graph (which can start this Ractor's GC) run. */ VALUE shells = rb_ary_hidden_new(c->count); for (uint32_t i = 0; i < c->count; i++) { struct courier_node *n = &c->nodes[i]; VALUE shell; switch (n->kind) { case COURIER_KIND_REF: shell = n->u.ref; break; case COURIER_KIND_STRING: /* Hand the courier's buffer to the String instead of copying it again: the * bytes were already copied (or taken from the source) when the node was * built. */ shell = rb_str_new_owned(n->u.str.ptr, n->u.str.len, n->u.str.capa, n->u.str.encidx); n->u.str.ptr = NULL; /* consumed: the new String owns it now */ courier_apply_klass(shell, n->u.str.klass); break; case COURIER_KIND_ARRAY: shell = rb_ary_new_capa(n->u.ary.len); courier_apply_klass(shell, n->u.ary.klass); break; case COURIER_KIND_HASH: shell = n->u.hash.compare_by_id ? rb_ident_hash_new() : rb_hash_new(); courier_apply_klass(shell, n->u.hash.klass); break; case COURIER_KIND_OBJECT: /* A singleton class cannot allocate, so make an instance of the real class * and re-attach it afterwards */ shell = rb_obj_alloc(rb_class_real(n->u.obj.klass)); courier_apply_klass(shell, n->u.obj.klass); break; case COURIER_KIND_STRUCT: shell = rb_obj_alloc(rb_class_real(n->u.strct.klass)); courier_apply_klass(shell, n->u.strct.klass); break; case COURIER_KIND_MATCH: shell = rb_match_blob_alloc(rb_class_real(n->u.match.klass), n->u.match.num_regs); courier_apply_klass(shell, n->u.match.klass); break; case COURIER_KIND_BACKTRACE: shell = rb_backtrace_blob_load(n->u.bt.blob, n->u.bt.size); break; case COURIER_KIND_REGEXP: /* Allocated as its real class up front, as Marshal does: initializing a * plain Regexp freezes it, and the freeze pass below decides that here. */ shell = rb_reg_init_str(rb_reg_s_alloc(rb_class_real(n->u.re.klass)), n->u.re.src, n->u.re.options); courier_apply_klass(shell, n->u.re.klass); break; case COURIER_KIND_HOOKED: if (n->u.hooked.hook == COURIER_HOOK_DUMP) { shell = Qnil; /* klass._load makes it below, once its String exists */ break; } /* Allocated now and filled by its load hook below, which is what lets a * cycle back through the payload resolve to the object itself. */ shell = rb_obj_alloc(rb_class_real(n->u.hooked.klass)); courier_apply_klass(shell, n->u.hooked.klass); break; case COURIER_KIND_IO: shell = rb_obj_alloc(rb_class_real(n->u.io.klass)); courier_apply_klass(shell, n->u.io.klass); RFILE(shell)->fptr = n->u.io.fptr; n->u.io.fptr->self = shell; n->u.io.fptr = NULL; /* consumed: the new IO owns it now */ break; default: rb_bug("rb_ractor_courier_materialize: bad node kind"); } rb_ary_push(shells, shell); } /* Fill in capture's post-order, so each node is settled after everything below it, * shared children included: a Hash sees complete keys (a content-based #hash would * collide on every key while the graph is still empty), a load hook sees a complete * payload, and a parent sees the object klass._load returned. Only a cycle reaches * a node still being filled (a #hash or a payload cycling through itself is out of * scope). */ for (uint32_t k = 0; k < c->count; k++) { uint32_t i = c->order[k]; struct courier_node *n = &c->nodes[i]; VALUE shell = RARRAY_AREF(shells, i); switch (n->kind) { case COURIER_KIND_ARRAY: { /* The length is known, so set it once and write the slots, rather than * pushing each element through the capacity check. */ const rb_len_t len = n->u.ary.len; if (len > 0) { rb_ary_resize(shell, len); for (rb_len_t j = 0; j < len; j++) { RARRAY_ASET(shell, j, courier_child(c, shells, n->u.ary.elems[j])); } } break; } case COURIER_KIND_HASH: for (long j = 0; j < n->u.hash.size; j++) { rb_hash_aset(shell, courier_child(c, shells, n->u.hash.kv[2 * j]), courier_child(c, shells, n->u.hash.kv[2 * j + 1])); } /* Restore the default value and default proc (before freezing) */ VALUE ifnone = courier_child(c, shells, n->u.hash.ifnone_id); if (n->u.hash.proc_default) { rb_hash_set_default_proc(shell, ifnone); } else if (ifnone != Qnil) { rb_hash_set_default(shell, ifnone); } break; case COURIER_KIND_HOOKED: { VALUE payload = courier_child(c, shells, n->u.hooked.payload_id); VALUE klass = n->u.hooked.klass; ID mid; switch (n->u.hooked.hook) { case COURIER_HOOK_DUMP: if (!rb_obj_respond_to(klass, id_load, TRUE)) { rb_raise(rb_eTypeError, "class %"PRIsVALUE" needs to have method '_load'", klass); } /* _load returns the object: it takes the place of the Qnil placeholder * so everything filled after this receives it, and the ivars restored * below land on it. */ shell = rb_funcallv(klass, id_load, 1, &payload); RARRAY_ASET(shells, i, shell); break; case COURIER_HOOK_MARSHAL_DUMP: case COURIER_HOOK_DUMP_DATA: mid = n->u.hooked.hook == COURIER_HOOK_MARSHAL_DUMP ? id_marshal_load : id_load_data; if (!rb_obj_respond_to(shell, mid, TRUE)) { rb_raise(rb_eTypeError, "instance of %"PRIsVALUE" needs to have method '%"PRIsVALUE"'", klass, rb_id2str(mid)); } rb_funcallv(shell, mid, 1, &payload); break; case COURIER_HOOK_COMPAT: { VALUE (*loader)(VALUE, VALUE); rb_marshal_compat_lookup(klass, NULL, &loader); loader(shell, payload); break; } default: rb_bug("rb_ractor_courier_materialize: no dump protocol"); } break; } case COURIER_KIND_STRUCT: for (rb_len_t j = 0; j < n->u.strct.len; j++) { RSTRUCT_SET(shell, (int)j, courier_child(c, shells, n->u.strct.elems[j])); } break; case COURIER_KIND_MATCH: rb_match_blob_load(shell, courier_child(c, shells, n->u.match.regexp_id), courier_child(c, shells, n->u.match.str_id), n->u.match.num_regs, n->u.match.regs); break; case COURIER_KIND_IO: { /* Write the rebuilt VALUE members back into fptr (capture detached them). * write_lock and wakeup_mutex stay nil; io.c recreates them lazily. */ struct rb_io *fptr = RFILE(shell)->fptr; RB_OBJ_WRITE(shell, &fptr->pathv, courier_child(c, shells, n->u.io.pathv_id)); RB_OBJ_WRITE(shell, &fptr->encs.ecopts, courier_child(c, shells, n->u.io.ecopts_id)); RB_OBJ_WRITE(shell, &fptr->writeconv_pre_ecopts, courier_child(c, shells, n->u.io.wc_pre_ecopts_id)); RB_OBJ_WRITE(shell, &fptr->writeconv_asciicompat, courier_child(c, shells, n->u.io.wc_asciicompat_id)); RB_OBJ_WRITE(shell, &fptr->timeout, courier_child(c, shells, n->u.io.timeout_id)); break; } default: break; } /* Restore instance and generic ivars (any non-REF node can have them) */ for (uint32_t j = 0; j < n->niv; j++) { rb_ivar_set(shell, n->iv_ids[j], courier_child(c, shells, n->iv_vals[j])); } } /* Freeze after filling, so frozen containers and strings can be built too. */ for (uint32_t i = 0; i < c->count; i++) { VALUE shell = RARRAY_AREF(shells, i); if (c->nodes[i].frozen && !RB_SPECIAL_CONST_P(shell)) { rb_obj_freeze(shell); } } VALUE root = (c->count || c->refs_count) ? courier_child(c, shells, c->root) : Qnil; RB_GC_GUARD(shells); return root; } void rb_ractor_courier_free(struct rb_ractor_courier *c) { for (uint32_t i = 0; i < c->count; i++) { struct courier_node *n = &c->nodes[i]; ruby_xfree(n->iv_ids); ruby_xfree(n->iv_vals); switch (n->kind) { case COURIER_KIND_STRING: ruby_xfree(n->u.str.ptr); break; case COURIER_KIND_ARRAY: ruby_xfree(n->u.ary.elems); break; case COURIER_KIND_HASH: ruby_xfree(n->u.hash.kv); break; case COURIER_KIND_STRUCT: ruby_xfree(n->u.strct.elems); break; case COURIER_KIND_MATCH: rb_match_blob_free(n->u.match.regs); break; case COURIER_KIND_BACKTRACE: ruby_xfree(n->u.bt.blob); break; case COURIER_KIND_IO: /* A delivered IO left fptr == NULL (the rebuilt IO owns it). An * undelivered one still owns the fd and its source is already a * RactorMovedObject nobody can close: close it here, not leak it. */ if (n->u.io.fptr) { rb_io_fptr_finalize(n->u.io.fptr); n->u.io.fptr = NULL; } break; default: break; } } ruby_xfree(c->nodes); ruby_xfree(c->order); ruby_xfree(c->refs); ruby_xfree(c); } /* Mark the only VALUEs a courier holds: shareable objects and immediates (REF) and the * classes of its objects. All of them are shareable, so marking cannot race, and the * global GC keeps them reachable through the courier. While it is being built it also * holds the sender's sources in seen; the basket is on the sender's own list then. */ void rb_ractor_courier_mark(const struct rb_gc_mark_ctx *ctx, struct rb_ractor_courier *c) { if (!c) return; if (c->seen) rb_mark_set(c->seen); for (uint32_t i = 0; i < c->refs_count; i++) { rb_gc_mark_ctx(ctx, c->refs[i]); } for (uint32_t i = 0; i < c->count; i++) { struct courier_node *n = &c->nodes[i]; if (n->kind == COURIER_KIND_REF) { rb_gc_mark_ctx(ctx, n->u.ref); } else if (n->kind == COURIER_KIND_OBJECT) { rb_gc_mark_ctx(ctx, n->u.obj.klass); } else if (n->kind == COURIER_KIND_STRUCT) { rb_gc_mark_ctx(ctx, n->u.strct.klass); } else if (n->kind == COURIER_KIND_MATCH) { rb_gc_mark_ctx(ctx, n->u.match.klass); } else if (n->kind == COURIER_KIND_IO) { rb_gc_mark_ctx(ctx, n->u.io.klass); } else if (n->kind == COURIER_KIND_STRING) { rb_gc_mark_ctx(ctx, n->u.str.klass); } else if (n->kind == COURIER_KIND_BACKTRACE) { rb_backtrace_blob_mark_ctx(ctx, n->u.bt.blob, n->u.bt.size); } else if (n->kind == COURIER_KIND_ARRAY) { rb_gc_mark_ctx(ctx, n->u.ary.klass); } else if (n->kind == COURIER_KIND_HASH) { rb_gc_mark_ctx(ctx, n->u.hash.klass); } else if (n->kind == COURIER_KIND_REGEXP) { rb_gc_mark_ctx(ctx, n->u.re.src); rb_gc_mark_ctx(ctx, n->u.re.klass); } else if (n->kind == COURIER_KIND_HOOKED) { rb_gc_mark_ctx(ctx, n->u.hooked.klass); } } } /* The message copy traversal never calls #clone or #initialize_clone. Core container * types get a native shallow copy here (the traversal then rewrites the children inside * the copy); any other unshareable type falls back to a full Marshal round trip. */ static VALUE ractor_native_shallow_copy(VALUE obj) { VALUE copy; /* An object with a singleton class cannot be copied natively; fall back to Marshal * so it reports a proper error. */ VALUE klass = RBASIC_CLASS(obj); if (klass == 0 || FL_TEST_RAW(klass, FL_SINGLETON)) { return Qundef; } switch (BUILTIN_TYPE(obj)) { case T_OBJECT: copy = rb_obj_alloc(rb_obj_class(obj)); rb_obj_copy_ivar(copy, obj); break; case T_STRING: copy = rb_enc_str_new(RSTRING_PTR(obj), RSTRING_LEN(obj), rb_enc_get(obj)); break; case T_ARRAY: copy = rb_ary_new_from_values(RARRAY_LEN(obj), RARRAY_CONST_PTR(obj)); break; case T_HASH: copy = rb_hash_dup(obj); break; case T_STRUCT: copy = rb_obj_alloc(rb_obj_class(obj)); rb_struct_init_copy(copy, obj); break; case T_MATCH: copy = rb_obj_alloc(rb_obj_class(obj)); rb_match_init_copy(copy, obj); break; case T_DATA: /* Keep a copied exception from carrying a raw pointer to the sender's backtrace * across objspaces */ if (rb_backtrace_p(obj)) { copy = rb_backtrace_dup(obj); break; } return Qundef; default: return Qundef; } /* A non-T_OBJECT host keeps its ivars in the generic fields table: copy them. * T_HASH is excluded: rb_hash_dup already ran rb_copy_generic_ivar, and a second * call asserts in rb_shape_rebuild (the first gave the copy an ivar shape). */ if (BUILTIN_TYPE(obj) != T_OBJECT && BUILTIN_TYPE(obj) != T_HASH && UNLIKELY(rb_obj_gen_fields_p(obj))) { rb_copy_generic_ivar(copy, obj); } /* The traversal rewrites the children inside the copy with raw stores, so the frozen * bit can be set now: by the time leave runs the original is out of sight. The shape * has to be transitioned along with the flag, because field writes are refused based * on the shape (see rb_check_ivar_modifiable). */ if (OBJ_FROZEN(obj)) { RB_FL_SET_RAW(copy, RUBY_FL_FREEZE); RBASIC_SET_SHAPE_ID(copy, rb_obj_shape_transition_frozen(copy)); } return copy; } static enum obj_traverse_iterator_result copy_enter(VALUE obj, struct obj_traverse_replace_data *data) { if (rb_ractor_shareable_p(obj)) { data->replacement = obj; return traverse_skip; } else { VALUE copy = ractor_native_shallow_copy(obj); if (UNDEF_P(copy)) return traverse_stop; /* no native copy for this type */ data->replacement = copy; return traverse_cont; } } static enum obj_traverse_iterator_result copy_leave(VALUE obj, struct obj_traverse_replace_data *data) { return traverse_cont; } /* Native deep copy of obj's graph. Returns Qundef when it contains a type the native * copier does not support, and the caller falls back to Marshal. */ static VALUE ractor_copy_native_try(VALUE obj) { return rb_obj_traverse_replace(obj, copy_enter, copy_leave, false); } /* Deep copy within one objspace (Ractor.make_shareable(obj, copy: true)): native first, * then a whole-graph Marshal round trip. */ static VALUE ractor_copy(VALUE obj) { VALUE copy = ractor_copy_native_try(obj); if (UNDEF_P(copy)) { copy = rb_marshal_load(rb_rescue2(ractor_marshal_dump_body, obj, ractor_marshal_dump_rescue, obj, rb_eTypeError, (VALUE)0)); } return copy; } // Ractor local storage struct rb_ractor_local_key_struct { const struct rb_ractor_local_storage_type *type; void *main_cache; }; static struct freed_ractor_local_keys_struct { int cnt; int capa; rb_ractor_local_key_t *keys; } freed_ractor_local_keys; /* Purge deleted ractor-local keys from the storage tables and run their free hooks. */ static void ractor_local_keys_purge(st_table *local_storage) { for (int i=0; itype->free) { (*key->type->free)((void *)val); } } } static int ractor_local_storage_mark_i(st_data_t key, st_data_t val, st_data_t dmy) { struct rb_ractor_local_key_struct *k = (struct rb_ractor_local_key_struct *)key; if (k->type->mark) (*k->type->mark)((void *)val); return ST_CONTINUE; } static enum rb_id_table_iterator_result idkey_local_storage_mark_i(VALUE val, void *dmy) { rb_gc_mark(val); return ID_TABLE_CONTINUE; } static void ractor_local_storage_mark(rb_ractor_t *r) { if (r->local_storage) { st_foreach(r->local_storage, ractor_local_storage_mark_i, 0); /* A deleted key is purged from every Ractor's storage in one collection, which * then frees its struct. Only a collection that visits every Ractor with no * other marker running can do that: a global GC, or a single objspace. */ if (rb_gc_single_objspace_p() || rb_gc_during_global_gc_p()) { ractor_local_keys_purge(r->local_storage); } } if (r->idkey_local_storage) { rb_id_table_foreach_values(r->idkey_local_storage, idkey_local_storage_mark_i, NULL); } rb_gc_mark(r->local_storage_store_lock); } static int ractor_local_storage_free_i(st_data_t key, st_data_t val, st_data_t dmy) { struct rb_ractor_local_key_struct *k = (struct rb_ractor_local_key_struct *)key; if (k->type->free) (*k->type->free)((void *)val); return ST_CONTINUE; } static void ractor_local_storage_free(rb_ractor_t *r) { if (r->local_storage) { st_foreach(r->local_storage, ractor_local_storage_free_i, 0); st_free_table(r->local_storage); } if (r->idkey_local_storage) { rb_id_table_free(r->idkey_local_storage); } } static void rb_ractor_local_storage_value_mark(void *ptr) { rb_gc_mark((VALUE)ptr); } static const struct rb_ractor_local_storage_type ractor_local_storage_type_null = { NULL, NULL, }; const struct rb_ractor_local_storage_type rb_ractor_local_storage_type_free = { NULL, ruby_xfree, }; static const struct rb_ractor_local_storage_type ractor_local_storage_type_value = { rb_ractor_local_storage_value_mark, NULL, }; rb_ractor_local_key_t rb_ractor_local_storage_ptr_newkey(const struct rb_ractor_local_storage_type *type) { rb_ractor_local_key_t key = ALLOC(struct rb_ractor_local_key_struct); key->type = type ? type : &ractor_local_storage_type_null; key->main_cache = (void *)Qundef; return key; } rb_ractor_local_key_t rb_ractor_local_storage_value_newkey(void) { return rb_ractor_local_storage_ptr_newkey(&ractor_local_storage_type_value); } void rb_ractor_local_storage_delkey(rb_ractor_local_key_t key) { RB_VM_LOCKING() { if (freed_ractor_local_keys.cnt == freed_ractor_local_keys.capa) { freed_ractor_local_keys.capa = freed_ractor_local_keys.capa ? freed_ractor_local_keys.capa * 2 : 4; SIZED_REALLOC_N(freed_ractor_local_keys.keys, rb_ractor_local_key_t, freed_ractor_local_keys.capa, freed_ractor_local_keys.cnt); } freed_ractor_local_keys.keys[freed_ractor_local_keys.cnt++] = key; } } static bool ractor_local_ref(rb_ractor_local_key_t key, void **pret) { if (rb_ractor_main_p()) { if (!UNDEF_P((VALUE)key->main_cache)) { *pret = key->main_cache; return true; } else { return false; } } else { rb_ractor_t *cr = GET_RACTOR(); if (cr->local_storage && st_lookup(cr->local_storage, (st_data_t)key, (st_data_t *)pret)) { return true; } else { return false; } } } static void ractor_local_set(rb_ractor_local_key_t key, void *ptr) { rb_ractor_t *cr = GET_RACTOR(); if (cr->local_storage == NULL) { cr->local_storage = st_init_numtable(); } st_insert(cr->local_storage, (st_data_t)key, (st_data_t)ptr); if (rb_ractor_main_p()) { key->main_cache = ptr; } } VALUE rb_ractor_local_storage_value(rb_ractor_local_key_t key) { void *val; if (ractor_local_ref(key, &val)) { return (VALUE)val; } else { return Qnil; } } bool rb_ractor_local_storage_value_lookup(rb_ractor_local_key_t key, VALUE *val) { if (ractor_local_ref(key, (void **)val)) { return true; } else { return false; } } void rb_ractor_local_storage_value_set(rb_ractor_local_key_t key, VALUE val) { ractor_local_set(key, (void *)val); } void * rb_ractor_local_storage_ptr(rb_ractor_local_key_t key) { void *ret; if (ractor_local_ref(key, &ret)) { return ret; } else { return NULL; } } void rb_ractor_local_storage_ptr_set(rb_ractor_local_key_t key, void *ptr) { ractor_local_set(key, ptr); } #define DEFAULT_KEYS_CAPA 0x10 void rb_ractor_finish_marking(bool full_mark) { /* A freed key's struct may only be released by a collection that purged every * Ractor's storage with no other marker running: a global GC, or a single objspace. * A local GC also reaches here (gc_marks_finish) and must do nothing. */ if (!(rb_gc_single_objspace_p() || rb_gc_during_global_gc_p())) { return; } /* The root scan's purge never reaches a zombie's storage (not in the set; * zombie_objspaces only marks the join slot): purge here, under the barrier, before * the struct is freed, or a later ractor_free reads a freed key. */ rb_vm_t *vm = GET_VM(); rb_ractor_t *r; for (size_t zi = 0; zi < vm->gc.zombie_objspaces_count; zi++) { rb_ractor_t *owner = vm->gc.zombie_objspaces[zi].owner; if (owner == NULL || owner->local_storage == NULL) continue; ractor_local_keys_purge(owner->local_storage); } for (int i=0; i DEFAULT_KEYS_CAPA) { freed_ractor_local_keys.capa = DEFAULT_KEYS_CAPA; SIZED_REALLOC_N(freed_ractor_local_keys.keys, rb_ractor_local_key_t, DEFAULT_KEYS_CAPA, freed_ractor_local_keys.capa); } /* Under a minor mark an unmarked port is not a dead one. */ if (full_mark) { ccan_list_for_each(&vm->ractor.set, r, vmlr_node) { rb_ractor_reap_dead_ports(r); } if (vm->ractor.cnt == 0 && vm->ractor.main_ractor) { rb_ractor_reap_dead_ports(vm->ractor.main_ractor); } } } static VALUE ractor_local_value(rb_execution_context_t *ec, VALUE self, VALUE sym) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); ID id = rb_check_id(&sym); struct rb_id_table *tbl = cr->idkey_local_storage; VALUE val; if (id && tbl && rb_id_table_lookup(tbl, id, &val)) { return val; } else { return Qnil; } } static VALUE ractor_local_value_set(rb_execution_context_t *ec, VALUE self, VALUE sym, VALUE val) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); ID id = SYM2ID(rb_to_symbol(sym)); struct rb_id_table *tbl = cr->idkey_local_storage; if (tbl == NULL) { tbl = cr->idkey_local_storage = rb_id_table_create(2); } rb_id_table_insert(tbl, id, val); return val; } struct ractor_local_storage_store_data { rb_execution_context_t *ec; struct rb_id_table *tbl; ID id; VALUE sym; }; static VALUE ractor_local_value_store_i(VALUE ptr) { VALUE val; struct ractor_local_storage_store_data *data = (struct ractor_local_storage_store_data *)ptr; if (rb_id_table_lookup(data->tbl, data->id, &val)) { // after synchronization, we found already registered entry } else { val = rb_yield(Qnil); ractor_local_value_set(data->ec, Qnil, data->sym, val); } return val; } static VALUE ractor_local_value_store_if_absent(rb_execution_context_t *ec, VALUE self, VALUE sym) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); struct ractor_local_storage_store_data data = { .ec = ec, .sym = sym, .id = SYM2ID(rb_to_symbol(sym)), .tbl = cr->idkey_local_storage, }; VALUE val; if (data.tbl == NULL) { data.tbl = cr->idkey_local_storage = rb_id_table_create(2); } else if (rb_id_table_lookup(data.tbl, data.id, &val)) { // already set return val; } if (!cr->local_storage_store_lock) { cr->local_storage_store_lock = rb_mutex_new(); } return rb_mutex_synchronize(cr->local_storage_store_lock, ractor_local_value_store_i, (VALUE)&data); } // shareable_proc static VALUE ractor_shareable_proc(rb_execution_context_t *ec, VALUE replace_self, bool is_lambda) { if (!rb_ractor_shareable_p(replace_self)) { rb_raise(rb_eRactorIsolationError, "self should be shareable: %" PRIsVALUE, replace_self); } else { VALUE proc = is_lambda ? rb_block_lambda() : rb_block_proc(); return rb_proc_ractor_make_shareable(rb_proc_dup(proc), replace_self); } } // Ractor#require struct cross_ractor_require { VALUE port; bool raised; union { struct { VALUE feature; } require; struct { VALUE module; ID name; } autoload; } as; bool silent; }; RUBY_REFERENCES(cross_ractor_require_refs) = { RUBY_REF_EDGE(struct cross_ractor_require, port), RUBY_REF_EDGE(struct cross_ractor_require, as.require.feature), RUBY_REF_END }; static const rb_data_type_t cross_ractor_require_data_type = { "ractor/cross_ractor_require", { RUBY_REFS_LIST_PTR(cross_ractor_require_refs), RUBY_DEFAULT_FREE, NULL, // memsize NULL, // compact }, 0, 0, RUBY_TYPED_THREAD_SAFE_FREE | RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_DECL_MARKING | RUBY_TYPED_EMBEDDABLE }; static VALUE require_body(VALUE crr_obj) { struct cross_ractor_require *crr; TypedData_Get_Struct(crr_obj, struct cross_ractor_require, &cross_ractor_require_data_type, crr); VALUE feature = crr->as.require.feature; ID require; CONST_ID(require, "require"); if (crr->silent) { int rb_require_internal_silent(VALUE fname); return INT2NUM(rb_require_internal_silent(feature)); } else { return rb_funcallv(Qnil, require, 1, &feature); } } static VALUE require_rescue(VALUE crr_obj, VALUE errinfo) { struct cross_ractor_require *crr; TypedData_Get_Struct(crr_obj, struct cross_ractor_require, &cross_ractor_require_data_type, crr); crr->raised = true; return errinfo; } static VALUE require_result_send_body(VALUE ary) { VALUE port = RARRAY_AREF(ary, 0); VALUE results = RARRAY_AREF(ary, 1); rb_execution_context_t *ec = GET_EC(); ractor_port_send(ec, port, results, Qfalse); return Qnil; } static VALUE require_result_send_resuce(VALUE port, VALUE errinfo) { // TODO: need rescue? ractor_port_send(GET_EC(), port, errinfo, Qfalse); return Qnil; } static VALUE ractor_require_protect(VALUE crr_obj, VALUE (*func)(VALUE)) { struct cross_ractor_require *crr; TypedData_Get_Struct(crr_obj, struct cross_ractor_require, &cross_ractor_require_data_type, crr); const bool silent = crr->silent; VALUE debug, errinfo; if (silent) { debug = ruby_debug; errinfo = rb_errinfo(); } // get normal result or raised exception (with crr->raised == true) VALUE result = rb_rescue2(func, crr_obj, require_rescue, crr_obj, rb_eException, 0); if (silent) { ruby_debug = debug; rb_set_errinfo(errinfo); } rb_rescue2(require_result_send_body, // [port, [result, raised]] rb_ary_new_from_args(2, crr->port, rb_ary_new_from_args(2, result, crr->raised ? Qtrue : Qfalse)), require_result_send_resuce, rb_eException, crr->port); RB_GC_GUARD(crr_obj); return Qnil; } static VALUE ractor_require_func(void *crr_obj) { return ractor_require_protect((VALUE)crr_obj, require_body); } VALUE rb_ractor_require(VALUE feature, bool silent) { // We're about to block on the main ractor, so if we're holding the global lock we'll deadlock. ASSERT_vm_unlocking(); struct cross_ractor_require *crr; VALUE crr_obj = TypedData_Make_Struct(0, struct cross_ractor_require, &cross_ractor_require_data_type, crr); RB_OBJ_SET_SHAREABLE(crr_obj); // TODO: internal data? // Convert feature to proper file path and make it shareable as fstring RB_OBJ_WRITE(crr_obj, &crr->as.require.feature, rb_fstring(FilePathValue(feature))); RB_OBJ_WRITE(crr_obj, &crr->port, rb_ractor_make_shareable(ractor_port_new(GET_RACTOR()))); crr->raised = false; crr->silent = silent; rb_execution_context_t *ec = GET_EC(); rb_ractor_t *main_r = GET_VM()->ractor.main_ractor; rb_ractor_interrupt_exec(main_r, ractor_require_func, (void *)crr_obj, rb_interrupt_exec_flag_value_data); // wait for require done VALUE results = ractor_port_receive(ec, crr->port, Qnil); ractor_port_close(ec, crr->port); VALUE exc = rb_ary_pop(results); VALUE result = rb_ary_pop(results); RB_GC_GUARD(crr_obj); if (RTEST(exc)) { rb_exc_raise(result); } else { return result; } } static VALUE ractor_require(rb_execution_context_t *ec, VALUE self, VALUE feature) { return rb_ractor_require(feature, false); } static VALUE autoload_load_body(VALUE crr_obj) { struct cross_ractor_require *crr; TypedData_Get_Struct(crr_obj, struct cross_ractor_require, &cross_ractor_require_data_type, crr); return rb_autoload_load(crr->as.autoload.module, crr->as.autoload.name); } static VALUE ractor_autoload_load_func(void *crr_obj) { return ractor_require_protect((VALUE)crr_obj, autoload_load_body); } VALUE rb_ractor_autoload_load(VALUE module, ID name) { struct cross_ractor_require *crr; VALUE crr_obj = TypedData_Make_Struct(0, struct cross_ractor_require, &cross_ractor_require_data_type, crr); RB_OBJ_SET_SHAREABLE(crr_obj); // TODO: internal data? RB_OBJ_WRITE(crr_obj, &crr->as.autoload.module, module); RB_OBJ_WRITE(crr_obj, &crr->as.autoload.name, name); RB_OBJ_WRITE(crr_obj, &crr->port, rb_ractor_make_shareable(ractor_port_new(GET_RACTOR()))); rb_execution_context_t *ec = GET_EC(); rb_ractor_t *main_r = GET_VM()->ractor.main_ractor; rb_ractor_interrupt_exec(main_r, ractor_autoload_load_func, (void *)crr_obj, rb_interrupt_exec_flag_value_data); // wait for require done VALUE results = ractor_port_receive(ec, crr->port, Qnil); ractor_port_close(ec, crr->port); VALUE exc = rb_ary_pop(results); VALUE result = rb_ary_pop(results); RB_GC_GUARD(crr_obj); if (RTEST(exc)) { rb_exc_raise(result); } else { return result; } } VALUE rb_builtin_shareable_proc(rb_execution_context_t *ec, VALUE self, VALUE arg_self) { return ractor_shareable_proc(ec, arg_self, false); } VALUE rb_builtin_shareable_lambda(rb_execution_context_t *ec, VALUE self, VALUE arg_self) { return ractor_shareable_proc(ec, arg_self, true); } #include "ractor.rbinc"