14static VALUE rb_cRactorPort;
20static void ractor_add_port(
rb_ractor_t *r, st_data_t
id);
31static void ractor_port_note_alive(
const struct ractor_port *rp);
34ractor_port_mark(
void *ptr)
39 rb_gc_mark(rp->r->pub.self);
44 if (rb_gc_single_objspace_p() || rb_gc_during_global_gc_p()) {
45 ractor_port_note_alive(rp);
58 0, 0, RUBY_TYPED_THREAD_SAFE_FREE | RUBY_TYPED_WB_PROTECTED | RUBY_TYPED_FROZEN_SHAREABLE | RUBY_TYPED_EMBEDDABLE,
65 return cr->sync.next_port_id++;
69RACTOR_PORT_PTR(
VALUE self)
71 VM_ASSERT(rb_typeddata_is_kind_of(self, &ractor_port_data_type));
72 return RTYPEDDATA_GET_DATA(self);
77ractor_port_ptr_check(
VALUE self)
81 if (UNLIKELY(rp->r == NULL)) {
89ractor_port_alloc(
VALUE klass)
104 rp->id_ = ractor_genid_for_port(r);
106 ractor_add_port(r, ractor_port_id(rp));
120ractor_port_initialize(
VALUE self)
122 return ractor_port_init(self, GET_RACTOR());
127ractor_port_initialize_copy(
VALUE self,
VALUE orig)
130 struct ractor_port *src = ractor_port_ptr_check(orig);
133 dst->id_ = ractor_port_id(src);
141 VALUE rpv = ractor_port_alloc(rb_cRactorPort);
142 ractor_port_init(rpv, r);
147ractor_port_p(
VALUE self)
149 return rb_typeddata_is_kind_of(self, &ractor_port_data_type);
152static const rb_hrtime_t *ractor_timeout_deadline(
VALUE timeout, rb_hrtime_t *storage);
157 const struct ractor_port *rp = ractor_port_ptr_check(self);
159 if (rp->r != rb_ec_ractor_ptr(ec)) {
160 rb_raise(rb_eRactorError,
"only allowed from the creator Ractor of this port");
163 rb_hrtime_t deadline;
164 const rb_hrtime_t *end = ractor_timeout_deadline(timeout, &deadline);
166 VALUE v = ractor_receive(ec, rp, end);
170 return UNDEF_P(v) ?
Qnil : v;
176 const struct ractor_port *rp = ractor_port_ptr_check(self);
177 ractor_send(ec, rp, obj,
RTEST(move));
188 const struct ractor_port *rp = ractor_port_ptr_check(self);
192 if (rb_ec_ractor_ptr(ec) == r) {
195 closed = ractor_closed_port_p(ec, r, rp);
204 closed = ractor_closed_port_p(ec, r, rp);
215 const struct ractor_port *rp = ractor_port_ptr_check(self);
219 rb_raise(rb_eRactorError,
"closing port by other ractors is not allowed");
222 ractor_close_port(ec, cr, rp);
230enum ractor_basket_type {
245 enum ractor_basket_type type;
257 struct ccan_list_node node;
258 struct ccan_list_node off_queue_node;
263ractor_basket_type_p(
const struct ractor_basket *b,
enum ractor_basket_type type)
265 return b->type == type;
271 return ractor_basket_type_p(b, basket_type_none);
278 if (b->p.courier != NULL) {
283 rb_ractor_courier_mark(b->p.courier);
290 rb_gc_mark(b->sender);
296 ractor_off_queue_remove(b);
299 rb_ractor_courier_free(b->p.courier);
306ractor_basket_alloc(
void)
312 b->type = basket_type_none;
316 b->p.exception =
false;
318 ccan_list_node_init(&b->off_queue_node);
328 VM_ASSERT(cr == rb_current_ractor_raw(
false));
329 ccan_list_add_tail(&cr->sync.off_queue_baskets, &b->off_queue_node);
335 ccan_list_del_init(&b->off_queue_node);
342 ccan_list_for_each(&r->sync.off_queue_baskets, b, off_queue_node) {
343 ractor_basket_mark(b);
350 struct ccan_list_head set;
358 ccan_list_head_init(&rq->set);
364ractor_queue_new(
void)
367 ractor_queue_init(rq);
372ractor_port_note_alive(
const struct ractor_port *rp)
376 if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, rp->id_, (st_data_t *)&rq)) {
389 ccan_list_for_each(&rq->set, b, node) {
390 ractor_basket_mark(b);
399 ccan_list_for_each_safe(&rq->set, b, nxt, node) {
400 ccan_list_del_init(&b->node);
401 ractor_basket_free(b);
404 VM_ASSERT(ccan_list_empty(&rq->set));
416 ccan_list_for_each(&rq->set, b, node) {
426 bool closed_now = !rq->closed;
434 struct ccan_list_head *src = &src_rq->set;
435 struct ccan_list_head *dst = &dst_rq->set;
437 dst->n.next = src->n.next;
438 dst->n.prev = src->n.prev;
439 dst->n.next->prev = &dst->n;
440 dst->n.prev->next = &dst->n;
441 ccan_list_head_init(src);
455 return ccan_list_empty(&rq->set);
461 VM_ASSERT(GET_RACTOR() == r);
469 ccan_list_add_tail(&rq->set, &basket->node);
478 ccan_list_for_each(&rq->set, b, node) {
479 fprintf(stderr,
"%d type:%s %p\n", i, basket_type_name(b->type), (
void *)b);
485static void ractor_delete_port(
rb_ractor_t *cr, st_data_t
id,
bool locked);
488ractor_get_queue(
rb_ractor_t *cr, st_data_t
id,
bool locked)
490 VM_ASSERT(cr == GET_RACTOR());
494 if (cr->sync.ports && st_lookup(cr->sync.ports,
id, (st_data_t *)&rq)) {
495 if (rq->closed && ractor_queue_empty_p(cr, rq)) {
496 ractor_delete_port(cr,
id, locked);
514 ASSERT_ractor_unlocking(r);
516 RUBY_DEBUG_LOG(
"id:%u", (
unsigned int)
id);
520 st_table *
const old_tab = r->sync.ports;
525 inserted = st_insert_no_rebuild(old_tab,
id, (st_data_t)rq) >= 0;
542 const st_index_t entries = st_table_size(old_tab);
544 new_tab = st_copy(old_tab);
546 if (st_table_size(old_tab) == entries) {
547 st_insert(new_tab,
id, (st_data_t)rq);
549 if (st_table_size(old_tab) == entries)
break;
552 st_free_table(new_tab);
557 VM_ASSERT(r->sync.ports == old_tab);
558 r->sync.ports = new_tab;
562 st_free_table(old_tab);
567ractor_delete_port_locked(
rb_ractor_t *cr, st_data_t
id)
569 ASSERT_ractor_locking(cr);
571 RUBY_DEBUG_LOG(
"id:%u", (
unsigned int)
id);
575 if (st_delete(cr->sync.ports, &
id, (st_data_t *)&rq)) {
576 ractor_queue_free(rq);
584ractor_delete_port(
rb_ractor_t *cr, st_data_t
id,
bool locked)
587 ractor_delete_port_locked(cr,
id);
590 RACTOR_LOCK_SELF(cr);
592 ractor_delete_port_locked(cr,
id);
594 RACTOR_UNLOCK_SELF(cr);
601 return RACTOR_PORT_PTR(r->sync.default_port_value);
607 return r->sync.default_port_value;
613 VM_ASSERT(rb_ec_ractor_ptr(ec) == rp->r ? 1 : (ASSERT_ractor_locking(rp->r), 1));
617 if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, ractor_port_id(rp), (st_data_t *)&rq)) {
628static bool ractor_wakeup_all(
rb_ractor_t *r,
enum ractor_wakeup_status wakeup_status);
633 VM_ASSERT(cr == rp->r);
635 bool closed_now =
false;
637 RACTOR_LOCK_SELF(cr);
639 ractor_deliver_incoming_messages(ec, cr);
641 if (st_lookup(rp->r->sync.ports, ractor_port_id(rp), (st_data_t *)&rq)) {
642 closed_now = ractor_queue_close(rq);
644 if (ractor_queue_empty_p(cr, rq)) {
646 ractor_delete_port(cr, ractor_port_id(rp),
true);
652 RACTOR_UNLOCK_SELF(cr);
657 ractor_wakeup_all(cr, wakeup_by_close);
667ractor_reap_dead_ports_i(st_data_t port_id, st_data_t val, st_data_t dat)
676 ractor_queue_free(rq);
688 if (recv_q == NULL)
return;
691 ccan_list_for_each_safe(&recv_q->set, b, nxt, node) {
692 if (!st_lookup(r->sync.ports, b->port_id, NULL)) {
693 ccan_list_del_init(&b->node);
694 ractor_basket_free(b);
703 VM_ASSERT(rb_gc_single_objspace_p() || rb_gc_during_global_gc_p());
706 st_foreach(r->sync.ports, ractor_reap_dead_ports_i, 0);
707 ractor_reap_undeliverable_messages(r);
712ractor_free_all_ports_i(st_data_t port_id, st_data_t val, st_data_t dat)
717 ractor_queue_free(rq);
724 if (cr->sync.ports) {
725 st_foreach(cr->sync.ports, ractor_free_all_ports_i, (st_data_t)cr);
726 st_free_table(cr->sync.ports);
727 cr->sync.ports = NULL;
730 if (cr->sync.recv_queue) {
731 ractor_queue_free(cr->sync.recv_queue);
732 cr->sync.recv_queue = NULL;
736#if defined(HAVE_WORKING_FORK)
740 ractor_free_all_ports(r);
741 r->sync.legacy =
Qnil;
749 struct ccan_list_node node;
759 ccan_list_for_each(&r->sync.monitors, rm, node) {
760 rb_gc_mark(rm->port.r->pub.self);
769 if (r->sync.legacy_exc) {
770 RUBY_DEBUG_LOG(
"aborted");
774 RUBY_DEBUG_LOG(
"exited");
783 bool terminated =
false;
784 const struct ractor_port *rp = ractor_port_ptr_check(port);
790 if (UNDEF_P(r->sync.legacy)) {
791 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));
792 ccan_list_add_tail(&r->sync.monitors, &rm->node);
795 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));
803 ractor_send_basket(ec, rp, ractor_basket_new_exit(self, ractor_exit_token(r)),
false);
816 const struct ractor_port *rp = ractor_port_ptr_check(port);
820 if (UNDEF_P(r->sync.legacy)) {
823 ccan_list_for_each_safe(&r->sync.monitors, rm, nxt, node) {
824 if (rm->port.r == rp->r && ractor_port_id(&rm->port) == ractor_port_id(rp)) {
825 RUBY_DEBUG_LOG(
"r:%u -> port:%u@r%u",
826 (
unsigned int)rb_ractor_id(r),
827 (
unsigned int)ractor_port_id(&rm->port),
828 (
unsigned int)rb_ractor_id(rm->port.r));
829 ccan_list_del(&rm->node);
843 RUBY_DEBUG_LOG(
"exc:%d", exc);
844 VM_ASSERT(!UNDEF_P(legacy));
845 VM_ASSERT(cr->sync.legacy ==
Qundef);
847 RACTOR_LOCK_SELF(cr);
849 ractor_free_all_ports(cr);
851 cr->sync.legacy = legacy;
852 cr->sync.legacy_exc = exc;
854 RACTOR_UNLOCK_SELF(cr);
863 VALUE token = ractor_exit_token(cr);
866 ccan_list_for_each_safe(&cr->sync.monitors, rm, nxt, node)
868 RUBY_DEBUG_LOG(
"port:%u@r%u", (
unsigned int)ractor_port_id(&rm->port), (
unsigned int)rb_ractor_id(rm->port.r));
870 ractor_send_basket(ec, &rm->port, ractor_basket_new_exit(cr->pub.self, token),
false);
872 ccan_list_del(&rm->node);
876 VM_ASSERT(ccan_list_empty(&cr->sync.monitors));
882ractor_mark_ports_i(st_data_t key, st_data_t val, st_data_t data)
886 ractor_queue_mark(rq);
895 const bool world_stopped = rb_gc_during_global_gc_p();
896 VM_ASSERT(world_stopped || r == rb_current_ractor_raw(
false));
898 rb_gc_mark(r->sync.default_port_value);
902 rb_gc_mark(r->sync.legacy);
910 if (!world_stopped) RACTOR_LOCK_SELF(r);
912 ractor_queue_mark(r->sync.recv_queue);
913 st_foreach(r->sync.ports, ractor_mark_ports_i, 0);
914 ractor_mark_monitors(r);
916 if (!world_stopped) RACTOR_UNLOCK_SELF(r);
927 ractor_mark_off_queue_baskets(r);
932ractor_sync_free_ports_i(st_data_t _key, st_data_t val, st_data_t _args)
936 ractor_queue_free(queue);
944 if (r->sync.recv_queue) {
945 ractor_queue_free(r->sync.recv_queue);
950 st_foreach(r->sync.ports, ractor_sync_free_ports_i, 0);
951 st_free_table(r->sync.ports);
952 r->sync.ports = NULL;
960 return st_memsize(r->sync.ports);
974 ccan_list_head_init(&r->sync.off_queue_baskets);
975 ccan_list_head_init(&r->sync.monitors);
978 ccan_list_head_init(&r->sync.waiters);
981 r->sync.recv_queue = ractor_queue_new();
984 r->sync.ports = st_init_numtable();
987 r->sync.default_port_value =
Qfalse;
1001 VM_ASSERT(r->sync.default_port_value ==
Qfalse);
1002 r->sync.default_port_value = ractor_port_new(r);
1011 if (r->sync.successor == NULL) {
1012 rb_ractor_t *successor = ATOMIC_PTR_CAS(r->sync.successor, NULL, cr);
1013 return successor == NULL ? cr : successor;
1016 return r->sync.successor;
1020ractor_make_remote_exception(
VALUE cause,
VALUE sender)
1024 rb_ec_setup_exception(NULL, err, cause);
1033 rb_ractor_t *sr = ractor_set_successor_once(r, cr);
1036 if (r->sync.legacy_taken) {
1037 rb_raise(rb_eRactorError,
"The value was already taken");
1044 while (!rb_ractor_status_p(r, ractor_terminated)) {
1050 if (r->sync.legacy_taken) {
1051 rb_raise(rb_eRactorError,
"The value was already taken");
1056 rb_ractor_absorb_registered_marks(GET_RACTOR(), r);
1058 rb_gc_objspace_absorb_into_current(&r->objspace);
1062 volatile VALUE legacy_keep = r->sync.legacy;
1067 ractor_local_storage_free(r);
1068 r->local_storage = NULL;
1069 r->idkey_local_storage = NULL;
1074 VALUE legacy = r->sync.legacy;
1075 r->sync.legacy =
Qnil;
1076 r->sync.legacy_taken =
true;
1079 if (r->sync.legacy_exc) {
1080 rb_exc_raise(ractor_make_remote_exception(legacy, self));
1085 rb_raise(rb_eRactorError,
"Only the successor ractor can take a value");
1090ractor_marshal_dump_body(
VALUE obj)
1096ractor_marshal_dump_rescue(
VALUE obj,
VALUE errinfo)
1098 rb_raise(rb_eRactorError,
"can not copy %"PRIsVALUE
" object.",
rb_class_of(obj));
1107 case basket_type_ref:
1111 *ptype = basket_type_ref;
1119 *ptype = basket_type_copy;
1120 if (rb_ractor_courier_build_copy(obj, pcourier) == NULL) {
1121 rb_raise(rb_eRactorError,
"can not copy %"PRIsVALUE
" object.",
rb_class_of(obj));
1128#if RBIMPL_COMPILER_IS(GCC) && defined(__OPTIMIZE__)
1135 enum ractor_basket_type type,
bool exc)
1137 b->p.exception = exc;
1138 if (type == basket_type_move) {
1143 rb_ractor_courier_build_move(obj, &b->p.courier);
1148 VALUE v = ractor_prepare_payload(ec, obj, &type, &b->p.courier);
1162 ractor_off_queue_add(cr, b);
1164 enum ruby_tag_type state;
1166 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
1167 ractor_basket_build_payload(ec, b, obj, type, exc);
1170 if (state != TAG_NONE) {
1171 ractor_basket_free(b);
1172 EC_JUMP_TAG(ec, state);
1182 case basket_type_ref:
1184 case basket_type_exit:
1186 return rb_ary_new_from_args(2, b->sender, b->p.v);
1187 case basket_type_copy:
1190 case basket_type_move: {
1208 enum ruby_tag_type state;
1210 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
1211 result = rb_ractor_courier_materialize(courier);
1214 if (state != TAG_NONE) {
1216 ractor_basket_free(b);
1217 EC_JUMP_TAG(ec, state);
1219 rb_ractor_courier_free(courier);
1220 b->p.courier = NULL;
1236 VALUE v = ractor_basket_value(b);
1238 if (b->p.exception) {
1239 VALUE err = ractor_make_remote_exception(v, b->sender);
1240 ractor_basket_free(b);
1244 ractor_basket_free(b);
1250#if VM_CHECK_MODE > 0
1254 ASSERT_ractor_locking(cr);
1258 ccan_list_for_each(&cr->sync.waiters, w, node) {
1268#if USE_RUBY_DEBUG_LOG
1271wakeup_status_str(
enum ractor_wakeup_status wakeup_status)
1273 switch (wakeup_status) {
1274 case wakeup_none:
return "none";
1275 case wakeup_by_send:
return "by_send";
1276 case wakeup_by_interrupt:
return "by_interrupt";
1277 case wakeup_by_close:
return "by_close";
1279 rb_bug(
"unreachable");
1283basket_type_name(
enum ractor_basket_type
type)
1286 case basket_type_none:
return "none";
1287 case basket_type_ref:
return "ref";
1288 case basket_type_copy:
return "copy";
1289 case basket_type_move:
return "move";
1290 case basket_type_exit:
return "exit";
1299ractor_wakeup_all(
rb_ractor_t *r,
enum ractor_wakeup_status wakeup_status)
1301 ASSERT_ractor_unlocking(r);
1303 RUBY_DEBUG_LOG(
"r:%"PRI_SERIALT_PREFIX
"u wakeup:%s", rb_ractor_id(r), wakeup_status_str(wakeup_status));
1305 bool wakeup_p =
false;
1312 VM_ASSERT(waiter->wakeup_status == wakeup_none);
1314 waiter->wakeup_status = wakeup_status;
1315 rb_ractor_sched_wakeup(r, waiter->th);
1329ubf_ractor_wait(
void *ptr)
1338 th->unblock.func = NULL;
1339 th->unblock.arg = NULL;
1345 if (
RUBY_ATOMIC_LOAD(th->unblock.event_serial) == event_serial && waiter->wakeup_status == wakeup_none) {
1346 RUBY_DEBUG_LOG(
"waiter:%p", (
void *)waiter);
1348 waiter->wakeup_status = wakeup_by_interrupt;
1349 ccan_list_del(&waiter->node);
1351 rb_ractor_sched_wakeup(r, waiter->th);
1360static enum ractor_wakeup_status
1366 .wakeup_status = wakeup_none,
1371 RUBY_DEBUG_LOG(
"wait%s",
"");
1373 ASSERT_ractor_locking(cr);
1375 VM_ASSERT(GET_RACTOR() == cr);
1376 VM_ASSERT(!ractor_waiter_included(cr, th));
1378 ccan_list_add_tail(&cr->sync.waiters, &waiter.node);
1381 rb_ractor_sched_wait(ec, cr, ubf_ractor_wait, &waiter);
1383 if (waiter.wakeup_status == wakeup_none) {
1384 ccan_list_del(&waiter.node);
1387 RUBY_DEBUG_LOG(
"wakeup_status:%s", wakeup_status_str(waiter.wakeup_status));
1389 RACTOR_UNLOCK_SELF(cr);
1391 rb_ec_check_ints(ec);
1393 RACTOR_LOCK_SELF(cr);
1395 VM_ASSERT(!ractor_waiter_included(cr, th));
1396 return waiter.wakeup_status;
1402 ASSERT_ractor_locking(cr);
1406 while ((b = ractor_queue_deq(cr, recv_q)) != NULL) {
1407 ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id,
true), b);
1414 struct ractor_queue *received_queue = cr->sync.recv_queue;
1415 bool received =
false;
1417 ASSERT_ractor_locking(cr);
1419 if (ractor_queue_empty_p(cr, received_queue)) {
1420 RUBY_DEBUG_LOG(
"empty");
1426 ractor_queue_init(messages);
1427 ractor_queue_move(messages, received_queue);
1430 VM_ASSERT(ractor_queue_empty_p(cr, received_queue));
1432 RUBY_DEBUG_LOG(
"received:%d", received);
1441ractor_deadline_passed_p(
const rb_hrtime_t *end)
1443 return end != NULL && rb_hrtime_now() >= *end;
1450 bool deliverred =
false;
1451 bool timedout =
false;
1453 RACTOR_LOCK_SELF(cr);
1455 if (ractor_check_received(cr, &messages)) {
1459 ractor_wait(ec, cr, NULL);
1461 else if (*end == 0) {
1467 timedout = ractor_wait(ec, cr, end) == wakeup_none && rb_hrtime_now() >= *end;
1470 RACTOR_UNLOCK_SELF(cr);
1473 VM_ASSERT(!ractor_queue_empty_p(cr, &messages));
1476 while ((b = ractor_queue_deq(cr, &messages)) != NULL) {
1477 ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id,
false), b);
1487 struct ractor_queue *rq = ractor_get_queue(cr, ractor_port_id(rp),
false);
1490 rb_raise(rb_eRactorClosedError,
"The port was already closed");
1495 if (b) ractor_off_queue_add(cr, b);
1497 if (rq->closed && ractor_queue_empty_p(cr, rq)) {
1498 ractor_delete_port(cr, ractor_port_id(rp),
false);
1502 return ractor_basket_accept(b);
1517 VM_ASSERT(cr == rp->r);
1519 RUBY_DEBUG_LOG(
"port:%u", (
unsigned int)ractor_port_id(rp));
1522 VALUE v = ractor_try_receive(ec, cr, rp);
1527 else if (!ractor_wait_receive(ec, cr, end)) {
1530 else if (ractor_deadline_passed_p(end)) {
1533 return ractor_try_receive(ec, cr, rp);
1540static const rb_hrtime_t *
1541ractor_timeout_deadline(
VALUE timeout, rb_hrtime_t *storage)
1543 if (
NIL_P(timeout))
return NULL;
1547 rb_hrtime_t rel = rb_timeval2hrtime(&tv);
1550 *storage = rb_hrtime_add(rb_hrtime_now(), rel);
1564 bool closed =
false;
1566 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);
1570 if (ractor_closed_port_p(ec, rp->r, rp)) {
1574 b->port_id = ractor_port_id(rp);
1576 ractor_off_queue_remove(b);
1577 ractor_queue_enq(rp->r, rp->r->sync.recv_queue, b);
1580 RACTOR_UNLOCK(rp->r);
1585 ractor_wakeup_all(rp->r, wakeup_by_send);
1588 RUBY_DEBUG_LOG(
"closed:%u@r%"PRI_SERIALT_PREFIX
"u", (
unsigned int)ractor_port_id(rp), rb_ractor_id(rp->r));
1592 ractor_basket_free(b);
1594 if (raise_on_error) {
1595 rb_raise(rb_eRactorClosedError,
"The port was already closed");
1605ractor_basket_new_exit(
VALUE sender,
VALUE token)
1609 b->type = basket_type_exit;
1619 struct ractor_basket *b = ractor_basket_new(ec, obj,
RTEST(move) ? basket_type_move : basket_type_none, false);
1620 ractor_send_basket(ec, rp, b, raise_on_error);
1622 return rp->r->pub.self;
1628 return ractor_send0(ec, rp, obj, move,
true);
1639ractor_selector_mark_i(st_data_t key, st_data_t val, st_data_t dmy)
1641 rb_gc_mark((
VALUE)key);
1647ractor_selector_mark(
void *ptr)
1652 st_foreach(s->ports, ractor_selector_mark_i, 0);
1657ractor_selector_free(
void *ptr)
1660 st_free_table(s->ports);
1665ractor_selector_memsize(
const void *ptr)
1670 size += st_memsize(s->ports);
1678 ractor_selector_mark,
1679 ractor_selector_free,
1680 ractor_selector_memsize,
1683 0, 0, RUBY_TYPED_THREAD_SAFE_FREE | RUBY_TYPED_WB_PROTECTED,
1687RACTOR_SELECTOR_PTR(
VALUE selv)
1689 VM_ASSERT(rb_typeddata_is_kind_of(selv, &ractor_selector_data_type));
1696ractor_selector_create(
VALUE klass)
1700 s->ports = st_init_numtable();
1716 if (!ractor_port_p(rpv)) {
1717 rb_raise(rb_eArgError,
"Not a Ractor::Port object");
1721 const struct ractor_port *rp = ractor_port_ptr_check(rpv);
1723 if (st_lookup(s->ports, (st_data_t)rpv, NULL)) {
1724 rb_raise(rb_eArgError,
"already added");
1727 st_insert(s->ports, (st_data_t)rpv, (st_data_t)rp);
1744 if (!ractor_port_p(rpv)) {
1745 rb_raise(rb_eArgError,
"Not a Ractor::Port object");
1750 if (!st_lookup(s->ports, (st_data_t)rpv, NULL)) {
1751 rb_raise(rb_eArgError,
"not added yet");
1754 st_delete(s->ports, (st_data_t *)&rpv, NULL);
1768ractor_selector_clear(
VALUE selv)
1782ractor_selector_empty_p(
VALUE selv)
1785 return s->ports->num_entries == 0 ?
Qtrue :
Qfalse;
1799ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t data)
1804 VALUE v = ractor_try_receive(p->ec, p->cr, rp);
1809 p->rpv = (
VALUE)key;
1830 st_foreach(s->ports, ractor_selector_wait_i, (st_data_t)&data);
1833 return rb_ary_new_from_args(2, data.rpv, data.v);
1835 else if (!ractor_wait_receive(ec, cr, end)) {
1838 else if (ractor_deadline_passed_p(end)) {
1839 st_foreach(s->ports, ractor_selector_wait_i, (st_data_t)&data);
1840 return data.found ? rb_ary_new_from_args(2, data.rpv, data.v) :
Qnil;
1852ractor_selector_wait(
VALUE selector)
1854 return ractor_selector__wait(GET_EC(), selector, NULL);
1858ractor_selector_new(
int argc,
VALUE *ractors,
VALUE klass)
1860 VALUE selector = ractor_selector_create(klass);
1862 for (
int i=0; i<argc; i++) {
1863 ractor_selector_add(selector, ractors[i]);
1872 rb_hrtime_t deadline;
1873 const rb_hrtime_t *end = ractor_timeout_deadline(timeout, &deadline);
1876 VALUE result = ractor_selector__wait(ec, selector, end);
1883#ifndef USE_RACTOR_SELECTOR
1884#define USE_RACTOR_SELECTOR 0
1887RUBY_SYMBOL_EXPORT_BEGIN
1888void rb_init_ractor_selector(
void);
1889RUBY_SYMBOL_EXPORT_END
1898rb_init_ractor_selector(
void)
1907 rb_define_method(rb_cRactorSelector,
"empty?", ractor_selector_empty_p, 0);
1912Init_RactorPort(
void)
1917 rb_define_method(rb_cRactorPort,
"initialize_copy", ractor_port_initialize_copy, 1);
1919#if USE_RACTOR_SELECTOR
1920 rb_init_ractor_selector();
std::atomic< unsigned > rb_atomic_t
Type that is eligible for atomic operations.
#define RUBY_ATOMIC_LOAD(var)
Atomic load.
#define rb_define_method(klass, mid, func, arity)
Defines klass#mid.
#define rb_define_singleton_method(klass, mid, func, arity)
Defines klass.mid.
#define ALLOC
Old name of RB_ALLOC.
#define Qundef
Old name of RUBY_Qundef.
#define ID2SYM
Old name of RB_ID2SYM.
#define UNREACHABLE_RETURN
Old name of RBIMPL_UNREACHABLE_RETURN.
#define T_NONE
Old name of RUBY_T_NONE.
#define Qtrue
Old name of RUBY_Qtrue.
#define Qnil
Old name of RUBY_Qnil.
#define Qfalse
Old name of RUBY_Qfalse.
#define FIX2LONG
Old name of RB_FIX2LONG.
#define NIL_P
Old name of RB_NIL_P.
#define FIXNUM_P
Old name of RB_FIXNUM_P.
void rb_exc_raise(VALUE mesg)
Raises an exception in the current thread.
VALUE rb_eTypeError
TypeError exception.
VALUE rb_cObject
Object class.
VALUE rb_cRactor
Ractor class.
static VALUE rb_class_of(VALUE obj)
Object to class mapping function.
VALUE rb_obj_class(VALUE obj)
Queries the class of an object.
VALUE rb_obj_freeze(VALUE obj)
Same as RB_OBJ_FREEZE(), but returns the given object.
#define RB_OBJ_WRITTEN(old, oldv, young)
Identical to RB_OBJ_WRITE(), except it doesn't write any values, but only a WB declaration.
VALUE rb_marshal_dump(VALUE obj, VALUE port)
Serialises the given object and all its referring objects, to write them down to the passed port.
#define rb_exc_new_cstr(exc, str)
Identical to rb_exc_new(), except it assumes the passed pointer is a pointer to a C string.
void rb_thread_schedule(void)
Tries to switch to another thread.
struct timeval rb_time_interval(VALUE num)
Creates a "time interval".
VALUE rb_ivar_set(VALUE obj, ID name, VALUE val)
Identical to rb_iv_set(), except it accepts the name as an ID instead of a C string.
void rb_undef_alloc_func(VALUE klass)
Deletes the allocator function of a class.
void rb_define_alloc_func(VALUE klass, rb_alloc_func_t func)
Sets the allocator function of a class.
#define RB_OBJ_SET_SHAREABLE(obj)
Wrapper of rb_obj_set_shareable().
static bool rb_ractor_shareable_p(VALUE obj)
Queries if multiple Ractors can share the passed object or not.
#define RBIMPL_ATTR_MAYBE_UNUSED()
Wraps (or simulates) [[maybe_unused]]
#define RB_GC_GUARD(v)
Prevents premature destruction of local objects.
VALUE type(ANYARGS)
ANYARGS-ed function type.
static int RARRAY_LENINT(VALUE ary)
Identical to rb_array_len(), except it differs for the return type.
#define RARRAY_CONST_PTR
Just another name of rb_array_const_ptr.
#define RUBY_TYPED_DEFAULT_FREE
This is a value you can set to rb_data_type_struct::dfree.
#define DATA_PTR(obj)
Convenient casting macro for backward compatibility.
#define TypedData_Make_Struct(klass, type, data_type, sval)
Identical to TypedData_Wrap_Struct, except it allocates a new data region internally instead of takin...
#define RTEST
This is an old name of RB_TEST.
This is the struct that holds necessary info for a struct.
void rb_native_mutex_lock(rb_nativethread_lock_t *lock)
Just another name of rb_nativethread_lock_lock.
void rb_native_mutex_initialize(rb_nativethread_lock_t *lock)
Just another name of rb_nativethread_lock_initialize.
void rb_native_mutex_unlock(rb_nativethread_lock_t *lock)
Just another name of rb_nativethread_lock_unlock.
uintptr_t VALUE
Type that represents a Ruby object.
static bool RB_TYPE_P(VALUE obj, enum ruby_value_type t)
Queries if the given object is of given type.