// this file is included by ractor.c struct ractor_port { rb_ractor_t *r; st_data_t id_; }; static st_data_t ractor_port_id(const struct ractor_port *rp) { return rp->id_; } static VALUE rb_cRactorPort; static VALUE ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp, const rb_hrtime_t *end); static VALUE ractor_send(rb_execution_context_t *ec, const struct ractor_port *rp, VALUE obj, VALUE move); static struct ractor_basket *ractor_basket_new_exit(VALUE sender, VALUE token); static void ractor_send_basket(rb_execution_context_t *ec, const struct ractor_port *rp, struct ractor_basket *b, bool raise_on_error); static void ractor_add_port(rb_ractor_t *r, st_data_t id); // The off-heap courier a copy or a move payload travels in. Defined in ractor.c. struct rb_ractor_courier *rb_ractor_courier_build_move(VALUE obj, struct rb_ractor_courier **slot); VALUE rb_ractor_courier_materialize(struct rb_ractor_courier *c); void rb_ractor_courier_free(struct rb_ractor_courier *c); static void ractor_off_queue_add(rb_ractor_t *cr, struct ractor_basket *b); static void ractor_off_queue_remove(struct ractor_basket *b); void rb_ractor_courier_mark(const struct rb_gc_mark_ctx *ctx, struct rb_ractor_courier *c); struct rb_ractor_courier *rb_ractor_courier_build_copy(VALUE obj, struct rb_ractor_courier **slot); static void ractor_port_note_alive(const struct ractor_port *rp); static void ractor_port_mark(void *ptr) { const struct ractor_port *rp = (struct ractor_port *)ptr; if (rp->r) { rb_gc_mark(rp->r->pub.self); /* Only a mark that covers every objspace can call a port dead. Ask the single * objspace first: it answers without the VM, which a GC worker thread cannot * reach (mmtk marks from several of them). */ if (rb_gc_single_objspace_p() || rb_gc_during_global_gc_p()) { ractor_port_note_alive(rp); } } } static const rb_data_type_t ractor_port_data_type = { "ractor/port", { ractor_port_mark, RUBY_TYPED_DEFAULT_FREE, NULL, // memsize NULL, // update }, 0, 0, RUBY_TYPED_THREAD_SAFE_FREE | RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_FROZEN_SHAREABLE | RUBY_TYPED_EMBEDDABLE, }; static st_data_t ractor_genid_for_port(rb_ractor_t *cr) { // TODO: enough? return cr->sync.next_port_id++; } static struct ractor_port * RACTOR_PORT_PTR(VALUE self) { VM_ASSERT(rb_typeddata_is_kind_of(self, &ractor_port_data_type)); return RTYPEDDATA_GET_DATA(self); } // r is NULL between Ractor::Port.allocate and ractor_port_init() static struct ractor_port * ractor_port_ptr_check(VALUE self) { struct ractor_port *rp = RACTOR_PORT_PTR(self); if (UNLIKELY(rp->r == NULL)) { rb_raise(rb_eTypeError, "uninitialized %"PRIsVALUE, rb_obj_class(self)); } return rp; } static VALUE ractor_port_alloc(VALUE klass) { struct ractor_port *rp; VALUE rpv = TypedData_Make_Struct(klass, struct ractor_port, &ractor_port_data_type, rp); rb_obj_freeze(rpv); return rpv; } static VALUE ractor_port_init(VALUE rpv, rb_ractor_t *r) { // Child threads can still run ensure blocks and report exceptions after // ractor_notify_exit has freed the ports. if (!r->sync.ports) { rb_raise(rb_eRactorClosedError, "The ractor has terminated"); } struct ractor_port *rp = RACTOR_PORT_PTR(rpv); rp->r = r; RB_OBJ_WRITTEN(rpv, Qundef, r->pub.self); rp->id_ = ractor_genid_for_port(r); ractor_add_port(r, ractor_port_id(rp)); rb_obj_freeze(rpv); return rpv; } /* * call-seq: * Ractor::Port.new -> new_port * * Returns a new Ractor::Port object. */ static VALUE ractor_port_initialize(VALUE self) { return ractor_port_init(self, GET_RACTOR()); } /* :nodoc: */ static VALUE ractor_port_initialize_copy(VALUE self, VALUE orig) { struct ractor_port *dst = RACTOR_PORT_PTR(self); // uninitialized by definition struct ractor_port *src = ractor_port_ptr_check(orig); dst->r = src->r; RB_OBJ_WRITTEN(self, Qundef, dst->r->pub.self); dst->id_ = ractor_port_id(src); return self; } static VALUE ractor_port_new(rb_ractor_t *r) { VALUE rpv = ractor_port_alloc(rb_cRactorPort); ractor_port_init(rpv, r); return rpv; } static bool ractor_port_p(VALUE self) { return rb_typeddata_is_kind_of(self, &ractor_port_data_type); } static const rb_hrtime_t *ractor_timeout_deadline(VALUE timeout, rb_hrtime_t *storage); static VALUE ractor_port_receive(rb_execution_context_t *ec, VALUE self, VALUE timeout) { const struct ractor_port *rp = ractor_port_ptr_check(self); if (rp->r != rb_ec_ractor_ptr(ec)) { rb_raise(rb_eRactorError, "only allowed from the creator Ractor of this port"); } rb_hrtime_t deadline; const rb_hrtime_t *end = ractor_timeout_deadline(timeout, &deadline); VALUE v = ractor_receive(ec, rp, end); RB_GC_GUARD(self); // no message before the timeout return UNDEF_P(v) ? Qnil : v; } static VALUE ractor_port_send(rb_execution_context_t *ec, VALUE self, VALUE obj, VALUE move) { const struct ractor_port *rp = ractor_port_ptr_check(self); ractor_send(ec, rp, obj, RTEST(move)); RB_GC_GUARD(self); return self; } static bool ractor_closed_port_p(rb_execution_context_t *ec, rb_ractor_t *r, const struct ractor_port *rp); static bool ractor_close_port(rb_execution_context_t *ec, rb_ractor_t *r, const struct ractor_port *rp); static VALUE ractor_port_closed_p(rb_execution_context_t *ec, VALUE self) { const struct ractor_port *rp = ractor_port_ptr_check(self); rb_ractor_t *r = rp->r; bool closed; if (rb_ec_ractor_ptr(ec) == r) { /* The owner's threads are serialized by the ractor GVL, so the ports * table can't change under this lookup. */ closed = ractor_closed_port_p(ec, r, rp); } else { /* A foreign Ractor races the owner's st_insert/st_delete on the ports * table; take the lock like every other foreign reader. ractor_closed_port_p * asserts the lock is held for foreign access, and Port#closed? was the * only path reaching it without the lock. */ RACTOR_LOCK(r); { closed = ractor_closed_port_p(ec, r, rp); } RACTOR_UNLOCK(r); } return closed ? Qtrue : Qfalse; } static VALUE ractor_port_close(rb_execution_context_t *ec, VALUE self) { const struct ractor_port *rp = ractor_port_ptr_check(self); rb_ractor_t *cr = rb_ec_ractor_ptr(ec); if (cr != rp->r) { rb_raise(rb_eRactorError, "closing port by other ractors is not allowed"); } ractor_close_port(ec, cr, rp); return self; } // ractor-internal // ractor-internal - ractor_basket enum ractor_basket_type { // basket is empty basket_type_none, // value is available basket_type_ref, basket_type_copy, basket_type_move, /* A ractor's exit token. The pair it becomes is built on the receiving side, * so a terminating ractor adds no shareable object to the vm, and building it * needs neither a tag nor a courier -- which the sender has no stack for. */ basket_type_exit, }; struct ractor_basket { VALUE sender; st_data_t port_id; /* The payload: either a VALUE, or the off-heap (xmalloc) courier its graph was * serialized into (copy and move both use one). */ union { VALUE v; struct rb_ractor_courier *courier; } p; /* The one list it is on: a port queue while it waits there, or its holder's * sync.off_queue_baskets while it is being built or materialized. */ struct ccan_list_node node; enum ractor_basket_type type : 8; bool has_courier; bool exception; }; #if 0 static inline bool ractor_basket_type_p(const struct ractor_basket *b, enum ractor_basket_type type) { return b->type == type; } static inline bool ractor_basket_none_p(const struct ractor_basket *b) { return ractor_basket_type_p(b, basket_type_none); } #endif static void ractor_basket_mark(const struct rb_gc_mark_ctx *ctx, const struct ractor_basket *b) { if (b->has_courier) { /* The payload became this Ractor's to root the moment the message was enqueued * here: the sender's own roots stop at the send. Before and after the queue the * basket is on its holder's off_queue_baskets instead, so a courier is rooted * from the moment it is allocated to the moment it is freed. */ rb_ractor_courier_mark(ctx, b->p.courier); } else { rb_gc_mark_ctx(ctx, b->p.v); } /* An exit token names a ractor that may be reachable from nothing else. */ rb_gc_mark_ctx(ctx, b->sender); } static void ractor_basket_free(struct ractor_basket *b) { ractor_off_queue_remove(b); if (b->has_courier && b->p.courier) { /* A courier that was never consumed (a queue being torn down, say). */ rb_ractor_courier_free(b->p.courier); b->p.courier = NULL; } SIZED_FREE(b); } static struct ractor_basket * ractor_basket_alloc(void) { struct ractor_basket *b = ALLOC(struct ractor_basket); /* Empty and mark-safe from the start: a basket goes on its holder's in-flight list * before it has a payload, so a GC can walk it while it is still being filled. */ b->type = basket_type_none; b->sender = Qnil; b->port_id = 0; b->p.v = Qnil; b->has_courier = false; b->exception = false; ccan_list_node_init(&b->node); return b; } /* A basket is rooted by whoever holds it: a port queue while it waits there, and its * holder's off-queue list while it is being built or materialized. */ static void ractor_off_queue_add(rb_ractor_t *cr, struct ractor_basket *b) { VM_ASSERT(cr == rb_current_ractor_raw(false)); ccan_list_add_tail(&cr->sync.off_queue_baskets, &b->node); } static void ractor_off_queue_remove(struct ractor_basket *b) { ccan_list_del_init(&b->node); } static void ractor_mark_off_queue_baskets(const struct rb_gc_mark_ctx *ctx, rb_ractor_t *r) { struct ractor_basket *b; ccan_list_for_each(&r->sync.off_queue_baskets, b, node) { ractor_basket_mark(ctx, b); } } // ractor-internal - ractor_queue struct ractor_queue { struct ccan_list_head set; bool closed; bool alive; /* its Ractor::Port is still reachable; see the reap below */ }; static void ractor_queue_init(struct ractor_queue *rq) { ccan_list_head_init(&rq->set); rq->closed = false; rq->alive = true; } static struct ractor_queue * ractor_queue_new(void) { struct ractor_queue *rq = ALLOC(struct ractor_queue); ractor_queue_init(rq); return rq; } static void ractor_port_note_alive(const struct ractor_port *rp) { struct ractor_queue *rq; if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, rp->id_, (st_data_t *)&rq)) { /* Several markers can reach the same port at once (mmtk marks from its GC worker * threads), but they all store the same value and the reap reads it once marking * is over. */ rq->alive = true; } } static void ractor_queue_mark(const struct rb_gc_mark_ctx *ctx, const struct ractor_queue *rq) { const struct ractor_basket *b; ccan_list_for_each(&rq->set, b, node) { ractor_basket_mark(ctx, b); } } static void ractor_queue_free(struct ractor_queue *rq) { struct ractor_basket *b, *nxt; ccan_list_for_each_safe(&rq->set, b, nxt, node) { ccan_list_del_init(&b->node); ractor_basket_free(b); } VM_ASSERT(ccan_list_empty(&rq->set)); SIZED_FREE(rq); } RBIMPL_ATTR_MAYBE_UNUSED() static size_t ractor_queue_size(const struct ractor_queue *rq) { size_t size = 0; const struct ractor_basket *b; ccan_list_for_each(&rq->set, b, node) { size++; } return size; } // Returns whether this call is the one that closed it. static bool ractor_queue_close(struct ractor_queue *rq) { bool closed_now = !rq->closed; rq->closed = true; return closed_now; } static void ractor_queue_move(struct ractor_queue *dst_rq, struct ractor_queue *src_rq) { struct ccan_list_head *src = &src_rq->set; struct ccan_list_head *dst = &dst_rq->set; dst->n.next = src->n.next; dst->n.prev = src->n.prev; dst->n.next->prev = &dst->n; dst->n.prev->next = &dst->n; ccan_list_head_init(src); } #if 0 static struct ractor_basket * ractor_queue_head(rb_ractor_t *r, struct ractor_queue *rq) { return ccan_list_top(&rq->set, struct ractor_basket, node); } #endif static bool ractor_queue_empty_p(rb_ractor_t *r, const struct ractor_queue *rq) { return ccan_list_empty(&rq->set); } static struct ractor_basket * ractor_queue_deq(rb_ractor_t *r, struct ractor_queue *rq) { VM_ASSERT(GET_RACTOR() == r); struct ractor_basket *b = ccan_list_pop(&rq->set, struct ractor_basket, node); if (b) ccan_list_node_init(&b->node); return b; } static void ractor_queue_enq(rb_ractor_t *r, struct ractor_queue *rq, struct ractor_basket *basket) { ccan_list_add_tail(&rq->set, &basket->node); } #if 0 static void rq_dump(const struct ractor_queue *rq) { int i=0; struct ractor_basket *b; ccan_list_for_each(&rq->set, b, node) { fprintf(stderr, "%d type:%s %p\n", i, basket_type_name(b->type), (void *)b); i++; } } #endif static void ractor_delete_port(rb_ractor_t *cr, st_data_t id, bool locked); static struct ractor_queue * ractor_get_queue(rb_ractor_t *cr, st_data_t id, bool locked) { VM_ASSERT(cr == GET_RACTOR()); struct ractor_queue *rq; if (cr->sync.ports && st_lookup(cr->sync.ports, id, (st_data_t *)&rq)) { if (rq->closed && ractor_queue_empty_p(cr, rq)) { ractor_delete_port(cr, id, locked); return NULL; } else { return rq; } } else { return NULL; } } // ractor-internal - ports static void ractor_add_port(rb_ractor_t *r, st_data_t id) { struct ractor_queue *rq = ractor_queue_new(); ASSERT_ractor_unlocking(r); RUBY_DEBUG_LOG("id:%u", (unsigned int)id); // Rebuilding the table on insertion can run GC by the allocation and the // GC acquires the VM lock, which is prohibited under the ractor lock. st_table *const old_tab = r->sync.ports; bool inserted; RACTOR_LOCK(r); { inserted = st_insert_no_rebuild(old_tab, id, (st_data_t)rq) >= 0; } RACTOR_UNLOCK(r); if (!inserted) { // The table is full. Rebuild it outside of the ractor lock (mutators // are serialized by the per-ractor GVL) and swap it under the lock // to exclude the readers (other ractors). st_table *new_tab; // Those allocations can run a global GC, whose reap frees dead ports and // drops them from old_tab; a copy taken across one would republish the // freed queues. Only the owner inserts into its own table, so a changed // count means a reap ran. Check after each allocation: st_copy fills the // header before it allocates the entries, so a reap in between leaves the // copy counting rows it does not have, which st_insert must not be given. while (1) { const st_index_t entries = st_table_size(old_tab); new_tab = st_copy(old_tab); if (st_table_size(old_tab) == entries) { st_insert(new_tab, id, (st_data_t)rq); if (st_table_size(old_tab) == entries) break; } st_free_table(new_tab); } RACTOR_LOCK(r); { VM_ASSERT(r->sync.ports == old_tab); r->sync.ports = new_tab; } RACTOR_UNLOCK(r); st_free_table(old_tab); } } static void ractor_delete_port_locked(rb_ractor_t *cr, st_data_t id) { ASSERT_ractor_locking(cr); RUBY_DEBUG_LOG("id:%u", (unsigned int)id); struct ractor_queue *rq; if (st_delete(cr->sync.ports, &id, (st_data_t *)&rq)) { ractor_queue_free(rq); } else { VM_ASSERT(0); } } static void ractor_delete_port(rb_ractor_t *cr, st_data_t id, bool locked) { if (locked) { ractor_delete_port_locked(cr, id); } else { RACTOR_LOCK_SELF(cr); { ractor_delete_port_locked(cr, id); } RACTOR_UNLOCK_SELF(cr); } } static const struct ractor_port * ractor_default_port(rb_ractor_t *r) { return RACTOR_PORT_PTR(r->sync.default_port_value); } static VALUE ractor_default_port_value(rb_ractor_t *r) { return r->sync.default_port_value; } static bool ractor_closed_port_p(rb_execution_context_t *ec, rb_ractor_t *r, const struct ractor_port *rp) { VM_ASSERT(rb_ec_ractor_ptr(ec) == rp->r ? 1 : (ASSERT_ractor_locking(rp->r), 1)); const struct ractor_queue *rq; if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, ractor_port_id(rp), (st_data_t *)&rq)) { return rq->closed; } else { return true; } } static void ractor_deliver_incoming_messages(rb_execution_context_t *ec, rb_ractor_t *cr); static bool ractor_queue_empty_p(rb_ractor_t *r, const struct ractor_queue *rq); static bool ractor_wakeup_all(rb_ractor_t *r, enum ractor_wakeup_status wakeup_status); static bool ractor_close_port(rb_execution_context_t *ec, rb_ractor_t *cr, const struct ractor_port *rp) { VM_ASSERT(cr == rp->r); struct ractor_queue *rq = NULL; bool closed_now = false; RACTOR_LOCK_SELF(cr); { ractor_deliver_incoming_messages(ec, cr); // check incoming messages if (st_lookup(rp->r->sync.ports, ractor_port_id(rp), (st_data_t *)&rq)) { closed_now = ractor_queue_close(rq); if (ractor_queue_empty_p(cr, rq)) { // delete from the table ractor_delete_port(cr, ractor_port_id(rp), true); } // TODO: free rq } } RACTOR_UNLOCK_SELF(cr); if (closed_now) { // Only when this call closed it: waking the Ractor is not free to the // other waiters, and a re-close of a closed port is news to nobody. ractor_wakeup_all(cr, wakeup_by_close); } return rq != NULL; } /* A port is the only way to receive from its queue, so a queue whose port is gone is * unreachable -- but the table is keyed by id, so no sweep finds it. Mark and sweep the * table itself: ractor_port_mark sets the flag, this clears it for the next cycle. */ static int ractor_reap_dead_ports_i(st_data_t port_id, st_data_t val, st_data_t dat) { struct ractor_queue *rq = (struct ractor_queue *)val; if (rq->alive) { rq->alive = false; return ST_CONTINUE; } else { ractor_queue_free(rq); return ST_DELETE; } } /* A message is only moved from recv_queue to its port queue when the owner receives or * closes, so one addressed to a port that died first is left here, out of the sweep * above. Its port is gone from the table by now: drop it. */ static void ractor_reap_undeliverable_messages(rb_ractor_t *r) { struct ractor_queue *recv_q = r->sync.recv_queue; if (recv_q == NULL) return; struct ractor_basket *b, *nxt; ccan_list_for_each_safe(&recv_q->set, b, nxt, node) { if (!st_lookup(r->sync.ports, b->port_id, NULL)) { ccan_list_del_init(&b->node); ractor_basket_free(b); } } } void rb_ractor_reap_dead_ports(rb_ractor_t *r) { /* No sync lock here: the caller's gate is what keeps foreign senders out. */ VM_ASSERT(rb_gc_single_objspace_p() || rb_gc_during_global_gc_p()); if (r->sync.ports) { st_foreach(r->sync.ports, ractor_reap_dead_ports_i, 0); ractor_reap_undeliverable_messages(r); } } static int ractor_free_all_ports_i(st_data_t port_id, st_data_t val, st_data_t dat) { struct ractor_queue *rq = (struct ractor_queue *)val; // rb_ractor_t *cr = (rb_ractor_t *)dat; ractor_queue_free(rq); return ST_CONTINUE; } static void ractor_free_all_ports(rb_ractor_t *cr) { if (cr->sync.ports) { st_foreach(cr->sync.ports, ractor_free_all_ports_i, (st_data_t)cr); st_free_table(cr->sync.ports); cr->sync.ports = NULL; } if (cr->sync.recv_queue) { ractor_queue_free(cr->sync.recv_queue); cr->sync.recv_queue = NULL; } } #if defined(HAVE_WORKING_FORK) static void ractor_sync_terminate_atfork(rb_vm_t *vm, rb_ractor_t *r) { ractor_free_all_ports(r); r->sync.legacy = Qnil; } #endif // Ractor#monitor struct ractor_monitor { struct ractor_port port; struct ccan_list_node node; }; /* Mark the Ractors monitoring r. ractor_notify_exit sends the exit token through each * entry's port, so the monitoring Ractor's struct must outlive r, and its wrapper is * what keeps it alive. */ static void ractor_mark_monitors(const struct rb_gc_mark_ctx *ctx, rb_ractor_t *r) { const struct ractor_monitor *rm; ccan_list_for_each(&r->sync.monitors, rm, node) { rb_gc_mark_ctx(ctx, rm->port.r->pub.self); } } /* Paired with the ractor it is about when it is received, so that many ractors * can report to one port. */ static VALUE ractor_exit_token(const rb_ractor_t *r) { if (r->sync.legacy_exc) { RUBY_DEBUG_LOG("aborted"); return ID2SYM(idAborted); } else { RUBY_DEBUG_LOG("exited"); return ID2SYM(idExited); } } static VALUE ractor_monitor(rb_execution_context_t *ec, VALUE self, VALUE port) { rb_ractor_t *r = RACTOR_PTR(self); bool terminated = false; const struct ractor_port *rp = ractor_port_ptr_check(port); struct ractor_monitor *rm = ALLOC(struct ractor_monitor); rm->port = *rp; // copy port information RACTOR_LOCK(r); { if (UNDEF_P(r->sync.legacy)) { // not terminated RUBY_DEBUG_LOG("OK/r:%u -> port:%u@r%u", (unsigned int)rb_ractor_id(r), (unsigned int)ractor_port_id(&rm->port), (unsigned int)rb_ractor_id(rm->port.r)); ccan_list_add_tail(&r->sync.monitors, &rm->node); } else { RUBY_DEBUG_LOG("NG/r:%u -> port:%u@r%u", (unsigned int)rb_ractor_id(r), (unsigned int)ractor_port_id(&rm->port), (unsigned int)rb_ractor_id(rm->port.r)); terminated = true; } } RACTOR_UNLOCK(r); if (terminated) { SIZED_FREE(rm); ractor_send_basket(ec, rp, ractor_basket_new_exit(self, ractor_exit_token(r)), false); } /* rp points into port, which is embedded and may be referenced only from here */ RB_GC_GUARD(port); return terminated ? Qfalse : Qtrue; } static VALUE ractor_alive_p(rb_execution_context_t *ec, VALUE self) { rb_ractor_t *r = RACTOR_PTR(self); bool alive; RACTOR_LOCK(r); { alive = UNDEF_P(r->sync.legacy); } RACTOR_UNLOCK(r); // rb_ractor_terminate_atfork terminates ractors without setting legacy return RBOOL(alive && !rb_ractor_status_p(r, ractor_terminated)); } static VALUE ractor_unmonitor(rb_execution_context_t *ec, VALUE self, VALUE port) { rb_ractor_t *r = RACTOR_PTR(self); const struct ractor_port *rp = ractor_port_ptr_check(port); RACTOR_LOCK(r); { if (UNDEF_P(r->sync.legacy)) { // not terminated struct ractor_monitor *rm, *nxt; ccan_list_for_each_safe(&r->sync.monitors, rm, nxt, node) { if (rm->port.r == rp->r && ractor_port_id(&rm->port) == ractor_port_id(rp)) { RUBY_DEBUG_LOG("r:%u -> port:%u@r%u", (unsigned int)rb_ractor_id(r), (unsigned int)ractor_port_id(&rm->port), (unsigned int)rb_ractor_id(rm->port.r)); ccan_list_del(&rm->node); SIZED_FREE(rm); } } } } RACTOR_UNLOCK(r); return self; } static void ractor_notify_exit(rb_execution_context_t *ec, rb_ractor_t *cr, VALUE legacy, bool exc) { RUBY_DEBUG_LOG("exc:%d", exc); VM_ASSERT(!UNDEF_P(legacy)); VM_ASSERT(cr->sync.legacy == Qundef); RACTOR_LOCK_SELF(cr); { ractor_free_all_ports(cr); cr->sync.legacy = legacy; cr->sync.legacy_exc = exc; } RACTOR_UNLOCK_SELF(cr); } /* Sent after the dying thread's post-mortem collection: waking a joiner any earlier makes * ractor_value spin for the whole of that collection. */ static void ractor_send_exit_tokens(rb_execution_context_t *ec, rb_ractor_t *cr) { VALUE token = ractor_exit_token(cr); struct ractor_monitor *rm, *nxt; ccan_list_for_each_safe(&cr->sync.monitors, rm, nxt, node) { RUBY_DEBUG_LOG("port:%u@r%u", (unsigned int)ractor_port_id(&rm->port), (unsigned int)rb_ractor_id(rm->port.r)); ractor_send_basket(ec, &rm->port, ractor_basket_new_exit(cr->pub.self, token), false); ccan_list_del(&rm->node); SIZED_FREE(rm); } VM_ASSERT(ccan_list_empty(&cr->sync.monitors)); } // ractor-internal - initialize, mark, free, memsize static int ractor_mark_ports_i(st_data_t key, st_data_t val, st_data_t data) { // id -> ractor_queue const struct ractor_queue *rq = (struct ractor_queue *)val; ractor_queue_mark((const struct rb_gc_mark_ctx *)data, rq); return ST_CONTINUE; } static void ractor_sync_mark(const struct rb_gc_mark_ctx *ctx, rb_ractor_t *r) { /* The owner rewrites the queues, the port table and the monitor list under its sync * lock, so only the owner itself or the stopped world may walk them. */ const bool world_stopped = rb_gc_during_global_gc_p(); VM_ASSERT(world_stopped || r == rb_current_ractor_raw(false)); rb_gc_mark_ctx(ctx, r->sync.default_port_value); /* Until the value is absorbed this is its only reliable root (Qundef while the * Ractor still runs); after Ractor#value returns it, the Ruby side roots it. */ rb_gc_mark_ctx(ctx, r->sync.legacy); /* ractor_sync_init builds the rest, and a root scan reaches the main Ractor before * that: ports is what tells the two apart (the lock and the list heads are still * zeroed, and walking those crashes). Lock out foreign senders while walking them * (self-lock: not recursive, and a held Ractor lock disables malloc-GC, so no GC * nests); a stopped world needs no lock. */ if (r->sync.ports) { if (!world_stopped) RACTOR_LOCK_SELF(r); { ractor_queue_mark(ctx, r->sync.recv_queue); st_foreach(r->sync.ports, ractor_mark_ports_i, (st_data_t)ctx); ractor_mark_monitors(ctx, r); } if (!world_stopped) RACTOR_UNLOCK_SELF(r); /* The baskets on no queue: one being built to send, one being materialized. * Walked in every collection, like the queues. What they hold is shareable, but * "only a global GC frees a shareable" does not hold: pinned_roots_mark, which * roots a shareable from its page bit, is skipped once the process is back to a * single Ractor (rb_gc_single_objspace_p), and then an ordinary local GC frees * one that nothing else names. A payload in flight is named by its basket and * nothing else, so this list has to be a root whenever the queues are. No sync * lock, though: only the owner touches it (the lock above guards the queues, * which a foreign sender writes). */ ractor_mark_off_queue_baskets(ctx, r); } } static int ractor_sync_free_ports_i(st_data_t _key, st_data_t val, st_data_t _args) { struct ractor_queue *queue = (struct ractor_queue *)val; ractor_queue_free(queue); return ST_CONTINUE; } static void ractor_sync_free(rb_ractor_t *r) { if (r->sync.recv_queue) { ractor_queue_free(r->sync.recv_queue); } // maybe NULL if (r->sync.ports) { st_foreach(r->sync.ports, ractor_sync_free_ports_i, 0); st_free_table(r->sync.ports); r->sync.ports = NULL; } } static size_t ractor_sync_memsize(const rb_ractor_t *r) { if (r->sync.ports) { return st_memsize(r->sync.ports); } else { return 0; } } static void ractor_sync_init(rb_ractor_t *r) { // lock rb_native_mutex_initialize(&r->sync.lock); // monitors ccan_list_head_init(&r->sync.off_queue_baskets); ccan_list_head_init(&r->sync.monitors); // waiters ccan_list_head_init(&r->sync.waiters); // receiving queue r->sync.recv_queue = ractor_queue_new(); // ports r->sync.ports = st_init_numtable(); /* ractor_setup_default_port creates it only after the Ractor joins * vm->ractor.set, so a global GC cannot free the rootless port in between. */ r->sync.default_port_value = Qfalse; // legacy r->sync.legacy = Qundef; // no receive is rebuilding a payload yet } /* Create the default port. Call only after the Ractor joined vm->ractor.set, so the * root scan can mark the shareable port from creation onwards. */ void rb_ractor_setup_default_port(rb_ractor_t *r) { VM_ASSERT(r->sync.default_port_value == Qfalse); r->sync.default_port_value = ractor_port_new(r); RB_OBJ_SET_SHAREABLE(r->sync.default_port_value); } // Ractor#value static rb_ractor_t * ractor_set_successor_once(rb_ractor_t *r, rb_ractor_t *cr) { if (r->sync.successor == NULL) { rb_ractor_t *successor = ATOMIC_PTR_CAS(r->sync.successor, NULL, cr); return successor == NULL ? cr : successor; } return r->sync.successor; } static VALUE ractor_make_remote_exception(VALUE cause, VALUE sender) { VALUE err = rb_exc_new_cstr(rb_eRactorRemoteError, "thrown by remote Ractor."); rb_ivar_set(err, rb_intern("@ractor"), sender); rb_ec_setup_exception(NULL, err, cause); return err; } static VALUE ractor_value(rb_execution_context_t *ec, VALUE self) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); rb_ractor_t *r = RACTOR_PTR(self); rb_ractor_t *sr = ractor_set_successor_once(r, cr); if (sr == cr) { if (r->sync.legacy_taken) { rb_raise(rb_eRactorError, "The value was already taken"); } /* The value is returned by reference: inherit the dead Ractor's objspace first, * making it our own object (containment without a copy). Wait for * ractor_terminated: a monitor-port wakeup arrives before the dying thread * finishes teardown (vm_remove_ractor still touches the objspace). */ while (!rb_ractor_status_p(r, ractor_terminated)) { rb_thread_schedule(); } /* The wait above yields the GVL, so another thread of this Ractor can take the * value first: re-check. */ if (r->sync.legacy_taken) { rb_raise(rb_eRactorError, "The value was already taken"); } /* Move r's rb_gc_register_mark_object pins to the joiner before the merge * below sweeps r's objspace, or the objects pinned there lose their root. */ rb_ractor_absorb_registered_marks(GET_RACTOR(), r); rb_ractor_absorb_registered_addrs_without_gc(GET_RACTOR(), r); rb_gc_objspace_absorb_into_current(&r->objspace); /* Keep legacy alive in a C local until it is returned: after the absorb only * the C struct reaches it, so let the conservative machine-stack mark find it. */ volatile VALUE legacy_keep = r->sync.legacy; /* A dead Ractor's local storage is unreachable from Ruby (Ractor#[] only works * from inside), so let the values die and keep ractor_mark and ractor_free from * walking a stale table later. */ ractor_local_storage_free(r); r->local_storage = NULL; r->idkey_local_storage = NULL; /* The value is returned to the caller and rooted from Ruby afterwards. Drop it * from the C struct: keeping it would leave a C-only reference into the * successor's objspace, needing marking and a pin against compaction. */ VALUE legacy = r->sync.legacy; r->sync.legacy = Qnil; r->sync.legacy_taken = true; RB_GC_GUARD(legacy_keep); if (r->sync.legacy_exc) { rb_exc_raise(ractor_make_remote_exception(legacy, self)); } return legacy; } else { rb_raise(rb_eRactorError, "Only the successor ractor can take a value"); } } static VALUE ractor_marshal_dump_body(VALUE obj) { return rb_marshal_dump(obj, Qnil); } static VALUE ractor_marshal_dump_rescue(VALUE obj, VALUE errinfo) { rb_raise(rb_eRactorError, "can not copy %"PRIsVALUE" object.", rb_class_of(obj)); UNREACHABLE_RETURN(Qnil); } static VALUE ractor_prepare_payload(rb_execution_context_t *ec, VALUE obj, enum ractor_basket_type *ptype, struct rb_ractor_courier **pcourier) { switch (*ptype) { case basket_type_ref: return obj; default: if (rb_ractor_shareable_p(obj)) { *ptype = basket_type_ref; return obj; } else { /* Snapshot the object on the sender side without calling the user-visible * #clone. The courier is off-heap, so an in-flight payload is never a GC * object and needs no pin: nothing of the sender's heap stays alive while * the message waits. */ *ptype = basket_type_copy; if (rb_ractor_courier_build_copy(obj, pcourier) == NULL) { rb_raise(rb_eRactorError, "can not copy %"PRIsVALUE" object.", rb_class_of(obj)); } return Qundef; } } } #if RBIMPL_COMPILER_IS(GCC) && defined(__OPTIMIZE__) /* GCC produces false-positive -Wclobbered warnings after inlining * this function into ractor_basket_new(). */ NOINLINE(static void ractor_basket_build_payload(rb_execution_context_t *ec, struct ractor_basket *b, VALUE obj, enum ractor_basket_type type, bool exc)); #endif static void ractor_basket_build_payload(rb_execution_context_t *ec, struct ractor_basket *b, VALUE obj, enum ractor_basket_type type, bool exc) { b->exception = exc; b->p.courier = NULL; b->has_courier = true; if (type == basket_type_move) { /* Serialize the graph into an off-heap courier; the sources become * RactorMovedObject. While in flight there is no GC object left for the * sender's GC to mark, sweep or move. The build publishes the courier into * the basket as soon as it exists. */ rb_ractor_courier_build_move(obj, &b->p.courier); b->type = type; } else { VALUE v = ractor_prepare_payload(ec, obj, &type, &b->p.courier); if (b->p.courier == NULL) { b->has_courier = false; b->p.v = v; } b->type = type; } } static struct ractor_basket * ractor_basket_new(rb_execution_context_t *ec, VALUE obj, enum ractor_basket_type type, bool exc) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); /* Allocate and list the basket before anything is built into it: from here the * courier it is about to hold is rooted by this Ractor's in-flight list, even * half-built, and every raise below frees it through one path. */ struct ractor_basket *b = ractor_basket_alloc(); ractor_off_queue_add(cr, b); enum ruby_tag_type state; EC_PUSH_TAG(ec); if ((state = EC_EXEC_TAG()) == TAG_NONE) { ractor_basket_build_payload(ec, b, obj, type, exc); } EC_POP_TAG(); if (state != TAG_NONE) { ractor_basket_free(b); /* leaves the list and frees a courier already built */ EC_JUMP_TAG(ec, state); } return b; } static VALUE ractor_basket_value(struct ractor_basket *b) { switch (b->type) { case basket_type_ref: break; case basket_type_exit: /* Allocated here, in the receiving ractor: a copy, not a shared object. */ return rb_ary_new_from_args(2, b->sender, b->p.v); case basket_type_copy: /* An off-heap copy courier rebuilds exactly like a move one; only the sources * differ (still alive here, already shells there). */ case basket_type_move: { /* Rebuild the moved graph from the off-heap courier into this Ractor's * objspace. The sources are already RactorMovedObject (set when the courier * was built), so move's snapshot semantics hold. The courier is xmalloc'd * rather than a GC object, so the sender's concurrent local GC never touches * it; the shareable VALUEs it carries are marked through this basket, which is * on this Ractor's off-queue list until it is freed. * * Rebuilding can raise here too (rb_hash_aset on a moved key with a custom * #hash runs user code, and an async interrupt can arrive). On a raise the * courier is still owned by the basket, whose teardown frees it. */ rb_execution_context_t *ec = rb_current_ec_noinline(); struct rb_ractor_courier *courier = b->p.courier; /* Keep the materialized graph on the machine stack (result): it is the only * root until it reaches the caller. courier_free below runs a long loop, and * only the malloc'd basket's p.v holding it would give a concurrent global GC a * wide window. */ VALUE result = Qundef; enum ruby_tag_type state; EC_PUSH_TAG(ec); if ((state = EC_EXEC_TAG()) == TAG_NONE) { result = rb_ractor_courier_materialize(courier); } EC_POP_TAG(); if (state != TAG_NONE) { /* An unconsumed courier stays in b->p.courier; basket_free frees it. */ ractor_basket_free(b); EC_JUMP_TAG(ec, state); } rb_ractor_courier_free(courier); b->p.courier = NULL; b->has_courier = false; b->p.v = result; RB_GC_GUARD(result); break; } default: VM_ASSERT(0); // unreachable } VM_ASSERT(!b->has_courier); VM_ASSERT(!RB_TYPE_P(b->p.v, T_NONE)); return b->p.v; } static VALUE ractor_basket_accept(struct ractor_basket *b) { VALUE v = ractor_basket_value(b); if (b->exception) { VALUE err = ractor_make_remote_exception(v, b->sender); ractor_basket_free(b); rb_exc_raise(err); } ractor_basket_free(b); return v; } // Ractor blocking by receive #if VM_CHECK_MODE > 0 static bool ractor_waiter_included(rb_ractor_t *cr, rb_thread_t *th) { ASSERT_ractor_locking(cr); struct ractor_waiter *w; ccan_list_for_each(&cr->sync.waiters, w, node) { if (w->th == th) { return true; } } return false; } #endif #if USE_RUBY_DEBUG_LOG static const char * wakeup_status_str(enum ractor_wakeup_status wakeup_status) { switch (wakeup_status) { case wakeup_none: return "none"; case wakeup_by_send: return "by_send"; case wakeup_by_interrupt: return "by_interrupt"; case wakeup_by_close: return "by_close"; } rb_bug("unreachable"); } static const char * basket_type_name(enum ractor_basket_type type) { switch (type) { case basket_type_none: return "none"; case basket_type_ref: return "ref"; case basket_type_copy: return "copy"; case basket_type_move: return "move"; case basket_type_exit: return "exit"; } VM_ASSERT(0); return NULL; } #endif // USE_RUBY_DEBUG_LOG static bool ractor_wakeup_all_locked(rb_ractor_t *r, enum ractor_wakeup_status wakeup_status) { ASSERT_ractor_locking(r); RUBY_DEBUG_LOG("r:%"PRI_SERIALT_PREFIX"u wakeup:%s", rb_ractor_id(r), wakeup_status_str(wakeup_status)); bool wakeup_p = false; while (1) { struct ractor_waiter *waiter = ccan_list_pop(&r->sync.waiters, struct ractor_waiter, node); if (waiter) { VM_ASSERT(waiter->wakeup_status == wakeup_none); waiter->wakeup_status = wakeup_status; rb_ractor_sched_wakeup(r, waiter->th); wakeup_p = true; } else { break; } } return wakeup_p; } static bool ractor_wakeup_all(rb_ractor_t *r, enum ractor_wakeup_status wakeup_status) { ASSERT_ractor_unlocking(r); bool wakeup_p; RACTOR_LOCK(r); { wakeup_p = ractor_wakeup_all_locked(r, wakeup_status); } RACTOR_UNLOCK(r); return wakeup_p; } static void ubf_ractor_wait(void *ptr) { struct ractor_waiter *waiter = (struct ractor_waiter *)ptr; rb_thread_t *th = waiter->th; rb_ractor_t *r = th->ractor; rb_atomic_t event_serial = waiter->event_serial; // clear ubf and nobody can kick UBF th->unblock.func = NULL; th->unblock.arg = NULL; rb_native_mutex_unlock(&th->interrupt_lock); { RACTOR_LOCK(r); { if (RUBY_ATOMIC_LOAD(th->unblock.event_serial) == event_serial && waiter->wakeup_status == wakeup_none) { RUBY_DEBUG_LOG("waiter:%p", (void *)waiter); waiter->wakeup_status = wakeup_by_interrupt; ccan_list_del(&waiter->node); rb_ractor_sched_wakeup(r, waiter->th); } } RACTOR_UNLOCK(r); } rb_native_mutex_lock(&th->interrupt_lock); } // Waits for an event on cr. `end` is an absolute deadline, NULL to wait forever. static enum ractor_wakeup_status ractor_wait(rb_execution_context_t *ec, rb_ractor_t *cr, const rb_hrtime_t *end) { rb_thread_t *th = rb_ec_thread_ptr(ec); struct ractor_waiter waiter = { .wakeup_status = wakeup_none, .th = th, .end = end, }; RUBY_DEBUG_LOG("wait%s", ""); ASSERT_ractor_locking(cr); VM_ASSERT(GET_RACTOR() == cr); VM_ASSERT(!ractor_waiter_included(cr, th)); ccan_list_add_tail(&cr->sync.waiters, &waiter.node); // resume another ready thread and wait for an event rb_ractor_sched_wait(ec, cr, ubf_ractor_wait, &waiter); if (waiter.wakeup_status == wakeup_none) { ccan_list_del(&waiter.node); } RUBY_DEBUG_LOG("wakeup_status:%s", wakeup_status_str(waiter.wakeup_status)); RACTOR_UNLOCK_SELF(cr); { rb_ec_check_ints(ec); } RACTOR_LOCK_SELF(cr); VM_ASSERT(!ractor_waiter_included(cr, th)); return waiter.wakeup_status; } static void ractor_deliver_incoming_messages(rb_execution_context_t *ec, rb_ractor_t *cr) { ASSERT_ractor_locking(cr); struct ractor_queue *recv_q = cr->sync.recv_queue; struct ractor_basket *b; while ((b = ractor_queue_deq(cr, recv_q)) != NULL) { ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id, true), b); } } static bool ractor_check_received(rb_ractor_t *cr, struct ractor_queue *messages) { struct ractor_queue *received_queue = cr->sync.recv_queue; bool received = false; ASSERT_ractor_locking(cr); if (ractor_queue_empty_p(cr, received_queue)) { RUBY_DEBUG_LOG("empty"); } else { received = true; // messages <- incoming ractor_queue_init(messages); ractor_queue_move(messages, received_queue); } VM_ASSERT(ractor_queue_empty_p(cr, received_queue)); RUBY_DEBUG_LOG("received:%d", received); return received; } // Returns false if the deadline `end` passed with nothing to deliver. Incoming // messages are delivered even then, so the caller retries its queue once more. // A wait can end on a wakeup meant for another port, so the caller keeps the // deadline: a stream of them must not hold a timed receive past its time. static bool ractor_deadline_passed_p(const rb_hrtime_t *end) { return end != NULL && rb_hrtime_now() >= *end; } static bool ractor_wait_receive(rb_execution_context_t *ec, rb_ractor_t *cr, const rb_hrtime_t *end) { struct ractor_queue messages; bool deliverred = false; bool timedout = false; RACTOR_LOCK_SELF(cr); { if (ractor_check_received(cr, &messages)) { deliverred = true; } else if (!end) { ractor_wait(ec, cr, NULL); // no timeout: wait until a message arrives } else if (*end == 0) { timedout = true; // `timeout: 0`: over without reading any clock } else { // only a wakeup nobody claimed can be the deadline, so only then look at // the clock: a send or an interrupt says what woke this thread by itself timedout = ractor_wait(ec, cr, end) == wakeup_none && rb_hrtime_now() >= *end; } } RACTOR_UNLOCK_SELF(cr); if (deliverred) { VM_ASSERT(!ractor_queue_empty_p(cr, &messages)); struct ractor_basket *b; while ((b = ractor_queue_deq(cr, &messages)) != NULL) { ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id, false), b); } } return !timedout; } static VALUE ractor_try_receive(rb_execution_context_t *ec, rb_ractor_t *cr, const struct ractor_port *rp) { struct ractor_queue *rq = ractor_get_queue(cr, ractor_port_id(rp), false); if (rq == NULL) { rb_raise(rb_eRactorClosedError, "The port was already closed"); } struct ractor_basket *b = ractor_queue_deq(cr, rq); /* Off the queue and not yet freed: this Ractor roots it while it materializes. */ if (b) ractor_off_queue_add(cr, b); if (rq->closed && ractor_queue_empty_p(cr, rq)) { ractor_delete_port(cr, ractor_port_id(rp), false); } if (b) { return ractor_basket_accept(b); } else { return Qundef; } } // Returns Qundef if the deadline passed first. It bounds how long this blocks; it // does not cut delivery off. A message that lands while the timeout is being // reported is still returned, as Thread::Queue#pop(timeout:) does. Either way // nothing is lost: a basket only leaves the queue when it is returned. static VALUE ractor_receive(rb_execution_context_t *ec, const struct ractor_port *rp, const rb_hrtime_t *end) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); VM_ASSERT(cr == rp->r); RUBY_DEBUG_LOG("port:%u", (unsigned int)ractor_port_id(rp)); while (1) { VALUE v = ractor_try_receive(ec, cr, rp); if (v != Qundef) { return v; } else if (!ractor_wait_receive(ec, cr, end)) { return Qundef; } else if (ractor_deadline_passed_p(end)) { // The wait ended on a wakeup meant for another port, which says // nothing about the clock. One more look, then the deadline stands. return ractor_try_receive(ec, cr, rp); } } } // A timeout argument becomes an absolute deadline, or 0 for `timeout: 0`, which // every wait reads as "do not wait". Returns NULL when there is no timeout. static const rb_hrtime_t * ractor_timeout_deadline(VALUE timeout, rb_hrtime_t *storage) { if (NIL_P(timeout)) return NULL; if (!(FIXNUM_P(timeout) && FIX2LONG(timeout) == 0)) { struct timeval tv = rb_time_interval(timeout); // raises on a negative timeout rb_hrtime_t rel = rb_timeval2hrtime(&tv); if (rel > 0) { *storage = rb_hrtime_add(rb_hrtime_now(), rel); return storage; } } *storage = 0; return storage; } // Ractor#send static void ractor_send_basket(rb_execution_context_t *ec, const struct ractor_port *rp, struct ractor_basket *b, bool raise_on_error) { bool closed = false; RUBY_DEBUG_LOG("port:%u@r%"PRI_SERIALT_PREFIX"u b:%s v:%p", (unsigned int)ractor_port_id(rp), rb_ractor_id(rp->r), basket_type_name(b->type), (void *)b->p.v); RACTOR_LOCK(rp->r); { if (ractor_closed_port_p(ec, rp->r, rp)) { closed = true; } else { b->port_id = ractor_port_id(rp); /* The receiver's queue roots it from here; drop it from ours. */ ractor_off_queue_remove(b); ractor_queue_enq(rp->r, rp->r->sync.recv_queue, b); ractor_wakeup_all_locked(rp->r, wakeup_by_send); } } RACTOR_UNLOCK(rp->r); // NOTE: ref r -> b->p.v is created, but Ractor is unprotected object, so no problem on that. if (closed) { RUBY_DEBUG_LOG("closed:%u@r%"PRI_SERIALT_PREFIX"u", (unsigned int)ractor_port_id(rp), rb_ractor_id(rp->r)); /* Nothing took the basket: it was not enqueued, so free it whether or not the * caller wants the error raised. */ ractor_basket_free(b); if (raise_on_error) { rb_raise(rb_eRactorClosedError, "The port was already closed"); } } } /* sender is the ractor the token is about; both it and the token are shareable, * so nothing is copied until the receiver builds the pair. It also skips the tag * ractor_basket_new pushes: the tokens go out from a thread whose EC has already * lost its VM stack, and EC_PUSH_TAG reads ec->cfp under ZJIT. */ static struct ractor_basket * ractor_basket_new_exit(VALUE sender, VALUE token) { struct ractor_basket *b = ractor_basket_alloc(); b->type = basket_type_exit; b->sender = sender; b->p.v = token; return b; } static VALUE ractor_send0(rb_execution_context_t *ec, const struct ractor_port *rp, VALUE obj, VALUE move, bool raise_on_error) { struct ractor_basket *b = ractor_basket_new(ec, obj, RTEST(move) ? basket_type_move : basket_type_none, false); ractor_send_basket(ec, rp, b, raise_on_error); RB_GC_GUARD(obj); return rp->r->pub.self; } static VALUE ractor_send(rb_execution_context_t *ec, const struct ractor_port *rp, VALUE obj, VALUE move) { return ractor_send0(ec, rp, obj, move, true); } // Ractor::Selector struct ractor_selector { struct st_table *ports; // rpv -> rp }; static int ractor_selector_mark_i(st_data_t key, st_data_t val, st_data_t dmy) { rb_gc_mark((VALUE)key); // rpv return ST_CONTINUE; } static void ractor_selector_mark(void *ptr) { struct ractor_selector *s = ptr; if (s->ports) { st_foreach(s->ports, ractor_selector_mark_i, 0); } } static void ractor_selector_free(void *ptr) { struct ractor_selector *s = ptr; st_free_table(s->ports); SIZED_FREE(s); } static size_t ractor_selector_memsize(const void *ptr) { const struct ractor_selector *s = ptr; size_t size = sizeof(struct ractor_selector); if (s->ports) { size += st_memsize(s->ports); } return size; } static const rb_data_type_t ractor_selector_data_type = { "ractor/selector", { ractor_selector_mark, ractor_selector_free, ractor_selector_memsize, NULL, // update }, 0, 0, RUBY_TYPED_THREAD_SAFE_FREE | RUBY_TYPED_WB_PROTECTED, }; static struct ractor_selector * RACTOR_SELECTOR_PTR(VALUE selv) { VM_ASSERT(rb_typeddata_is_kind_of(selv, &ractor_selector_data_type)); return (struct ractor_selector *)DATA_PTR(selv); } // Ractor::Selector.new static VALUE ractor_selector_create(VALUE klass) { struct ractor_selector *s; VALUE selv = TypedData_Make_Struct(klass, struct ractor_selector, &ractor_selector_data_type, s); s->ports = st_init_numtable(); // TODO return selv; } // Ractor::Selector#add(r) /* * call-seq: * add(ractor) -> ractor * * Adds _ractor_ to +self+. Raises an exception if _ractor_ is already added. * Returns _ractor_. */ static VALUE ractor_selector_add(VALUE selv, VALUE rpv) { if (!ractor_port_p(rpv)) { rb_raise(rb_eArgError, "Not a Ractor::Port object"); } struct ractor_selector *s = RACTOR_SELECTOR_PTR(selv); const struct ractor_port *rp = ractor_port_ptr_check(rpv); if (st_lookup(s->ports, (st_data_t)rpv, NULL)) { rb_raise(rb_eArgError, "already added"); } st_insert(s->ports, (st_data_t)rpv, (st_data_t)rp); RB_OBJ_WRITTEN(selv, Qundef, rpv); return selv; } // Ractor::Selector#remove(r) /* call-seq: * remove(ractor) -> ractor * * Removes _ractor_ from +self+. Raises an exception if _ractor_ is not added. * Returns the removed _ractor_. */ static VALUE ractor_selector_remove(VALUE selv, VALUE rpv) { if (!ractor_port_p(rpv)) { rb_raise(rb_eArgError, "Not a Ractor::Port object"); } struct ractor_selector *s = RACTOR_SELECTOR_PTR(selv); if (!st_lookup(s->ports, (st_data_t)rpv, NULL)) { rb_raise(rb_eArgError, "not added yet"); } st_delete(s->ports, (st_data_t *)&rpv, NULL); return selv; } // Ractor::Selector#clear /* * call-seq: * clear -> self * * Removes all ractors from +self+. Raises +self+. */ static VALUE ractor_selector_clear(VALUE selv) { struct ractor_selector *s = RACTOR_SELECTOR_PTR(selv); st_clear(s->ports); return selv; } /* * call-seq: * empty? -> true or false * * Returns +true+ if no ractor is added. */ static VALUE ractor_selector_empty_p(VALUE selv) { struct ractor_selector *s = RACTOR_SELECTOR_PTR(selv); return s->ports->num_entries == 0 ? Qtrue : Qfalse; } // Ractor::Selector#wait struct ractor_selector_wait_data { rb_ractor_t *cr; rb_execution_context_t *ec; bool found; VALUE v; VALUE rpv; }; static int ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t data) { struct ractor_selector_wait_data *p = (struct ractor_selector_wait_data *)data; const struct ractor_port *rp = (const struct ractor_port *)val; VALUE v = ractor_try_receive(p->ec, p->cr, rp); if (v != Qundef) { p->found = true; p->v = v; p->rpv = (VALUE)key; return ST_STOP; } else { return ST_CONTINUE; } } static VALUE ractor_selector__wait(rb_execution_context_t *ec, VALUE selector, const rb_hrtime_t *end) { rb_ractor_t *cr = rb_ec_ractor_ptr(ec); struct ractor_selector *s = RACTOR_SELECTOR_PTR(selector); struct ractor_selector_wait_data data = { .ec = ec, .cr = cr, .found = false, }; while (1) { st_foreach(s->ports, ractor_selector_wait_i, (st_data_t)&data); if (data.found) { return rb_ary_new_from_args(2, data.rpv, data.v); } else if (!ractor_wait_receive(ec, cr, end)) { return Qnil; } else if (ractor_deadline_passed_p(end)) { st_foreach(s->ports, ractor_selector_wait_i, (st_data_t)&data); return data.found ? rb_ary_new_from_args(2, data.rpv, data.v) : Qnil; } } } /* * call-seq: * wait(receive: false, yield_value: undef, move: false) -> [ractor, value] * * Waits until any ractor in _selector_ can be active. */ static VALUE ractor_selector_wait(VALUE selector) { return ractor_selector__wait(GET_EC(), selector, NULL); } static VALUE ractor_selector_new(int argc, VALUE *ractors, VALUE klass) { VALUE selector = ractor_selector_create(klass); for (int i=0; i