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 {
241 enum ractor_basket_type type;
262 struct ccan_list_node node;
263 struct ccan_list_node off_queue_node;
268ractor_basket_type_p(
const struct ractor_basket *b,
enum ractor_basket_type type)
270 return b->type == type;
276 return ractor_basket_type_p(b, basket_type_none);
283 if (b->p.courier != NULL) {
288 rb_ractor_courier_mark(b->p.courier);
290 else if (b->p.mbuf == NULL) {
299 ractor_off_queue_remove(b);
300 ruby_xfree(b->p.mbuf);
305 rb_ractor_courier_free(b->p.courier);
312ractor_basket_alloc(
void)
318 b->type = basket_type_none;
322 b->p.exception =
false;
323 b->p.marshaled =
false;
327 ccan_list_node_init(&b->off_queue_node);
337 VM_ASSERT(cr == rb_current_ractor_raw(
false));
338 ccan_list_add_tail(&cr->sync.off_queue_baskets, &b->off_queue_node);
344 ccan_list_del_init(&b->off_queue_node);
351 ccan_list_for_each(&r->sync.off_queue_baskets, b, off_queue_node) {
352 ractor_basket_mark(b);
359 struct ccan_list_head set;
367 ccan_list_head_init(&rq->set);
373ractor_queue_new(
void)
376 ractor_queue_init(rq);
381ractor_port_note_alive(
const struct ractor_port *rp)
385 if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, rp->id_, (st_data_t *)&rq)) {
398 ccan_list_for_each(&rq->set, b, node) {
399 ractor_basket_mark(b);
408 ccan_list_for_each_safe(&rq->set, b, nxt, node) {
409 ccan_list_del_init(&b->node);
410 ractor_basket_free(b);
413 VM_ASSERT(ccan_list_empty(&rq->set));
425 ccan_list_for_each(&rq->set, b, node) {
440 struct ccan_list_head *src = &src_rq->set;
441 struct ccan_list_head *dst = &dst_rq->set;
443 dst->n.next = src->n.next;
444 dst->n.prev = src->n.prev;
445 dst->n.next->prev = &dst->n;
446 dst->n.prev->next = &dst->n;
447 ccan_list_head_init(src);
461 return ccan_list_empty(&rq->set);
467 VM_ASSERT(GET_RACTOR() == r);
475 ccan_list_add_tail(&rq->set, &basket->node);
484 ccan_list_for_each(&rq->set, b, node) {
485 fprintf(stderr,
"%d type:%s %p\n", i, basket_type_name(b->type), (
void *)b);
491static void ractor_delete_port(
rb_ractor_t *cr, st_data_t
id,
bool locked);
494ractor_get_queue(
rb_ractor_t *cr, st_data_t
id,
bool locked)
496 VM_ASSERT(cr == GET_RACTOR());
500 if (cr->sync.ports && st_lookup(cr->sync.ports,
id, (st_data_t *)&rq)) {
501 if (rq->closed && ractor_queue_empty_p(cr, rq)) {
502 ractor_delete_port(cr,
id, locked);
520 ASSERT_ractor_unlocking(r);
522 RUBY_DEBUG_LOG(
"id:%u", (
unsigned int)
id);
526 st_table *
const old_tab = r->sync.ports;
531 inserted = st_insert_no_rebuild(old_tab,
id, (st_data_t)rq) >= 0;
539 st_table *
const new_tab = st_copy(old_tab);
540 st_insert(new_tab,
id, (st_data_t)rq);
544 VM_ASSERT(r->sync.ports == old_tab);
545 r->sync.ports = new_tab;
549 st_free_table(old_tab);
554ractor_delete_port_locked(
rb_ractor_t *cr, st_data_t
id)
556 ASSERT_ractor_locking(cr);
558 RUBY_DEBUG_LOG(
"id:%u", (
unsigned int)
id);
562 if (st_delete(cr->sync.ports, &
id, (st_data_t *)&rq)) {
563 ractor_queue_free(rq);
571ractor_delete_port(
rb_ractor_t *cr, st_data_t
id,
bool locked)
574 ractor_delete_port_locked(cr,
id);
577 RACTOR_LOCK_SELF(cr);
579 ractor_delete_port_locked(cr,
id);
581 RACTOR_UNLOCK_SELF(cr);
588 return RACTOR_PORT_PTR(r->sync.default_port_value);
594 return r->sync.default_port_value;
600 VM_ASSERT(rb_ec_ractor_ptr(ec) == rp->r ? 1 : (ASSERT_ractor_locking(rp->r), 1));
604 if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, ractor_port_id(rp), (st_data_t *)&rq)) {
618 VM_ASSERT(cr == rp->r);
621 RACTOR_LOCK_SELF(cr);
623 ractor_deliver_incoming_messages(ec, cr);
625 if (st_lookup(rp->r->sync.ports, ractor_port_id(rp), (st_data_t *)&rq)) {
626 ractor_queue_close(rq);
628 if (ractor_queue_empty_p(cr, rq)) {
630 ractor_delete_port(cr, ractor_port_id(rp),
true);
636 RACTOR_UNLOCK_SELF(cr);
645ractor_reap_dead_ports_i(st_data_t port_id, st_data_t val, st_data_t dat)
654 ractor_queue_free(rq);
663 st_foreach(r->sync.ports, ractor_reap_dead_ports_i, 0);
668ractor_free_all_ports_i(st_data_t port_id, st_data_t val, st_data_t dat)
673 ractor_queue_free(rq);
680 if (cr->sync.ports) {
681 st_foreach(cr->sync.ports, ractor_free_all_ports_i, (st_data_t)cr);
682 st_free_table(cr->sync.ports);
683 cr->sync.ports = NULL;
686 if (cr->sync.recv_queue) {
687 ractor_queue_free(cr->sync.recv_queue);
688 cr->sync.recv_queue = NULL;
692#if defined(HAVE_WORKING_FORK)
696 ractor_free_all_ports(r);
697 r->sync.legacy =
Qnil;
705 struct ccan_list_node node;
715 ccan_list_for_each(&r->sync.monitors, rm, node) {
716 rb_gc_mark(rm->port.r->pub.self);
721ractor_exit_token(
bool exc)
724 RUBY_DEBUG_LOG(
"aborted");
728 RUBY_DEBUG_LOG(
"exited");
737 bool terminated =
false;
738 const struct ractor_port *rp = ractor_port_ptr_check(port);
744 if (UNDEF_P(r->sync.legacy)) {
745 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));
746 ccan_list_add_tail(&r->sync.monitors, &rm->node);
749 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));
757 ractor_port_send(ec, port, ractor_exit_token(r->sync.legacy_exc),
Qfalse);
770 const struct ractor_port *rp = ractor_port_ptr_check(port);
774 if (UNDEF_P(r->sync.legacy)) {
777 ccan_list_for_each_safe(&r->sync.monitors, rm, nxt, node) {
778 if (rm->port.r == rp->r && ractor_port_id(&rm->port) == ractor_port_id(rp)) {
779 RUBY_DEBUG_LOG(
"r:%u -> port:%u@r%u",
780 (
unsigned int)rb_ractor_id(r),
781 (
unsigned int)ractor_port_id(&rm->port),
782 (
unsigned int)rb_ractor_id(rm->port.r));
783 ccan_list_del(&rm->node);
797 RUBY_DEBUG_LOG(
"exc:%d", exc);
798 VM_ASSERT(!UNDEF_P(legacy));
799 VM_ASSERT(cr->sync.legacy ==
Qundef);
801 RACTOR_LOCK_SELF(cr);
803 ractor_free_all_ports(cr);
805 cr->sync.legacy = legacy;
806 cr->sync.legacy_exc = exc;
808 RACTOR_UNLOCK_SELF(cr);
817 VALUE token = ractor_exit_token(cr->sync.legacy_exc);
820 ccan_list_for_each_safe(&cr->sync.monitors, rm, nxt, node)
822 RUBY_DEBUG_LOG(
"port:%u@r%u", (
unsigned int)ractor_port_id(&rm->port), (
unsigned int)rb_ractor_id(rm->port.r));
824 ractor_send_basket(ec, &rm->port, ractor_basket_new_ref(token),
false);
826 ccan_list_del(&rm->node);
830 VM_ASSERT(ccan_list_empty(&cr->sync.monitors));
836ractor_mark_ports_i(st_data_t key, st_data_t val, st_data_t data)
840 ractor_queue_mark(rq);
849 const bool world_stopped = rb_gc_during_global_gc_p();
850 VM_ASSERT(world_stopped || r == rb_current_ractor_raw(
false));
852 rb_gc_mark(r->sync.default_port_value);
856 rb_gc_mark(r->sync.legacy);
864 if (!world_stopped) RACTOR_LOCK_SELF(r);
866 ractor_queue_mark(r->sync.recv_queue);
867 st_foreach(r->sync.ports, ractor_mark_ports_i, 0);
868 ractor_mark_monitors(r);
870 if (!world_stopped) RACTOR_UNLOCK_SELF(r);
881 ractor_mark_off_queue_baskets(r);
886ractor_sync_free_ports_i(st_data_t _key, st_data_t val, st_data_t _args)
890 ractor_queue_free(queue);
898 if (r->sync.recv_queue) {
899 ractor_queue_free(r->sync.recv_queue);
904 st_foreach(r->sync.ports, ractor_sync_free_ports_i, 0);
905 st_free_table(r->sync.ports);
906 r->sync.ports = NULL;
914 return st_memsize(r->sync.ports);
928 ccan_list_head_init(&r->sync.off_queue_baskets);
929 ccan_list_head_init(&r->sync.monitors);
932 ccan_list_head_init(&r->sync.waiters);
935 r->sync.recv_queue = ractor_queue_new();
938 r->sync.ports = st_init_numtable();
941 r->sync.default_port_value =
Qfalse;
955 VM_ASSERT(r->sync.default_port_value ==
Qfalse);
956 r->sync.default_port_value = ractor_port_new(r);
958 rb_gc_obj_became_shareable(r->sync.default_port_value);
966 if (r->sync.successor == NULL) {
967 rb_ractor_t *successor = ATOMIC_PTR_CAS(r->sync.successor, NULL, cr);
968 return successor == NULL ? cr : successor;
971 return r->sync.successor;
975ractor_make_remote_exception(
VALUE cause,
VALUE sender)
979 rb_ec_setup_exception(NULL, err, cause);
988 rb_ractor_t *sr = ractor_set_successor_once(r, cr);
991 if (r->sync.legacy_taken) {
992 rb_raise(rb_eRactorError,
"The value was already taken");
999 while (!rb_ractor_status_p(r, ractor_terminated)) {
1005 if (r->sync.legacy_taken) {
1006 rb_raise(rb_eRactorError,
"The value was already taken");
1011 rb_ractor_absorb_registered_marks(GET_RACTOR(), r);
1013 rb_gc_objspace_absorb_into_current(&r->objspace);
1017 volatile VALUE legacy_keep = r->sync.legacy;
1022 ractor_local_storage_free(r);
1023 r->local_storage = NULL;
1024 r->idkey_local_storage = NULL;
1029 VALUE legacy = r->sync.legacy;
1030 r->sync.legacy =
Qnil;
1031 r->sync.legacy_taken =
true;
1034 if (r->sync.legacy_exc) {
1035 rb_exc_raise(ractor_make_remote_exception(legacy, self));
1040 rb_raise(rb_eRactorError,
"Only the successor ractor can take a value");
1044static VALUE ractor_copy_native_try(
VALUE obj);
1047ractor_marshal_dump_body(
VALUE obj)
1053ractor_marshal_dump_rescue(
VALUE obj,
VALUE errinfo)
1055 rb_raise(rb_eRactorError,
"can not copy %"PRIsVALUE
" object.",
rb_class_of(obj));
1064 case basket_type_ref:
1068 *ptype = basket_type_ref;
1078 *ptype = basket_type_copy;
1079 if (rb_ractor_courier_build_copy(obj, pcourier) != NULL)
return Qundef;
1082 return rb_rescue2(ractor_marshal_dump_body, obj,
1083 ractor_marshal_dump_rescue, obj,
1089#if RBIMPL_COMPILER_IS(GCC) && defined(__OPTIMIZE__)
1096 enum ractor_basket_type type,
bool exc)
1098 b->p.exception = exc;
1099 if (type == basket_type_move) {
1104 rb_ractor_courier_build_move(obj, &b->p.courier);
1109 bool marshaled =
false;
1110 VALUE v = ractor_prepare_payload(ec, obj, &type, &marshaled, &b->p.courier);
1112 if (type == basket_type_copy && marshaled) {
1115 size_t mlen = (size_t)RSTRING_LEN(v);
1116 char *mbuf =
ALLOC_N(
char, mlen > 0 ? mlen : 1);
1117 b->p.marshaled = marshaled;
1120 memcpy(mbuf, RSTRING_PTR(v), mlen);
1136 ractor_off_queue_add(cr, b);
1138 enum ruby_tag_type state;
1140 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
1141 ractor_basket_build_payload(ec, b, obj, type, exc);
1144 if (state != TAG_NONE) {
1145 ractor_basket_free(b);
1146 EC_JUMP_TAG(ec, state);
1156 case basket_type_ref:
1158 case basket_type_copy: {
1161 if (b->p.courier != NULL)
goto materialize_courier;
1168 enum ruby_tag_type state;
1170 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
1183 ec->gen_fields_cache.obj =
Qundef;
1184 ec->gen_fields_cache.fields_obj =
Qundef;
1185 if (state != TAG_NONE) {
1188 ractor_basket_free(b);
1189 EC_JUMP_TAG(ec, state);
1196 case basket_type_move:
1197 materialize_courier: {
1215 enum ruby_tag_type state;
1217 if ((state = EC_EXEC_TAG()) == TAG_NONE) {
1218 result = rb_ractor_courier_materialize(courier);
1221 if (state != TAG_NONE) {
1223 ractor_basket_free(b);
1224 EC_JUMP_TAG(ec, state);
1226 rb_ractor_courier_free(courier);
1227 b->p.courier = NULL;
1243 VALUE v = ractor_basket_value(b);
1245 if (b->p.exception) {
1246 VALUE err = ractor_make_remote_exception(v, b->sender);
1247 ractor_basket_free(b);
1251 ractor_basket_free(b);
1257#if VM_CHECK_MODE > 0
1261 ASSERT_ractor_locking(cr);
1265 ccan_list_for_each(&cr->sync.waiters, w, node) {
1275#if USE_RUBY_DEBUG_LOG
1278wakeup_status_str(
enum ractor_wakeup_status wakeup_status)
1280 switch (wakeup_status) {
1281 case wakeup_none:
return "none";
1282 case wakeup_by_send:
return "by_send";
1283 case wakeup_by_interrupt:
return "by_interrupt";
1286 rb_bug(
"unreachable");
1290basket_type_name(
enum ractor_basket_type
type)
1293 case basket_type_none:
return "none";
1294 case basket_type_ref:
return "ref";
1295 case basket_type_copy:
return "copy";
1296 case basket_type_move:
return "move";
1305ractor_wakeup_all(
rb_ractor_t *r,
enum ractor_wakeup_status wakeup_status)
1307 ASSERT_ractor_unlocking(r);
1309 RUBY_DEBUG_LOG(
"r:%u wakeup:%s", rb_ractor_id(r), wakeup_status_str(wakeup_status));
1311 bool wakeup_p =
false;
1318 VM_ASSERT(waiter->wakeup_status == wakeup_none);
1320 waiter->wakeup_status = wakeup_status;
1321 rb_ractor_sched_wakeup(r, waiter->th);
1335ubf_ractor_wait(
void *ptr)
1344 th->unblock.func = NULL;
1345 th->unblock.arg = NULL;
1351 if (
RUBY_ATOMIC_LOAD(th->unblock.event_serial) == event_serial && waiter->wakeup_status == wakeup_none) {
1352 RUBY_DEBUG_LOG(
"waiter:%p", (
void *)waiter);
1354 waiter->wakeup_status = wakeup_by_interrupt;
1355 ccan_list_del(&waiter->node);
1357 rb_ractor_sched_wakeup(r, waiter->th);
1366static enum ractor_wakeup_status
1372 .wakeup_status = wakeup_none,
1377 RUBY_DEBUG_LOG(
"wait%s",
"");
1379 ASSERT_ractor_locking(cr);
1381 VM_ASSERT(GET_RACTOR() == cr);
1382 VM_ASSERT(!ractor_waiter_included(cr, th));
1384 ccan_list_add_tail(&cr->sync.waiters, &waiter.node);
1387 rb_ractor_sched_wait(ec, cr, ubf_ractor_wait, &waiter);
1389 if (waiter.wakeup_status == wakeup_none) {
1390 ccan_list_del(&waiter.node);
1393 RUBY_DEBUG_LOG(
"wakeup_status:%s", wakeup_status_str(waiter.wakeup_status));
1395 RACTOR_UNLOCK_SELF(cr);
1397 rb_ec_check_ints(ec);
1399 RACTOR_LOCK_SELF(cr);
1401 VM_ASSERT(!ractor_waiter_included(cr, th));
1402 return waiter.wakeup_status;
1408 ASSERT_ractor_locking(cr);
1412 while ((b = ractor_queue_deq(cr, recv_q)) != NULL) {
1413 ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id,
true), b);
1420 struct ractor_queue *received_queue = cr->sync.recv_queue;
1421 bool received =
false;
1423 ASSERT_ractor_locking(cr);
1425 if (ractor_queue_empty_p(cr, received_queue)) {
1426 RUBY_DEBUG_LOG(
"empty");
1432 ractor_queue_init(messages);
1433 ractor_queue_move(messages, received_queue);
1436 VM_ASSERT(ractor_queue_empty_p(cr, received_queue));
1438 RUBY_DEBUG_LOG(
"received:%d", received);
1448 bool deliverred =
false;
1449 bool timedout =
false;
1451 RACTOR_LOCK_SELF(cr);
1453 if (ractor_check_received(cr, &messages)) {
1457 ractor_wait(ec, cr, NULL);
1459 else if (*end == 0) {
1465 timedout = ractor_wait(ec, cr, end) == wakeup_none && rb_hrtime_now() >= *end;
1468 RACTOR_UNLOCK_SELF(cr);
1471 VM_ASSERT(!ractor_queue_empty_p(cr, &messages));
1474 while ((b = ractor_queue_deq(cr, &messages)) != NULL) {
1475 ractor_queue_enq(cr, ractor_get_queue(cr, b->port_id,
false), b);
1485 struct ractor_queue *rq = ractor_get_queue(cr, ractor_port_id(rp),
false);
1488 rb_raise(rb_eRactorClosedError,
"The port was already closed");
1493 if (b) ractor_off_queue_add(cr, b);
1495 if (rq->closed && ractor_queue_empty_p(cr, rq)) {
1496 ractor_delete_port(cr, ractor_port_id(rp),
false);
1500 return ractor_basket_accept(b);
1515 VM_ASSERT(cr == rp->r);
1517 RUBY_DEBUG_LOG(
"port:%u", (
unsigned int)ractor_port_id(rp));
1520 VALUE v = ractor_try_receive(ec, cr, rp);
1525 else if (!ractor_wait_receive(ec, cr, end)) {
1533static const rb_hrtime_t *
1534ractor_timeout_deadline(
VALUE timeout, rb_hrtime_t *storage)
1536 if (
NIL_P(timeout))
return NULL;
1540 rb_hrtime_t rel = rb_timeval2hrtime(&tv);
1543 *storage = rb_hrtime_add(rb_hrtime_now(), rel);
1557 bool closed =
false;
1559 RUBY_DEBUG_LOG(
"port:%u@r%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);
1563 if (ractor_closed_port_p(ec, rp->r, rp)) {
1567 b->port_id = ractor_port_id(rp);
1569 ractor_off_queue_remove(b);
1570 ractor_queue_enq(rp->r, rp->r->sync.recv_queue, b);
1573 RACTOR_UNLOCK(rp->r);
1578 ractor_wakeup_all(rp->r, wakeup_by_send);
1581 RUBY_DEBUG_LOG(
"closed:%u@r%u", (
unsigned int)ractor_port_id(rp), rb_ractor_id(rp->r));
1585 ractor_basket_free(b);
1587 if (raise_on_error) {
1588 rb_raise(rb_eRactorClosedError,
"The port was already closed");
1597ractor_basket_new_ref(
VALUE shareable)
1601 b->type = basket_type_ref;
1604 b->p.exception =
false;
1605 b->p.marshaled =
false;
1606 b->p.courier = NULL;
1616 struct ractor_basket *b = ractor_basket_new(ec, obj,
RTEST(move) ? basket_type_move : basket_type_none, false);
1617 ractor_send_basket(ec, rp, b, raise_on_error);
1619 return rp->r->pub.self;
1625 return ractor_send0(ec, rp, obj, move,
true);
1636ractor_selector_mark_i(st_data_t key, st_data_t val, st_data_t dmy)
1638 rb_gc_mark((
VALUE)key);
1644ractor_selector_mark(
void *ptr)
1649 st_foreach(s->ports, ractor_selector_mark_i, 0);
1654ractor_selector_free(
void *ptr)
1657 st_free_table(s->ports);
1662ractor_selector_memsize(
const void *ptr)
1667 size += st_memsize(s->ports);
1675 ractor_selector_mark,
1676 ractor_selector_free,
1677 ractor_selector_memsize,
1680 0, 0, RUBY_TYPED_THREAD_SAFE_FREE | RUBY_TYPED_WB_PROTECTED,
1684RACTOR_SELECTOR_PTR(
VALUE selv)
1686 VM_ASSERT(rb_typeddata_is_kind_of(selv, &ractor_selector_data_type));
1693ractor_selector_create(
VALUE klass)
1697 s->ports = st_init_numtable();
1713 if (!ractor_port_p(rpv)) {
1714 rb_raise(rb_eArgError,
"Not a Ractor::Port object");
1718 const struct ractor_port *rp = ractor_port_ptr_check(rpv);
1720 if (st_lookup(s->ports, (st_data_t)rpv, NULL)) {
1721 rb_raise(rb_eArgError,
"already added");
1724 st_insert(s->ports, (st_data_t)rpv, (st_data_t)rp);
1741 if (!ractor_port_p(rpv)) {
1742 rb_raise(rb_eArgError,
"Not a Ractor::Port object");
1747 if (!st_lookup(s->ports, (st_data_t)rpv, NULL)) {
1748 rb_raise(rb_eArgError,
"not added yet");
1751 st_delete(s->ports, (st_data_t *)&rpv, NULL);
1765ractor_selector_clear(
VALUE selv)
1779ractor_selector_empty_p(
VALUE selv)
1782 return s->ports->num_entries == 0 ?
Qtrue :
Qfalse;
1796ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t data)
1801 VALUE v = ractor_try_receive(p->ec, p->cr, rp);
1806 p->rpv = (
VALUE)key;
1827 st_foreach(s->ports, ractor_selector_wait_i, (st_data_t)&data);
1830 return rb_ary_new_from_args(2, data.rpv, data.v);
1832 else if (!ractor_wait_receive(ec, cr, end)) {
1845ractor_selector_wait(
VALUE selector)
1847 return ractor_selector__wait(GET_EC(), selector, NULL);
1851ractor_selector_new(
int argc,
VALUE *ractors,
VALUE klass)
1853 VALUE selector = ractor_selector_create(klass);
1855 for (
int i=0; i<argc; i++) {
1856 ractor_selector_add(selector, ractors[i]);
1865 rb_hrtime_t deadline;
1866 const rb_hrtime_t *end = ractor_timeout_deadline(timeout, &deadline);
1869 VALUE result = ractor_selector__wait(ec, selector, end);
1876#ifndef USE_RACTOR_SELECTOR
1877#define USE_RACTOR_SELECTOR 0
1880RUBY_SYMBOL_EXPORT_BEGIN
1881void rb_init_ractor_selector(
void);
1882RUBY_SYMBOL_EXPORT_END
1891rb_init_ractor_selector(
void)
1900 rb_define_method(rb_cRactorSelector,
"empty?", ractor_selector_empty_p, 0);
1905Init_RactorPort(
void)
1910 rb_define_method(rb_cRactorPort,
"initialize_copy", ractor_port_initialize_copy, 1);
1912#if USE_RACTOR_SELECTOR
1913 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.
@ RUBY_FL_SHAREABLE
This flag has something to do with Ractor.
#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 ALLOC_N
Old name of RB_ALLOC_N.
#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.
#define FL_SET_RAW
Old name of RB_FL_SET_RAW.
VALUE rb_rescue2(VALUE(*b_proc)(VALUE), VALUE data1, VALUE(*r_proc)(VALUE, VALUE), VALUE data2,...)
An equivalent of rescue clause.
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)
Just calls rb_obj_freeze_inline() inside.
#define RB_OBJ_WRITTEN(old, oldv, young)
Identical to RB_OBJ_WRITE(), except it doesn't write any values, but only a WB declaration.
#define rb_str_new(str, len)
Allocates an instance of rb_cString.
#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.
static bool rb_ractor_shareable_p(VALUE obj)
Queries if multiple Ractors can share the passed object or not.
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.
VALUE rb_marshal_load(VALUE port)
Deserialises a previous output of rb_marshal_dump() into a network of objects.
#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.