- r.take

- r.take

- r.take

- r.take

- 'ok' if r.take == ['ping', 'pong']

- r.take

-assert_equal '[2, 3, 1]', %q{

- r = Ractor.new Ractor.current do |main|

- main << 1

- main << 2

- main << 3

- end

- a = []

- a << Ractor.receive_if{|msg| msg == 2}

- a << Ractor.receive_if{|msg| msg == 3}

- a << Ractor.receive

-}

-# Ractor#receive_if with break

-assert_equal '[2, [1, :break], 3]', %q{

- r = Ractor.new Ractor.current do |main|

- main << 1

- main << 2

- main << 3

-

- a << Ractor.receive_if{|msg| msg == 2}

- a << Ractor.receive_if{|msg| break [msg, :break]}

- a << Ractor.receive

-}

-

-# Ractor#receive_if can't be called recursively

-assert_equal '[[:e1, 1], [:e2, 2]]', %q{

- r = Ractor.new Ractor.current do |main|

- main << 1

- main << 2

- main << 3

- end

-

- a = []

-

- Ractor.receive_if do |msg|

- begin

- Ractor.receive

- rescue Ractor::Error

- a << [:e1, msg]

- end

- true # delete 1 from queue

- end

-

- Ractor.receive_if do |msg|

- begin

- Ractor.receive_if{}

- rescue Ractor::Error

- a << [:e2, msg]

- end

- true # delete 2 from queue

- end

-

- a #

- }.map(&:take)

-# Ractor.select(*ractors) receives a values from a ractors.

-# It is similar to select(2) and Go's select syntax.

-# The return value is [ch, received_value]

- r, obj = Ractor.select(r1)

- 'ok' if r == r1 and obj == 'r1'

- r1 = Ractor.new{'r1'}

- r2 = Ractor.new{'r2'}

- rs = [r1, r2]

- r, obj = Ractor.select(*rs)

- rs.delete(r)

- r, obj = Ractor.select(*rs)

-# Outgoing port of a ractor will be closed when the Ractor is terminated.

-assert_equal 'ok', %q{

- r = Ractor.new do

- 'finish'

- end

-

- r.take

- sleep 0.1 until r.inspect =~ /terminated/

-

- begin

- o = r.take

- rescue Ractor::ClosedError

- 'ok'

- else

- "ng: #{o}"

- end

-}

-

- r.take # closed

-# Raise Ractor::ClosedError when try to send into a closed actor

-assert_equal 'ok', %q{

- r = Ractor.new { Ractor.receive }

- r.close_incoming

-

- begin

- r.send(1)

- rescue Ractor::ClosedError

- 'ok'

- else

- 'ng'

- end

-}

-

-# Raise Ractor::ClosedError when try to take from closed actor

-assert_equal 'ok', %q{

- r = Ractor.new do

- Ractor.yield 1

- Ractor.receive

- end

-

- r.close_outgoing

- begin

- r.take

- rescue Ractor::ClosedError

- 'ok'

- else

- 'ng'

- end

-}

-

-# Can mix with Thread#interrupt and Ractor#take [Bug #17366]

- Ractor.new{

- }.take

- }.take.inspect #=> nil

-# Ractor.yield raises Ractor::ClosedError when outgoing port is closed.

- r = Ractor.new Ractor.current do |main|

- main << true

- Ractor.yield 1

- end

-

- r.close_outgoing

- r << true

- Ractor.receive

-

- begin

- r.take

- rescue Ractor::ClosedError

- 'ok'

- else

- 'ng'

- end

-}

-# Raise Ractor::ClosedError when try to send into a ractor with closed incoming port

-assert_equal 'ok', %q{

- r = Ractor.new { Ractor.receive }

- r.close_incoming

-# A ractor with closed incoming port still can send messages out

-assert_equal '[1, 2]', %q{

- r = Ractor.new do

- Ractor.yield 1

- 2

- end

- r.close_incoming

-

- [r.take, r.take]

-}

-

-# Raise Ractor::ClosedError when try to take from a ractor with closed outgoing port

-assert_equal 'ok', %q{

- r = Ractor.new do

- Ractor.yield 1

- Ractor.receive

- end

-

- sleep 0.01 # wait for Ractor.yield in r

- r.close_outgoing

- begin

- r.take

- rescue Ractor::ClosedError

- 'ok'

- else

- 'ng'

- end

-}

-

-# A ractor with closed outgoing port still can receive messages from incoming port

-assert_equal 'ok', %q{

- r = Ractor.new do

- Ractor.receive

- end

-

- r.close_outgoing

- begin

- r.send(1)

- rescue Ractor::ClosedError

- 'ng'

- else

- 'ok'

- end

-}

-

- }.take == Ractor.current

- close_outgoing

-# multiple Ractors can receive (wait) from one Ractor

-assert_equal '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]', %q{

- pipe = Ractor.new do

- loop do

- Ractor.yield Ractor.receive

- end

- end

-

- RN = 10

- rs = RN.times.map{|i|

- Ractor.new pipe, i do |pipe, i|

- msg = pipe.take

- msg # ping-pong

- end

- }

- RN.times{|i|

- pipe << i

- }

- RN.times.map{

- r, n = Ractor.select(*rs)

- rs.delete r

- n

- }.sort

-} unless /mswin/ =~ RUBY_PLATFORM # randomly hangs on mswin https://.com/ruby/ruby/actions/runs/3753871445/jobs/6377551069#step:20:131

-

-# Ractor.select also support multiple take, receive and yield

-assert_equal '[true, true, true]', %q{

- RN = 10

- CR = Ractor.current

-

- rs = (1..RN).map{

- Ractor.new do

- CR.send 'send' + CR.take #=> 'sendyield'

- 'take'

- end

- }

- received = []

- taken = []

- yielded = []

- until received.size == RN && taken.size == RN && yielded.size == RN

- r, v = Ractor.select(CR, *rs, yield_value: 'yield')

- case r

- when :receive

- received << v

- when :yield

- yielded << v

- else

- taken << v

- rs.delete r

- end

- r = [received == ['sendyield'] * RN,

- yielded == [nil] * RN,

- taken == ['take'] * RN,

- ]

-

- STDERR.puts [received, yielded, taken].inspect

- r

-}

-

-# multiple Ractors can send to one Ractor

-assert_equal '[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]', %q{

- pipe = Ractor.new do

- loop do

- Ractor.yield Ractor.receive

- end

-

- RN = 10

- RN.times.map{|i|

- Ractor.new pipe, i do |pipe, i|

- pipe << i

- end

- }

- RN.times.map{

- pipe.take

- }.sort

-# an exception in a Ractor main thread will be re-raised at Ractor#receive

- r.take

- r.take

- obj.object_id == r.take

- echo_ractor = Ractor.new do

- Ractor.yield v

- o2 = echo_ractor.take

- o2 = echo_ractor.take

- end.take

- modified = r.take

- a2 = r.take

-r = Ractor.new do

- obj = receive

- { frozen: obj.frozen? }

-end

-obj = [Object.new].freeze

-r.send(obj, move: true)

-r.take[:frozen]

-}

-

-# move with yield

-assert_equal 'hello', %q{

- r = Ractor.new do

- Thread.current.report_on_exception = false

- obj = 'hello'

- Ractor.yield obj, move: true

- obj << 'world'

- end

-

- str = r.take

- begin

- r.take

- rescue Ractor::RemoteError

- str #=> "hello"

- end

-}

-

-# yield/move should not make moved object when the yield is not succeeded

-assert_equal '"str"', %q{

- R = Ractor.new{}

- M = Ractor.current

- s = 'str'

- selected_r, v = Ractor.select R, yield_value: s, move: true

- raise if selected_r != R # taken from R

- M.send s.inspect # s should not be a moved object

-

- Ractor.receive

-}

-

-# yield/move can fail

-assert_equal "allocator undefined for Thread", %q{

- r = Ractor.new do

- obj = Thread.new{}

- Ractor.yield obj

- rescue => e

- e.message

- end

- r.take

- r.take

- r.take

- [$stdin, $stdout, $stderr].zip(r.take){|io, (oid, fno)|

- r.take

- end.take

- r.take == self.object_id #=> false

- r.object_id == r.take #=> true

- r.take

- r.take

- r.take

- r.take

- end.take.to_s

- a = Ractor.new{ C.int }.take

- end.take

- end.take

- d = Ractor.new{ M.int }.take

- end.take

- end.take

- }.take

-class Obj

- attr_accessor :a, :b, :c, :d

- def initialize

- @a = 1

- @b = 2

- @c = 3

-end

-r = Ractor.new do

- obj = receive

- obj.d = 4

- [obj.a, obj.b, obj.c, obj.d]

-end

-obj = Obj.new

-r.send(obj, move: true)

-values = r.take

-values.join

- r.take

- r.take

- r.take

- Ractor.new{ str() }.take

- r.take

- Ractor.new{buggy(10)}.take

- Ractor.new{ [A.size, H.size] }.take

- ractors[0].send('End 0').take

- ractors[1].send('End 1').take

- ractors[2].send('End 2').take

- }.take

- end.take

- Ractor.new{ C.new.foo }.take

- }.take

- }.take

- }.take

- }.take

- }.take

- Ractor.yield self[:key]

- ans << r.take

- ans << r.take

- }.take

- }.take

- }.map{|r| r.take}.join

- }.map{|r| r.take}

- }.map{|r| r.take}.join

- Ractor.new do

- Ractor.yield Time.now

- 1_000.times { idle_worker, tmp_reporter = Ractor.select(*workers) }

- rs.map{|r| r.take} == Array.new(RN){n}

- eval("Ractor.new{}.take", nil, "test_ractor.rb", 1)

- r.take

- Ractor.main << Ractor.new{ Ractor.yield :v3; :v4 }

- Ractor.main << Ractor.new{ Ractor.yield :v5; :v6 }

- }.take.fileno > 0

- end.take

- end.take

- r.take.fileno > 0

- r.take

- end.each(&:take)

- roundtripped_obj = ractor.take

- roundtripped_obj = ractor.take

- roundtripped_obj = ractor.take

- roundtripped_obj = ractor.take

- roundtripped_obj = ractor.take

- roundtripped_obj = ractor.take

- roundtripped_obj = ractor.take[0]

- roundtripped_obj = ractor.take

- r = Ractor.new do

- Ractor.yield val

- val = r.take

- r.take

- raise "ng"

- rescue Ractor::ClosedError

- raise unless r.take == 123

- raise "ng"

- raise unless r.take == 123

- Ractor.new do

- Ractor.yield nil

- 100.times { Ractor.select(*workers) }

- end.take

- end.take

-# There are some bugs in Windows with multiple threads in same ractor calling ractor actions

-# Ex: https://.com/ruby/ruby/actions/runs/14998660285/job/42139383905

-unless /mswin/ =~ RUBY_PLATFORM

- # r.send and r.take from multiple threads

- # [Bug #21037]

- assert_equal '[true, true]', %q{

- class Map

- def initialize

- @r = Ractor.new {

- loop do

- key = Ractor.receive

- Ractor.yield key

- end

- }

- end

- def fetch(key)

- @r.send key

- @r.take

- end

- tm = Map.new

- t1 = Thread.new { 10.times.map { tm.fetch("t1") } }

- t2 = Thread.new { 10.times.map { tm.fetch("t2") } }

- vals = t1.value + t2.value

- [

- vals.first(10).all? { |v| v == "t1" },

- vals.last(10).all? { |v| v == "t2" }

- ]

- }

- # r.send and Ractor.select from multiple threads

- assert_equal '[true, true]', %q{

- class Map

- def initialize

- @r = Ractor.new {

- loop do

- key = Ractor.receive

- Ractor.yield key

- end

- }

- end

- def fetch(key)

- @r.send key

- _r, val = Ractor.select(@r)

- val

- end

- tm = Map.new

- t1 = Thread.new { 10.times.map { tm.fetch("t1") } }

- t2 = Thread.new { 10.times.map { tm.fetch("t2") } }

- vals = t1.value + t2.value

- [

- vals.first(10).all? { |v| v == "t1" },

- vals.last(10).all? { |v| v == "t2" }

- ]

- }

- # Ractor.receive in multiple threads in same ractor

- # [Bug #17624]

- assert_equal '["T1 received", "T2 received"]', %q{

- r1 = Ractor.new do

- output = []

- m = Mutex.new

- # Start two listener threads

- t1 = Thread.new do

- Ractor.receive

- m.synchronize do

- output << "T1 received"

- end

- end

- t2 = Thread.new do

- Ractor.receive

- m.synchronize do

- output << "T2 received"

- end

- end

- sleep 0.1 until [t1,t2].all? { |t| t.status == "sleep" }

- Ractor.main.send(:both_blocking)

- [t1, t2].each(&:join)

- output

- Ractor.receive # wait until both threads have blocked

- r1.send(1)

- r1.send(2)

- r1.take.sort

- }

-end

-# Moving an old object

-assert_equal 'ok', %q{

- o = Ractor.receive

- GC.verify_internal_consistency

- GC.start

- o

- o = "ok"

- # Make o an old object

- 3.times { GC.start }

- r.send(o, move: true)

- r.take

- tracing_ractor = Ractor.new do

- Ractor.yield(nil)

- Ractor.yield(nil)

- tracing_ractor.take

- tracing_ractor.take

- tracing_ractor.take

- ractor = Ractor.new do

- Ractor.yield(nil)

- ractor.take

- ractor.take

- Ractor.new { A.foo }.take

-# Regression test for deadlock between branch_stub_hit and ractor_receive_if

-assert_equal '10', %q{

- r = Ractor.new Ractor.current do |main|

- main << 1

- main << 2

- main << 3

- main << 4

- main << 5

- main << 6

- main << 7

- main << 8

- main << 9

- main << 10

- end

-

- a = []

- a << Ractor.receive_if{|msg| msg == 10}

- a << Ractor.receive_if{|msg| msg == 9}

- a << Ractor.receive_if{|msg| msg == 8}

- a << Ractor.receive_if{|msg| msg == 7}

- a << Ractor.receive_if{|msg| msg == 6}

- a << Ractor.receive_if{|msg| msg == 5}

- a << Ractor.receive_if{|msg| msg == 4}

- a << Ractor.receive_if{|msg| msg == 3}

- a << Ractor.receive_if{|msg| msg == 2}

- a << Ractor.receive_if{|msg| msg == 1}

-

- a.length

-}

-

- r.take

- r.take

- 'ok' if r.take == ['ping', 'pong']

- r.take

- RB_VM_LOCK_LEAVE_LEV(&lev);

- RUBY_DEBUG_LOG("obj:%p (%s)", (void *)obj, rb_obj_info(obj));

-static struct rb_ractor_basket *ractor_queue_at(rb_ractor_t *r, struct rb_ractor_queue *rq, int i);

-static void

-ractor_queue_mark(struct rb_ractor_queue *rq)

-{

- for (int i=0; i<rq->cnt; i++) {

- struct rb_ractor_basket *b = ractor_queue_at(NULL, rq, i);

- rb_gc_mark(b->sender);

-

- switch (b->type.e) {

- case basket_type_yielding:

- case basket_type_take_basket:

- case basket_type_deleted:

- case basket_type_reserved:

- // ignore

- break;

- default:

- rb_gc_mark(b->p.send.v);

- }

- }

-}

- ractor_queue_mark(&r->sync.recv_queue);

- ractor_queue_mark(&r->sync.takers_queue);

-ractor_queue_free(struct rb_ractor_queue *rq)

-{

- free(rq->baskets);

-}

-

-static void

- ractor_queue_free(&r->sync.recv_queue);

- ractor_queue_free(&r->sync.takers_queue);

-ractor_queue_memsize(const struct rb_ractor_queue *rq)

-{

- return sizeof(struct rb_ractor_basket) * rq->size;

-}

-

-static size_t

- return sizeof(rb_ractor_t) +

- ractor_queue_memsize(&r->sync.recv_queue) +

- ractor_queue_memsize(&r->sync.takers_queue);

-// Ractor queue

-

-static void

-ractor_queue_setup(struct rb_ractor_queue *rq)

-{

- rq->size = 2;

- rq->cnt = 0;

- rq->start = 0;

- rq->baskets = malloc(sizeof(struct rb_ractor_basket) * rq->size);

-}

-

-static struct rb_ractor_basket *

-ractor_queue_head(rb_ractor_t *r, struct rb_ractor_queue *rq)

-{

- if (r != NULL) ASSERT_ractor_locking(r);

- return &rq->baskets[rq->start];

-}

-

-static struct rb_ractor_basket *

-ractor_queue_at(rb_ractor_t *r, struct rb_ractor_queue *rq, int i)

-{

- if (r != NULL) ASSERT_ractor_locking(r);

- return &rq->baskets[(rq->start + i) % rq->size];

-}

-

-static void

-ractor_queue_advance(rb_ractor_t *r, struct rb_ractor_queue *rq)

-{

- ASSERT_ractor_locking(r);

-

- if (rq->reserved_cnt == 0) {

- rq->cnt--;

- rq->start = (rq->start + 1) % rq->size;

- rq->serial++;

- }

- else {

- ractor_queue_at(r, rq, 0)->type.e = basket_type_deleted;

- }

-}

-

-static bool

-ractor_queue_skip_p(rb_ractor_t *r, struct rb_ractor_queue *rq, int i)

-{

- struct rb_ractor_basket *b = ractor_queue_at(r, rq, i);

- return basket_type_p(b, basket_type_deleted) ||

- basket_type_p(b, basket_type_reserved);

-}

-

-static void

-ractor_queue_compact(rb_ractor_t *r, struct rb_ractor_queue *rq)

-{

- ASSERT_ractor_locking(r);

-

- while (rq->cnt > 0 && basket_type_p(ractor_queue_at(r, rq, 0), basket_type_deleted)) {

- ractor_queue_advance(r, rq);

- }

-}

-

-static bool

-ractor_queue_empty_p(rb_ractor_t *r, struct rb_ractor_queue *rq)

-{

- ASSERT_ractor_locking(r);

-

- if (rq->cnt == 0) {

- return true;

- }

-

- ractor_queue_compact(r, rq);

-

- for (int i=0; i<rq->cnt; i++) {

- if (!ractor_queue_skip_p(r, rq, i)) {

- return false;

- }

- }

-

- return true;

-}

-

-static bool

-ractor_queue_deq(rb_ractor_t *r, struct rb_ractor_queue *rq, struct rb_ractor_basket *basket)

-{

- ASSERT_ractor_locking(r);

-

- for (int i=0; i<rq->cnt; i++) {

- if (!ractor_queue_skip_p(r, rq, i)) {

- struct rb_ractor_basket *b = ractor_queue_at(r, rq, i);

- *basket = *b;

-

- // remove from queue

- b->type.e = basket_type_deleted;

- ractor_queue_compact(r, rq);

- return true;

- }

- }

-

- return false;

-}

-

-static void

-ractor_queue_enq(rb_ractor_t *r, struct rb_ractor_queue *rq, struct rb_ractor_basket *basket)

-{

- ASSERT_ractor_locking(r);

-

- if (rq->size <= rq->cnt) {

- rq->baskets = realloc(rq->baskets, sizeof(struct rb_ractor_basket) * rq->size * 2);

- for (int i=rq->size - rq->start; i<rq->cnt; i++) {

- rq->baskets[i + rq->start] = rq->baskets[i + rq->start - rq->size];

- }

- rq->size *= 2;

- }

- // copy basket into queue

- rq->baskets[(rq->start + rq->cnt++) % rq->size] = *basket;

- // fprintf(stderr, "%s %p->cnt:%d\n", RUBY_FUNCTION_NAME_STRING, (void *)rq, rq->cnt);

-}

-

-static void

-ractor_queue_delete(rb_ractor_t *r, struct rb_ractor_queue *rq, struct rb_ractor_basket *basket)

-{

- basket->type.e = basket_type_deleted;

-}

-

-// Ractor basket

-

-static VALUE ractor_reset_belonging(VALUE obj); // in this file

-

-static VALUE

-ractor_basket_value(struct rb_ractor_basket *b)

-{

- switch (b->type.e) {

- case basket_type_ref:

- break;

- case basket_type_copy:

- case basket_type_move:

- case basket_type_will:

- b->type.e = basket_type_ref;

- b->p.send.v = ractor_reset_belonging(b->p.send.v);

- break;

- default:

- rb_bug("unreachable");

- }

-

- return b->p.send.v;

-}

-

-static VALUE

-ractor_basket_accept(struct rb_ractor_basket *b)

-{

- VALUE v = ractor_basket_value(b);

-

- // a ractor's main thread had an error and yielded us this exception during its dying moments

- if (b->p.send.exception) {

- VALUE cause = v;

- VALUE err = rb_exc_new_cstr(rb_eRactorRemoteError, "thrown by remote Ractor.");

- rb_ivar_set(err, rb_intern("@ractor"), b->sender);

- rb_ec_setup_exception(NULL, err, cause);

- rb_exc_raise(err);

- }

-

- return v;

-}

-

-// Ractor synchronizations

-

-#if USE_RUBY_DEBUG_LOG

-static const char *

-wait_status_str(enum rb_ractor_wait_status wait_status)

-{

- switch ((int)wait_status) {

- case wait_none: return "none";

- case wait_receiving: return "receiving";

- case wait_taking: return "taking";

- case wait_yielding: return "yielding";

- case wait_receiving|wait_taking: return "receiving|taking";

- case wait_receiving|wait_yielding: return "receiving|yielding";

- case wait_taking|wait_yielding: return "taking|yielding";

- case wait_receiving|wait_taking|wait_yielding: return "receiving|taking|yielding";

- }

- rb_bug("unreachable");

-}

-

-static const char *

-wakeup_status_str(enum rb_ractor_wakeup_status wakeup_status)

-{

- switch (wakeup_status) {

- case wakeup_none: return "none";

- case wakeup_by_send: return "by_send";

- case wakeup_by_yield: return "by_yield";

- case wakeup_by_take: return "by_take";

- case wakeup_by_close: return "by_close";

- case wakeup_by_interrupt: return "by_interrupt";

- case wakeup_by_retry: return "by_retry";

- }

- rb_bug("unreachable");

-}

-

-static const char *

-basket_type_name(enum rb_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_will: return "will";

- case basket_type_deleted: return "deleted";

- case basket_type_reserved: return "reserved";

- case basket_type_take_basket: return "take_basket";

- case basket_type_yielding: return "yielding";

- }

- VM_ASSERT(0);

- return NULL;

-}

-#endif // USE_RUBY_DEBUG_LOG

-

-static rb_thread_t *

-ractor_sleeping_by(const rb_ractor_t *r, rb_thread_t *th, enum rb_ractor_wait_status wait_status)

-{

- if (th) {

- if ((th->ractor_waiting.wait_status & wait_status) && th->ractor_waiting.wakeup_status == wakeup_none) {

- return th;

- }

- }

- else {

- // find any thread that has this ractor wait status that is blocked

- ccan_list_for_each(&r->sync.wait.waiting_threads, th, ractor_waiting.waiting_node) {

- if ((th->ractor_waiting.wait_status & wait_status) && th->ractor_waiting.wakeup_status == wakeup_none) {

- return th;

- }

- }

- }

- return NULL;

-}

-

-#ifdef RUBY_THREAD_PTHREAD_H

-// thread_*.c

-void rb_ractor_sched_wakeup(rb_ractor_t *r, rb_thread_t *th);

-#else

-

-// win32

-static void

-rb_ractor_sched_wakeup(rb_ractor_t *r, rb_thread_t *th)

-{

- (void)r;

- ASSERT_ractor_locking(r);

- rb_native_cond_signal(&th->ractor_waiting.cond);

-

-}

-#endif

-

-

-/*

- * Wakeup `r` if the given `th` is blocked and has the given ractor `wait_status`.

- * Wakeup any blocked thread in `r` with the given ractor `wait_status` if `th` is NULL.

- */

-static bool

-ractor_wakeup(rb_ractor_t *r, rb_thread_t *th /* can be NULL */, enum rb_ractor_wait_status wait_status, enum rb_ractor_wakeup_status wakeup_status)

-{

- ASSERT_ractor_locking(r);

-

- RUBY_DEBUG_LOG("r:%u wait:%s wakeup:%s",

- rb_ractor_id(r),

- wait_status_str(wait_status),

- wakeup_status_str(wakeup_status));

-

- if ((th = ractor_sleeping_by(r, th, wait_status)) != NULL) {

- th->ractor_waiting.wakeup_status = wakeup_status;

- rb_ractor_sched_wakeup(r, th);

- return true;

- }

- else {

- return false;

- }

-}

-

-// unblock function (UBF). This gets called when another thread on this or another ractor sets our thread's interrupt flag.

-// This is not async-safe.

-static void

-ractor_sleep_interrupt(void *ptr)

-{

- rb_execution_context_t *ec = ptr;

- rb_ractor_t *r = rb_ec_ractor_ptr(ec);

- rb_thread_t *th = rb_ec_thread_ptr(ec);

-

- RACTOR_LOCK(r);

- {

- ractor_wakeup(r, th, wait_receiving | wait_taking | wait_yielding, wakeup_by_interrupt);

- }

- RACTOR_UNLOCK(r);

-}

-

-typedef void (*ractor_sleep_cleanup_function)(rb_ractor_t *cr, void *p);

-

-// Checks the current thread for ruby interrupts and runs the cleanup function `cf_func` with `cf_data` if

-// `rb_ec_check_ints` is going to raise. See the `rb_threadptr_execute_interrupts` for info on when it can raise.

-static void

-ractor_check_ints(rb_execution_context_t *ec, rb_ractor_t *cr, rb_thread_t *cur_th, ractor_sleep_cleanup_function cf_func, void *cf_data)

-{

- if (cur_th->ractor_waiting.wait_status != wait_none) {

- enum rb_ractor_wait_status prev_wait_status = cur_th->ractor_waiting.wait_status;

- cur_th->ractor_waiting.wait_status = wait_none;

- cur_th->ractor_waiting.wakeup_status = wakeup_by_interrupt;

-

- RACTOR_UNLOCK(cr);

- {

- if (cf_func) {

- enum ruby_tag_type state;

- EC_PUSH_TAG(ec);

- if ((state = EC_EXEC_TAG()) == TAG_NONE) {

- rb_ec_check_ints(ec);

- }

- EC_POP_TAG();

-

- if (state) {

- (*cf_func)(cr, cf_data); // cleanup function is run after the ubf, if it had ubf

- EC_JUMP_TAG(ec, state);

- }

- }

- else {

- rb_ec_check_ints(ec);

- }

- }

-

- RACTOR_LOCK(cr);

- cur_th->ractor_waiting.wait_status = prev_wait_status;

- }

-}

-

-#ifdef RUBY_THREAD_PTHREAD_H

-void rb_ractor_sched_sleep(rb_execution_context_t *ec, rb_ractor_t *cr, rb_unblock_function_t *ubf);

-#else

-

-static void

-ractor_cond_wait(rb_ractor_t *r, rb_thread_t *th)

-{

-#if RACTOR_CHECK_MODE > 0

- VALUE locked_by = r->sync.locked_by;

- r->sync.locked_by = Qnil;

-#endif

- rb_native_cond_wait(&th->ractor_waiting.cond, &r->sync.lock);

-

-#if RACTOR_CHECK_MODE > 0

- r->sync.locked_by = locked_by;

-#endif

-}

-

-static void *

-ractor_sleep_wo_gvl(void *ptr)

-{

- rb_ractor_t *cr = ptr;

- rb_execution_context_t *ec = cr->threads.running_ec;

- VM_ASSERT(GET_EC() == ec);

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

- RACTOR_LOCK_SELF(cr);

- {

- VM_ASSERT(cur_th->ractor_waiting.wait_status != wait_none);

- // it's possible that another ractor has woken us up (ractor_wakeup),

- // so check this condition

- if (cur_th->ractor_waiting.wakeup_status == wakeup_none) {

- cur_th->status = THREAD_STOPPED_FOREVER;

- ractor_cond_wait(cr, cur_th);

- cur_th->status = THREAD_RUNNABLE;

- VM_ASSERT(cur_th->ractor_waiting.wakeup_status != wakeup_none);

- }

- else {

- RUBY_DEBUG_LOG("rare timing, no cond wait");

- }

- cur_th->ractor_waiting.wait_status = wait_none;

- }

- RACTOR_UNLOCK_SELF(cr);

- return NULL;

-}

-

-static void

-rb_ractor_sched_sleep(rb_execution_context_t *ec, rb_ractor_t *cr, rb_unblock_function_t *ubf_ractor_sleep_interrupt)

-{

- ASSERT_ractor_locking(cr);

- rb_thread_t *th = rb_ec_thread_ptr(ec);

- struct ccan_list_node *waitn = &th->ractor_waiting.waiting_node;

- VM_ASSERT(waitn->next == waitn->prev && waitn->next == waitn); // it should be unlinked

- ccan_list_add(&cr->sync.wait.waiting_threads, waitn);

- RACTOR_UNLOCK(cr);

- {

- rb_nogvl(ractor_sleep_wo_gvl, cr, ubf_ractor_sleep_interrupt, ec, RB_NOGVL_INTR_FAIL);

- }

- RACTOR_LOCK(cr);

- ccan_list_del_init(waitn);

-}

-#endif

-

-/*

- * Sleep the current ractor's current thread until another ractor wakes us up or another thread calls our unblock function.

- * The following ractor actions can cause this function to be called:

- * Ractor#take (wait_taking)

- * Ractor.yield (wait_yielding)

- * Ractor.receive (wait_receiving)

- * Ractor.select (can be a combination of the above wait states, depending on the states of the ractors passed to Ractor.select)

- */

-static enum rb_ractor_wakeup_status

-ractor_sleep_with_cleanup(rb_execution_context_t *ec, rb_ractor_t *cr, rb_thread_t *cur_th, enum rb_ractor_wait_status wait_status,

- ractor_sleep_cleanup_function cf_func, void *cf_data)

-{

- ASSERT_ractor_locking(cr);

- enum rb_ractor_wakeup_status wakeup_status;

- VM_ASSERT(GET_RACTOR() == cr);

-

- VM_ASSERT(cur_th->ractor_waiting.wait_status == wait_none);

- VM_ASSERT(wait_status != wait_none);

- cur_th->ractor_waiting.wait_status = wait_status;

- cur_th->ractor_waiting.wakeup_status = wakeup_none;

-

- // fprintf(stderr, "%s r:%p status:%s, wakeup_status:%s\n", RUBY_FUNCTION_NAME_STRING, (void *)cr,

- // wait_status_str(cr->sync.wait.status), wakeup_status_str(cr->sync.wait.wakeup_status));

-

- RUBY_DEBUG_LOG("sleep by %s", wait_status_str(wait_status));

-

- while (cur_th->ractor_waiting.wakeup_status == wakeup_none) {

- rb_ractor_sched_sleep(ec, cr, ractor_sleep_interrupt);

- ractor_check_ints(ec, cr, cur_th, cf_func, cf_data);

- }

-

- cur_th->ractor_waiting.wait_status = wait_none;

-

- wakeup_status = cur_th->ractor_waiting.wakeup_status;

- cur_th->ractor_waiting.wakeup_status = wakeup_none;

-

- RUBY_DEBUG_LOG("wakeup %s", wakeup_status_str(wakeup_status));

-

- ASSERT_ractor_locking(cr);

- return wakeup_status;

-}

-

-static enum rb_ractor_wakeup_status

-ractor_sleep(rb_execution_context_t *ec, rb_ractor_t *cr, rb_thread_t *cur_th, enum rb_ractor_wait_status wait_status)

-{

- return ractor_sleep_with_cleanup(ec, cr, cur_th, wait_status, 0, NULL);

-}

-

-// Ractor.receive

-

-static void

-ractor_recursive_receive_if(rb_thread_t *th)

-{

- if (th->ractor_waiting.receiving_mutex && rb_mutex_owned_p(th->ractor_waiting.receiving_mutex)) {

- rb_raise(rb_eRactorError, "can not call receive/receive_if recursively");

- }

-}

-

-static VALUE

-ractor_try_receive(rb_execution_context_t *ec, rb_ractor_t *cr, struct rb_ractor_queue *rq)

-{

- struct rb_ractor_basket basket;

- ractor_recursive_receive_if(rb_ec_thread_ptr(ec));

- bool received = false;

-

- RACTOR_LOCK_SELF(cr);

- {

- RUBY_DEBUG_LOG("rq->cnt:%d", rq->cnt);

- received = ractor_queue_deq(cr, rq, &basket);

- }

- RACTOR_UNLOCK_SELF(cr);

-

- if (!received) {

- if (cr->sync.incoming_port_closed) {

- rb_raise(rb_eRactorClosedError, "The incoming port is already closed");

- }

- return Qundef;

- }

- else {

- return ractor_basket_accept(&basket);

- }

-}

-

-static void

-ractor_wait_receive(rb_execution_context_t *ec, rb_ractor_t *cr, struct rb_ractor_queue *rq)

-{

- VM_ASSERT(cr == rb_ec_ractor_ptr(ec));

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

- ractor_recursive_receive_if(cur_th);

-

- RACTOR_LOCK(cr);

- {

- while (ractor_queue_empty_p(cr, rq) && !cr->sync.incoming_port_closed) {

- ractor_sleep(ec, cr, cur_th, wait_receiving);

- }

- }

- RACTOR_UNLOCK(cr);

-}

-

-static VALUE

-ractor_receive(rb_execution_context_t *ec, rb_ractor_t *cr)

-{

- VM_ASSERT(cr == rb_ec_ractor_ptr(ec));

- VALUE v;

- struct rb_ractor_queue *rq = &cr->sync.recv_queue;

-

- while (UNDEF_P(v = ractor_try_receive(ec, cr, rq))) {

- ractor_wait_receive(ec, cr, rq);

- }

-

- return v;

-}

-

-#if 0

-static void

-rq_dump(struct rb_ractor_queue *rq)

-{

- bool bug = false;

- for (int i=0; i<rq->cnt; i++) {

- struct rb_ractor_basket *b = ractor_queue_at(NULL, rq, i);

- fprintf(stderr, "%d (start:%d) type:%s %p %s\n", i, rq->start, basket_type_name(b->type),

- (void *)b, RSTRING_PTR(RARRAY_AREF(b->v, 1)));

- if (basket_type_p(b, basket_type_reserved) bug = true;

- }

- if (bug) rb_bug("!!");

-}

-#endif

-

-struct receive_block_data {

- rb_ractor_t *cr;

- rb_thread_t *th;

- struct rb_ractor_queue *rq;

- VALUE v;

- int index;

- bool success;

-};

-

-static void

-ractor_receive_if_lock(rb_thread_t *th)

-{

- VALUE m = th->ractor_waiting.receiving_mutex;

- if (m == Qfalse) {

- m = th->ractor_waiting.receiving_mutex = rb_mutex_new();

- }

- rb_mutex_lock(m);

-}

-

-static VALUE

-receive_if_body(VALUE ptr)

-{

- struct receive_block_data *data = (struct receive_block_data *)ptr;

-

- ractor_receive_if_lock(data->th);

- VALUE block_result = rb_yield(data->v);

- rb_ractor_t *cr = data->cr;

-

- RACTOR_LOCK_SELF(cr);

- {

- struct rb_ractor_basket *b = ractor_queue_at(cr, data->rq, data->index);

- VM_ASSERT(basket_type_p(b, basket_type_reserved));

- data->rq->reserved_cnt--;

-

- if (RTEST(block_result)) {

- ractor_queue_delete(cr, data->rq, b);

- ractor_queue_compact(cr, data->rq);

- }

- else {

- b->type.e = basket_type_ref;

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-

- data->success = true;

-

- if (RTEST(block_result)) {

- return data->v;

- }

- else {

- return Qundef;

- }

-}

-

-static VALUE

-receive_if_ensure(VALUE v)

-{

- struct receive_block_data *data = (struct receive_block_data *)v;

- rb_ractor_t *cr = data->cr;

- rb_thread_t *cur_th = data->th;

-

- if (!data->success) {

- RACTOR_LOCK_SELF(cr);

- {

- struct rb_ractor_basket *b = ractor_queue_at(cr, data->rq, data->index);

- VM_ASSERT(basket_type_p(b, basket_type_reserved));

- b->type.e = basket_type_deleted;

- data->rq->reserved_cnt--;

- }

- RACTOR_UNLOCK_SELF(cr);

- }

-

- rb_mutex_unlock(cur_th->ractor_waiting.receiving_mutex);

- return Qnil;

-}

-

-static VALUE

-ractor_receive_if(rb_execution_context_t *ec, VALUE crv, VALUE b)

-{

- if (!RTEST(b)) rb_raise(rb_eArgError, "no block given");

-

- rb_ractor_t *cr = rb_ec_ractor_ptr(ec);

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

- unsigned int serial = (unsigned int)-1;

- int index = 0;

- struct rb_ractor_queue *rq = &cr->sync.recv_queue;

-

- while (1) {

- VALUE v = Qundef;

-

- ractor_wait_receive(ec, cr, rq);

-

- RACTOR_LOCK_SELF(cr);

- {

- if (serial != rq->serial) {

- serial = rq->serial;

- index = 0;

- }

-

- // check newer version

- for (int i=index; i<rq->cnt; i++) {

- if (!ractor_queue_skip_p(cr, rq, i)) {

- struct rb_ractor_basket *b = ractor_queue_at(cr, rq, i);

- v = ractor_basket_value(b);

- b->type.e = basket_type_reserved;

- rq->reserved_cnt++;

- index = i;

- break;

- }

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-

- if (!UNDEF_P(v)) {

- struct receive_block_data data = {

- .cr = cr,

- .th = cur_th,

- .rq = rq,

- .v = v,

- .index = index,

- .success = false,

- };

-

- VALUE result = rb_ensure(receive_if_body, (VALUE)&data,

- receive_if_ensure, (VALUE)&data);

-

- if (!UNDEF_P(result)) return result;

- index++;

- }

-

- RUBY_VM_CHECK_INTS(ec);

- }

-}

-

-static void

-ractor_send_basket(rb_execution_context_t *ec, rb_ractor_t *r, struct rb_ractor_basket *b)

-{

- bool closed = false;

-

- RACTOR_LOCK(r);

- {

- if (r->sync.incoming_port_closed) {

- closed = true;

- }

- else {

- ractor_queue_enq(r, &r->sync.recv_queue, b);

- // wakeup any receiving thread in `r`

- ractor_wakeup(r, NULL, wait_receiving, wakeup_by_send);

- }

- }

- RACTOR_UNLOCK(r);

-

- if (closed) {

- rb_raise(rb_eRactorClosedError, "The incoming-port is already closed");

- }

-}

-

-// Ractor#send

-

-static VALUE ractor_move(VALUE obj); // in this file

-static VALUE ractor_copy(VALUE obj); // in this file

-

-static void

-ractor_basket_prepare_contents(VALUE obj, VALUE move, volatile VALUE *pobj, enum rb_ractor_basket_type *ptype)

-{

- VALUE v;

- enum rb_ractor_basket_type type;

-

- if (rb_ractor_shareable_p(obj)) {

- type = basket_type_ref;

- v = obj;

- }

- else if (!RTEST(move)) {

- v = ractor_copy(obj);

- type = basket_type_copy;

- }

- else {

- type = basket_type_move;

- v = ractor_move(obj);

- }

-

- *pobj = v;

- *ptype = type;

-}

-

-static void

-ractor_basket_fill_(rb_ractor_t *cr, rb_thread_t *cur_th, struct rb_ractor_basket *basket, VALUE obj, bool exc)

-{

- VM_ASSERT(cr == GET_RACTOR());

-

- basket->sender = cr->pub.self;

- basket->sending_th = cur_th;

- basket->p.send.exception = exc;

- basket->p.send.v = obj;

-}

-

-static void

-ractor_basket_fill(rb_ractor_t *cr, rb_thread_t *cur_th, struct rb_ractor_basket *basket, VALUE obj, VALUE move, bool exc)

-{

- VALUE v;

- enum rb_ractor_basket_type type;

- ractor_basket_prepare_contents(obj, move, &v, &type);

- ractor_basket_fill_(cr, cur_th, basket, v, exc);

- basket->type.e = type;

-}

-

-static void

-ractor_basket_fill_will(rb_ractor_t *cr, rb_thread_t *cur_th, struct rb_ractor_basket *basket, VALUE obj, bool exc)

-{

- ractor_basket_fill_(cr, cur_th, basket, obj, exc);

- basket->type.e = basket_type_will;

-}

-

-static VALUE

-ractor_send(rb_execution_context_t *ec, rb_ractor_t *recv_r, VALUE obj, VALUE move)

-{

- struct rb_ractor_basket basket;

- rb_ractor_t *cr = rb_ec_ractor_ptr(ec);

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

- // TODO: Ractor local GC

- ractor_basket_fill(cr, cur_th, &basket, obj, move, false);

- ractor_send_basket(ec, recv_r, &basket);

- return recv_r->pub.self;

-}

-

-// Ractor#take

-

-static bool

-ractor_take_has_will(rb_ractor_t *r)

-{

- ASSERT_ractor_locking(r);

-

- return basket_type_p(&r->sync.will_basket, basket_type_will);

-}

-

-static bool

-ractor_take_will(rb_ractor_t *r, struct rb_ractor_basket *b)

-{

- ASSERT_ractor_locking(r);

-

- if (ractor_take_has_will(r)) {

- *b = r->sync.will_basket;

- r->sync.will_basket.type.e = basket_type_none;

- return true;

- }

- else {

- VM_ASSERT(basket_type_p(&r->sync.will_basket, basket_type_none));

- return false;

- }

-}

-

-static bool

-ractor_take_will_lock(rb_ractor_t *r, struct rb_ractor_basket *b)

-{

- ASSERT_ractor_unlocking(r);

- bool taken;

-

- RACTOR_LOCK(r);

- {

- taken = ractor_take_will(r, b);

- }

- RACTOR_UNLOCK(r);

-

- return taken;

-}

-

-static bool

-ractor_register_take(rb_ractor_t *cr, rb_thread_t *cur_th, rb_ractor_t *r, struct rb_ractor_basket *take_basket,

- bool is_take, struct rb_ractor_selector_take_config *config, bool ignore_error)

-{

- struct rb_ractor_basket b = {

- .type.e = basket_type_take_basket,

- .sender = cr->pub.self,

- .sending_th = cur_th,

- .p = {

- .take = {

- .basket = take_basket, // pointer to our stack value saved in ractor `r` queue

- .config = config,

- },

- },

- };

- bool closed = false;

-

- RACTOR_LOCK(r);

- {

- if (is_take && ractor_take_will(r, take_basket)) {

- RUBY_DEBUG_LOG("take over a will of r:%d", rb_ractor_id(r));

- }

- else if (!is_take && ractor_take_has_will(r)) {

- RUBY_DEBUG_LOG("has_will");

- VM_ASSERT(config != NULL);

- config->closed = true;

- }

- else if (r->sync.outgoing_port_closed) {

- closed = true;

- }

- else {

- RUBY_DEBUG_LOG("register in r:%d", rb_ractor_id(r));

- ractor_queue_enq(r, &r->sync.takers_queue, &b);

-

- if (basket_none_p(take_basket)) {

- // wakeup any thread in `r` that has yielded, if there is any.

- ractor_wakeup(r, NULL, wait_yielding, wakeup_by_take);

- }

- }

- }

- RACTOR_UNLOCK(r);

-

- if (closed) {

- if (!ignore_error) rb_raise(rb_eRactorClosedError, "The outgoing-port is already closed");

- return false;

- }

- else {

- return true;

- }

-}

-

-static bool

-ractor_deregister_take(rb_ractor_t *r, struct rb_ractor_basket *take_basket)

-{

- struct rb_ractor_queue *ts = &r->sync.takers_queue;

- bool deleted = false;

-

- RACTOR_LOCK(r);

- {

- if (r->sync.outgoing_port_closed) {

- // ok

- }

- else {

- for (int i=0; i<ts->cnt; i++) {

- struct rb_ractor_basket *b = ractor_queue_at(r, ts, i);

- if (basket_type_p(b, basket_type_take_basket) && b->p.take.basket == take_basket) {

- ractor_queue_delete(r, ts, b);

- deleted = true;

- }

- }

- if (deleted) {

- ractor_queue_compact(r, ts);

- }

- }

- }

- RACTOR_UNLOCK(r);

-

- return deleted;

-}

-

-static VALUE

-ractor_try_take(rb_ractor_t *cr, rb_thread_t *cur_th, rb_ractor_t *recv_r, struct rb_ractor_basket *take_basket)

-{

- bool taken;

-

- RACTOR_LOCK_SELF(cr);

- {

- // If it hasn't yielded yet or is currently in the process of yielding, sleep more

- if (basket_none_p(take_basket) || basket_type_p(take_basket, basket_type_yielding)) {

- taken = false;

- }

- else {

- taken = true; // basket type might be, for ex, basket_type_copy if value was copied during yield

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-

- if (taken) {

- RUBY_DEBUG_LOG("taken");

- if (basket_type_p(take_basket, basket_type_deleted)) {

- VM_ASSERT(recv_r->sync.outgoing_port_closed);

- rb_raise(rb_eRactorClosedError, "The outgoing-port is already closed");

- }

- return ractor_basket_accept(take_basket);

- }

- else {

- RUBY_DEBUG_LOG("not taken");

- return Qundef;

- }

-}

-

-

-#if VM_CHECK_MODE > 0

-static bool

-ractor_check_specific_take_basket_lock(rb_ractor_t *r, struct rb_ractor_basket *tb)

-{

- bool ret = false;

- struct rb_ractor_queue *ts = &r->sync.takers_queue;

-

- RACTOR_LOCK(r);

- {

- for (int i=0; i<ts->cnt; i++) {

- struct rb_ractor_basket *b = ractor_queue_at(r, ts, i);

- if (basket_type_p(b, basket_type_take_basket) && b->p.take.basket == tb) {

- ret = true;

- break;

- }

- }

- }

- RACTOR_UNLOCK(r);

-

- return ret;

-}

-#endif

-

-// cleanup function, cr is unlocked

-static void

-ractor_take_cleanup(rb_ractor_t *cr, rb_ractor_t *r, struct rb_ractor_basket *tb)

-{

- retry:

- if (basket_none_p(tb)) { // not yielded yet

- if (!ractor_deregister_take(r, tb)) {

- // not in r's takers queue

- rb_thread_sleep(0);

- goto retry;

- }

- }

- else {

- VM_ASSERT(!ractor_check_specific_take_basket_lock(r, tb));

- }

-}

-

-struct take_wait_take_cleanup_data {

- rb_ractor_t *r;

- struct rb_ractor_basket *tb;

-};

-

-static void

-ractor_wait_take_cleanup(rb_ractor_t *cr, void *ptr)

-{

- struct take_wait_take_cleanup_data *data = (struct take_wait_take_cleanup_data *)ptr;

- ractor_take_cleanup(cr, data->r, data->tb);

-}

-

-static void

-ractor_wait_take(rb_execution_context_t *ec, rb_ractor_t *cr, rb_thread_t *cur_th, rb_ractor_t *r, struct rb_ractor_basket *take_basket)

-{

- struct take_wait_take_cleanup_data data = {

- .r = r,

- .tb = take_basket,

- };

-

- RACTOR_LOCK_SELF(cr);

- {

- if (basket_none_p(take_basket) || basket_type_p(take_basket, basket_type_yielding)) {

- ractor_sleep_with_cleanup(ec, cr, cur_th, wait_taking, ractor_wait_take_cleanup, &data);

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-}

-

-static VALUE

-ractor_take(rb_execution_context_t *ec, rb_ractor_t *recv_r)

-{

- RUBY_DEBUG_LOG("from r:%u", rb_ractor_id(recv_r));

- VALUE v;

- rb_ractor_t *cr = rb_ec_ractor_ptr(ec);

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

-

- struct rb_ractor_basket take_basket = {

- .type.e = basket_type_none,

- .sender = 0,

- };

-

- ractor_register_take(cr, cur_th, recv_r, &take_basket, true, NULL, false);

-

- while (UNDEF_P(v = ractor_try_take(cr, cur_th, recv_r, &take_basket))) {

- ractor_wait_take(ec, cr, cur_th, recv_r, &take_basket);

- }

-

- VM_ASSERT(!basket_none_p(&take_basket)); // might be, for ex, basket_type_copy

- VM_ASSERT(!ractor_check_specific_take_basket_lock(recv_r, &take_basket));

-

- return v;

-}

-

-// Ractor.yield

-

-static bool

-ractor_check_take_basket(rb_ractor_t *cr, struct rb_ractor_queue *rs)

-{

- ASSERT_ractor_locking(cr);

-

- for (int i=0; i<rs->cnt; i++) {

- struct rb_ractor_basket *b = ractor_queue_at(cr, rs, i);

- if (basket_type_p(b, basket_type_take_basket) &&

- basket_none_p(b->p.take.basket)) {

- return true;

- }

- }

-

- return false;

-}

-

-// Find another ractor that is taking from this ractor, so we can yield to it

-static bool

-ractor_deq_take_basket(rb_ractor_t *cr, struct rb_ractor_queue *rs, struct rb_ractor_basket *b)

-{

- ASSERT_ractor_unlocking(cr);

- struct rb_ractor_basket *first_tb = NULL;

- bool found = false;

-

- RACTOR_LOCK_SELF(cr);

- {

- while (ractor_queue_deq(cr, rs, b)) {

- if (basket_type_p(b, basket_type_take_basket)) { // some other ractor is taking

- struct rb_ractor_basket *tb = b->p.take.basket;

-

- if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_none, basket_type_yielding) == basket_type_none) {

- found = true; // payload basket is now "yielding" type

- break;

- }

- else {

- ractor_queue_enq(cr, rs, b);

- if (first_tb == NULL) first_tb = tb;

- struct rb_ractor_basket *head = ractor_queue_head(cr, rs);

- VM_ASSERT(head != NULL);

- if (basket_type_p(head, basket_type_take_basket) && head->p.take.basket == first_tb) {

- break; // loop detected

- }

- }

- }

- else {

- VM_ASSERT(basket_none_p(b));

- }

- }

-

- if (found && b->p.take.config && !b->p.take.config->oneshot) {

- ractor_queue_enq(cr, rs, b);

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-

- return found;

-}

-

-// Try yielding to a taking ractor

-static bool

-ractor_try_yield(rb_execution_context_t *ec, rb_ractor_t *cr, struct rb_ractor_queue *ts, volatile VALUE obj, VALUE move, bool exc, bool is_will)

-{

- // Don't lock yielding ractor at same time as taking ractor. This could deadlock due to timing

- // issue because we don't have a lock hierarchy.

- ASSERT_ractor_unlocking(cr);

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

-

- struct rb_ractor_basket b;

-

- if (ractor_deq_take_basket(cr, ts, &b)) { // deq a take basket from takers queue of `cr` into `b`

- VM_ASSERT(basket_type_p(&b, basket_type_take_basket));

- VM_ASSERT(basket_type_p(b.p.take.basket, basket_type_yielding));

-

- rb_ractor_t *tr = RACTOR_PTR(b.sender); // taking ractor

- rb_thread_t *tr_th = b.sending_th; // taking thread

- struct rb_ractor_basket *tb = b.p.take.basket; // payload basket

- enum rb_ractor_basket_type type;

-

- RUBY_DEBUG_LOG("basket from r:%u", rb_ractor_id(tr));

-

- if (is_will) {

- type = basket_type_will; // last message

- }

- else {

- enum ruby_tag_type state;

-

- // begin

- EC_PUSH_TAG(ec);

- if ((state = EC_EXEC_TAG()) == TAG_NONE) {

- // TODO: Ractor local GC

- ractor_basket_prepare_contents(obj, move, &obj, &type);

- }

- EC_POP_TAG();

- // rescue ractor copy/move error, then re-raise

- if (state) {

- RACTOR_LOCK_SELF(cr);

- {

- b.p.take.basket->type.e = basket_type_none;

- ractor_queue_enq(cr, ts, &b);

- }

- RACTOR_UNLOCK_SELF(cr);

- EC_JUMP_TAG(ec, state);

- }

- }

-

- RACTOR_LOCK(tr);

- {

- VM_ASSERT(basket_type_p(tb, basket_type_yielding));

- // fill atomic

- RUBY_DEBUG_LOG("fill %sbasket from r:%u", is_will ? "will " : "", rb_ractor_id(tr));

- ractor_basket_fill_(cr, cur_th, tb, obj, exc); // fill the take basket payload

- if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_yielding, type) != basket_type_yielding) {

- rb_bug("unreachable");

- }

- ractor_wakeup(tr, tr_th, wait_taking, wakeup_by_yield);

- }

- RACTOR_UNLOCK(tr);

-

- return true;

- }

- else if (cr->sync.outgoing_port_closed) {

- rb_raise(rb_eRactorClosedError, "The outgoing-port is already closed");

- }

- else {

- RUBY_DEBUG_LOG("no take basket");

- return false;

- }

-}

-

-static void

-ractor_wait_yield(rb_execution_context_t *ec, rb_ractor_t *cr, struct rb_ractor_queue *ts)

-{

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

- RACTOR_LOCK_SELF(cr);

- {

- while (!ractor_check_take_basket(cr, ts) && !cr->sync.outgoing_port_closed) {

- ractor_sleep(ec, cr, cur_th, wait_yielding);

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-}

-

-// In order to yield, we wait until our takers queue has at least one element. Then, we wakeup a taker.

-static VALUE

-ractor_yield(rb_execution_context_t *ec, rb_ractor_t *cr, VALUE obj, VALUE move)

-{

- struct rb_ractor_queue *ts = &cr->sync.takers_queue;

-

- while (!ractor_try_yield(ec, cr, ts, obj, move, false, false)) {

- ractor_wait_yield(ec, cr, ts);

- }

-

- return Qnil;

-}

-

-// Ractor::Selector

-

-struct rb_ractor_selector {

- rb_ractor_t *r;

- struct rb_ractor_basket take_basket;

- st_table *take_ractors; // rb_ractor_t * => (struct rb_ractor_selector_take_config *)

-};

-

-static int

-ractor_selector_mark_ractors_i(st_data_t key, st_data_t value, st_data_t data)

-{

- const rb_ractor_t *r = (rb_ractor_t *)key;

- rb_gc_mark(r->pub.self);

- return ST_CONTINUE;

-}

-

-static void

-ractor_selector_mark(void *ptr)

-{

- struct rb_ractor_selector *s = ptr;

-

- if (s->take_ractors) {

- st_foreach(s->take_ractors, ractor_selector_mark_ractors_i, 0);

- }

-

- switch (s->take_basket.type.e) {

- case basket_type_ref:

- case basket_type_copy:

- case basket_type_move:

- case basket_type_will:

- rb_gc_mark(s->take_basket.sender);

- rb_gc_mark(s->take_basket.p.send.v);

- break;

- default:

- break;

- }

-}

-

-static int

-ractor_selector_release_i(st_data_t key, st_data_t val, st_data_t data)

-{

- struct rb_ractor_selector *s = (struct rb_ractor_selector *)data;

- struct rb_ractor_selector_take_config *config = (struct rb_ractor_selector_take_config *)val;

-

- if (!config->closed) {

- ractor_deregister_take((rb_ractor_t *)key, &s->take_basket);

- }

- free(config);

- return ST_CONTINUE;

-}

-

-static void

-ractor_selector_free(void *ptr)

-{

- struct rb_ractor_selector *s = ptr;

- st_foreach(s->take_ractors, ractor_selector_release_i, (st_data_t)s);

- st_free_table(s->take_ractors);

- ruby_xfree(ptr);

-}

-

-static size_t

-ractor_selector_memsize(const void *ptr)

-{

- const struct rb_ractor_selector *s = ptr;

- return sizeof(struct rb_ractor_selector) +

- st_memsize(s->take_ractors) +

- s->take_ractors->num_entries * sizeof(struct rb_ractor_selector_take_config);

-}

-

-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_FREE_IMMEDIATELY,

-};

-

-static struct rb_ractor_selector *

-RACTOR_SELECTOR_PTR(VALUE selv)

-{

- VM_ASSERT(rb_typeddata_is_kind_of(selv, &ractor_selector_data_type));

-

- return (struct rb_ractor_selector *)DATA_PTR(selv);

-}

-

-// Ractor::Selector.new

-

-static VALUE

-ractor_selector_create(VALUE klass)

-{

- struct rb_ractor_selector *s;

- VALUE selv = TypedData_Make_Struct(klass, struct rb_ractor_selector, &ractor_selector_data_type, s);

- s->take_basket.type.e = basket_type_reserved;

- s->take_ractors = st_init_numtable(); // ractor (ptr) -> take_config

- 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 rv)

-{

- if (!rb_ractor_p(rv)) {

- rb_raise(rb_eArgError, "Not a ractor object");

- }

-

- rb_ractor_t *r = RACTOR_PTR(rv);

- struct rb_ractor_selector *s = RACTOR_SELECTOR_PTR(selv);

-

- if (st_lookup(s->take_ractors, (st_data_t)r, NULL)) {

- rb_raise(rb_eArgError, "already added");

- }

-

- struct rb_ractor_selector_take_config *config = malloc(sizeof(struct rb_ractor_selector_take_config));

- VM_ASSERT(config != NULL);

- config->closed = false;

- config->oneshot = false;

-

- if (ractor_register_take(GET_RACTOR(), GET_THREAD(), r, &s->take_basket, false, config, true)) {

- st_insert(s->take_ractors, (st_data_t)r, (st_data_t)config);

- }

-

- return rv;

-}

-

-// 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 rv)

-{

- if (!rb_ractor_p(rv)) {

- rb_raise(rb_eArgError, "Not a ractor object");

- }

-

- rb_ractor_t *r = RACTOR_PTR(rv);

- struct rb_ractor_selector *s = RACTOR_SELECTOR_PTR(selv);

-

- RUBY_DEBUG_LOG("r:%u", rb_ractor_id(r));

-

- if (!st_lookup(s->take_ractors, (st_data_t)r, NULL)) {

- rb_raise(rb_eArgError, "not added yet");

- }

-

- ractor_deregister_take(r, &s->take_basket);

- struct rb_ractor_selector_take_config *config;

- st_delete(s->take_ractors, (st_data_t *)&r, (st_data_t *)&config);

- free(config);

-

- return rv;

-}

-

-// Ractor::Selector#clear

-

-struct ractor_selector_clear_data {

- VALUE selv;

- rb_execution_context_t *ec;

-};

-

-static int

-ractor_selector_clear_i(st_data_t key, st_data_t val, st_data_t data)

-{

- VALUE selv = (VALUE)data;

- rb_ractor_t *r = (rb_ractor_t *)key;

- ractor_selector_remove(selv, r->pub.self);

- return ST_CONTINUE;

-}

-

-/*

- * call-seq:

- * clear -> self

- *

- * Removes all ractors from +self+. Raises +self+.

- */

-static VALUE

-ractor_selector_clear(VALUE selv)

-{

- struct rb_ractor_selector *s = RACTOR_SELECTOR_PTR(selv);

-

- st_foreach(s->take_ractors, ractor_selector_clear_i, (st_data_t)selv);

- st_clear(s->take_ractors);

- return selv;

-}

-

-/*

- * call-seq:

- * empty? -> true or false

- *

- * Returns +true+ if no ractor is added.

- */

-static VALUE

-ractor_selector_empty_p(VALUE selv)

-{

- struct rb_ractor_selector *s = RACTOR_SELECTOR_PTR(selv);

- return s->take_ractors->num_entries == 0 ? Qtrue : Qfalse;

-}

-

-static int

-ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t dat)

-{

- rb_ractor_t *r = (rb_ractor_t *)key;

- struct rb_ractor_basket *tb = (struct rb_ractor_basket *)dat;

- int ret;

-

- if (!basket_none_p(tb)) {

- RUBY_DEBUG_LOG("already taken:%s", basket_type_name(tb->type.e));

- return ST_STOP;

- }

-

- RACTOR_LOCK(r);

- {

- if (basket_type_p(&r->sync.will_basket, basket_type_will)) {

- RUBY_DEBUG_LOG("r:%u has will", rb_ractor_id(r));

-

- if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_none, basket_type_will) == basket_type_none) {

- ractor_take_will(r, tb);

- ret = ST_STOP;

- }

- else {

- RUBY_DEBUG_LOG("has will, but already taken (%s)", basket_type_name(tb->type.e));

- ret = ST_CONTINUE;

- }

- }

- else if (r->sync.outgoing_port_closed) {

- RUBY_DEBUG_LOG("r:%u is closed", rb_ractor_id(r));

-

- if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_none, basket_type_deleted) == basket_type_none) {

- tb->sender = r->pub.self;

- ret = ST_STOP;

- }

- else {

- RUBY_DEBUG_LOG("closed, but already taken (%s)", basket_type_name(tb->type.e));

- ret = ST_CONTINUE;

- }

- }

- else {

- RUBY_DEBUG_LOG("wakeup r:%u", rb_ractor_id(r));

- ractor_wakeup(r, NULL, wait_yielding, wakeup_by_take);

- ret = ST_CONTINUE;

- }

- }

- RACTOR_UNLOCK(r);

-

- return ret;

-}

-

-// Ractor::Selector#wait

-

-// cleanup function, cr is unlocked

-static void

-ractor_selector_wait_cleanup(rb_ractor_t *cr, void *ptr)

-{

- struct rb_ractor_basket *tb = (struct rb_ractor_basket *)ptr;

-

- RACTOR_LOCK_SELF(cr);

- {

- while (basket_type_p(tb, basket_type_yielding)) {

- RACTOR_UNLOCK_SELF(cr);

- {

- rb_thread_sleep(0);

- }

- RACTOR_LOCK_SELF(cr);

- }

- // if tb->type is not none, taking is succeeded, but interruption ignore it unfortunately.

- tb->type.e = basket_type_reserved;

- }

- RACTOR_UNLOCK_SELF(cr);

-}

-

-/* :nodoc: */

-static VALUE

-ractor_selector__wait(VALUE selv, VALUE do_receivev, VALUE do_yieldv, VALUE yield_value, VALUE move)

-{

- rb_execution_context_t *ec = GET_EC();

- struct rb_ractor_selector *s = RACTOR_SELECTOR_PTR(selv);

- struct rb_ractor_basket *tb = &s->take_basket;

- struct rb_ractor_basket taken_basket;

- rb_ractor_t *cr = rb_ec_ractor_ptr(ec);

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

- bool do_receive = !!RTEST(do_receivev);

- bool do_yield = !!RTEST(do_yieldv);

- VALUE ret_v, ret_r;

- enum rb_ractor_wait_status wait_status;

- struct rb_ractor_queue *rq = &cr->sync.recv_queue;

- struct rb_ractor_queue *ts = &cr->sync.takers_queue;

-

- RUBY_DEBUG_LOG("start");

-

- retry:

- RUBY_DEBUG_LOG("takers:%ld", s->take_ractors->num_entries);

-

- // setup wait_status

- wait_status = wait_none;

- if (s->take_ractors->num_entries > 0) wait_status |= wait_taking;

- if (do_receive) wait_status |= wait_receiving;

- if (do_yield) wait_status |= wait_yielding;

-

- RUBY_DEBUG_LOG("wait:%s", wait_status_str(wait_status));

-

- if (wait_status == wait_none) {

- rb_raise(rb_eRactorError, "no taking ractors");

- }

-

- // check recv_queue

- if (do_receive && !UNDEF_P(ret_v = ractor_try_receive(ec, cr, rq))) {

- ret_r = ID2SYM(rb_intern("receive"));

- goto success;

- }

-

- // check takers

- if (do_yield && ractor_try_yield(ec, cr, ts, yield_value, move, false, false)) {

- ret_v = Qnil;

- ret_r = ID2SYM(rb_intern("yield"));

- goto success;

- }

-

- // check take_basket

- VM_ASSERT(basket_type_p(&s->take_basket, basket_type_reserved));

- s->take_basket.type.e = basket_type_none;

- // kick all take target ractors

- st_foreach(s->take_ractors, ractor_selector_wait_i, (st_data_t)tb);

-

- RACTOR_LOCK_SELF(cr);

- {

- retry_waiting:

- while (1) {

- if (!basket_none_p(tb)) {

- RUBY_DEBUG_LOG("taken:%s from r:%u", basket_type_name(tb->type.e),

- tb->sender ? rb_ractor_id(RACTOR_PTR(tb->sender)) : 0);

- break;

- }

- if (do_receive && !ractor_queue_empty_p(cr, rq)) {

- RUBY_DEBUG_LOG("can receive (%d)", rq->cnt);

- break;

- }

- if (do_yield && ractor_check_take_basket(cr, ts)) {

- RUBY_DEBUG_LOG("can yield");

- break;

- }

-

- ractor_sleep_with_cleanup(ec, cr, cur_th, wait_status, ractor_selector_wait_cleanup, tb);

- }

-

- taken_basket = *tb;

-

- // ensure

- // tb->type.e = basket_type_reserved # do it atomic in the following code

- if (taken_basket.type.e == basket_type_yielding ||

- RUBY_ATOMIC_CAS(tb->type.atomic, taken_basket.type.e, basket_type_reserved) != taken_basket.type.e) {

-

- if (basket_type_p(tb, basket_type_yielding)) {

- RACTOR_UNLOCK_SELF(cr);

- {

- rb_thread_sleep(0);

- }

- RACTOR_LOCK_SELF(cr);

- }

- goto retry_waiting;

- }

- }

- RACTOR_UNLOCK_SELF(cr);

-

- // check the taken result

- switch (taken_basket.type.e) {

- case basket_type_none:

- VM_ASSERT(do_receive || do_yield);

- goto retry;

- case basket_type_yielding:

- rb_bug("unreachable");

- case basket_type_deleted: {

- ractor_selector_remove(selv, taken_basket.sender);

-

- rb_ractor_t *r = RACTOR_PTR(taken_basket.sender);

- if (ractor_take_will_lock(r, &taken_basket)) {

- RUBY_DEBUG_LOG("has_will");

- }

- else {

- RUBY_DEBUG_LOG("no will");

- // rb_raise(rb_eRactorClosedError, "The outgoing-port is already closed");

- // remove and retry wait

- goto retry;

- }

- break;

- }

- case basket_type_will:

- // no more messages

- ractor_selector_remove(selv, taken_basket.sender);

- break;

- default:

- break;

- }

-

- RUBY_DEBUG_LOG("taken_basket:%s", basket_type_name(taken_basket.type.e));

-

- ret_v = ractor_basket_accept(&taken_basket);

- ret_r = taken_basket.sender;

- success:

- return rb_ary_new_from_args(2, ret_r, ret_v);

-}

-

-/*

- * 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(int argc, VALUE *argv, VALUE selector)

-{

- VALUE options;

- ID keywords[3];

- VALUE values[3];

-

- keywords[0] = rb_intern("receive");

- keywords[1] = rb_intern("yield_value");

- keywords[2] = rb_intern("move");

-

- rb_scan_args(argc, argv, "0:", &options);

- rb_get_kwargs(options, keywords, 0, numberof(values), values);

- return ractor_selector__wait(selector,

- values[0] == Qundef ? Qfalse : RTEST(values[0]),

- values[1] != Qundef, values[1], values[2]);

-}

-

-static VALUE

-ractor_selector_new(int argc, VALUE *ractors, VALUE klass)

-{

- VALUE selector = ractor_selector_create(klass);

-

- for (int i=0; i<argc; i++) {

- ractor_selector_add(selector, ractors[i]);

- }

-

- return selector;

-}

-

-static VALUE

-ractor_select_internal(rb_execution_context_t *ec, VALUE self, VALUE ractors, VALUE do_receive, VALUE do_yield, VALUE yield_value, VALUE move)

-{

- VALUE selector = ractor_selector_new(RARRAY_LENINT(ractors), (VALUE *)RARRAY_CONST_PTR(ractors), rb_cRactorSelector);

- VALUE result;

- int state;

-

- EC_PUSH_TAG(ec);

- if ((state = EC_EXEC_TAG()) == TAG_NONE) {

- result = ractor_selector__wait(selector, do_receive, do_yield, yield_value, move);

- }

- EC_POP_TAG();

- if (state != TAG_NONE) {

- // ensure

- ractor_selector_clear(selector);

-

- // jump

- EC_JUMP_TAG(ec, state);

- }

-

- RB_GC_GUARD(ractors);

- return result;

-}

-

-// Ractor#close_incoming

-

-static VALUE

-ractor_close_incoming(rb_execution_context_t *ec, rb_ractor_t *r)

-{

- VALUE prev;

- rb_thread_t *r_th = NULL;

- if (r == rb_ec_ractor_ptr(ec)) {

- r_th = rb_ec_thread_ptr(ec);

- }

-

- RACTOR_LOCK(r);

- {

- if (!r->sync.incoming_port_closed) {

- prev = Qfalse;

- r->sync.incoming_port_closed = true;

- if (ractor_wakeup(r, r_th, wait_receiving, wakeup_by_close)) {

- VM_ASSERT(ractor_queue_empty_p(r, &r->sync.recv_queue));

- RUBY_DEBUG_LOG("cancel receiving");

- }

- }

- else {

- prev = Qtrue;

- }

- }

- RACTOR_UNLOCK(r);

- return prev;

-}

-

-// Ractor#close_outgoing

-

-static VALUE

-ractor_close_outgoing(rb_execution_context_t *ec, rb_ractor_t *r)

-{

- VALUE prev;

-

- RACTOR_LOCK(r);

- {

- struct rb_ractor_queue *ts = &r->sync.takers_queue;

- rb_ractor_t *tr;

- struct rb_ractor_basket b;

-

- if (!r->sync.outgoing_port_closed) {

- prev = Qfalse;

- r->sync.outgoing_port_closed = true;

- }

- else {

- VM_ASSERT(ractor_queue_empty_p(r, ts));

- prev = Qtrue;

- }

-

- // wakeup all taking ractors

- while (ractor_queue_deq(r, ts, &b)) {

- if (basket_type_p(&b, basket_type_take_basket)) {

- tr = RACTOR_PTR(b.sender);

- rb_thread_t *tr_th = b.sending_th;

- struct rb_ractor_basket *tb = b.p.take.basket;

-

- if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_none, basket_type_yielding) == basket_type_none) {

- b.p.take.basket->sender = r->pub.self;

- if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_yielding, basket_type_deleted) != basket_type_yielding) {

- rb_bug("unreachable");

- }

- RUBY_DEBUG_LOG("set delete for r:%u", rb_ractor_id(RACTOR_PTR(b.sender)));

- }

-

- if (b.p.take.config) {

- b.p.take.config->closed = true;

- }

-

- // TODO: deadlock-able?

- RACTOR_LOCK(tr);

- {

- ractor_wakeup(tr, tr_th, wait_taking, wakeup_by_close);

- }

- RACTOR_UNLOCK(tr);

- }

- }

-

- // raising yielding Ractor

- ractor_wakeup(r, NULL, wait_yielding, wakeup_by_close);

-

- VM_ASSERT(ractor_queue_empty_p(r, ts));

- }

- RACTOR_UNLOCK(r);

- return prev;

-}

- r->sync.outgoing_port_closed = true;

- r->sync.incoming_port_closed = true;

- r->sync.will_basket.type.e = basket_type_none;

- ractor_queue_setup(&r->sync.recv_queue);

- ractor_queue_setup(&r->sync.takers_queue);

- rb_native_mutex_initialize(&r->sync.lock);

- rb_native_cond_initialize(&r->barrier_wait_cond);

-

-#ifdef RUBY_THREAD_WIN32_H

- rb_native_cond_initialize(&r->barrier_wait_cond);

-#endif

- ccan_list_head_init(&r->sync.wait.waiting_threads);

-ractor_yield_atexit(rb_execution_context_t *ec, rb_ractor_t *cr, VALUE v, bool exc)

- if (cr->sync.outgoing_port_closed) {

- return;

- }

-

- ASSERT_ractor_unlocking(cr);

-

- struct rb_ractor_queue *ts = &cr->sync.takers_queue;

- rb_thread_t *cur_th = rb_ec_thread_ptr(ec);

-

- retry:

- if (ractor_try_yield(ec, cr, ts, v, Qfalse, exc, true)) {

- // OK.

- }

- else {

- bool retry = false;

- RACTOR_LOCK(cr);

- {

- if (!ractor_check_take_basket(cr, ts)) {

- VM_ASSERT(cur_th->ractor_waiting.wait_status == wait_none);

- RUBY_DEBUG_LOG("leave a will");

- ractor_basket_fill_will(cr, cur_th, &cr->sync.will_basket, v, exc);

- }

- else {

- RUBY_DEBUG_LOG("rare timing!");

- retry = true; // another ractor is waiting for the yield.

- }

- }

- RACTOR_UNLOCK(cr);

-

- if (retry) goto retry;

- }

- ractor_yield_atexit(ec, cr, result, false);

- ractor_yield_atexit(ec, cr, ec->errinfo, true);

- ractor_close_incoming(ec, cr);

- ractor_close_outgoing(ec, cr);

- ptr[i] = ractor_receive(ec, r);

- ractor_send(ec, r, RARRAY_AREF(args, i), false);

-#ifndef USE_RACTOR_SELECTOR

-#define USE_RACTOR_SELECTOR 0

-#endif

-

-RUBY_SYMBOL_EXPORT_BEGIN

-void rb_init_ractor_selector(void);

-RUBY_SYMBOL_EXPORT_END

-

-/*

- * Document-class: Ractor::Selector

- * :nodoc: currently

- *

- * Selects multiple Ractors to be activated.

- */

-void

-rb_init_ractor_selector(void)

-{

- rb_cRactorSelector = rb_define_class_under(rb_cRactor, "Selector", rb_cObject);

- rb_undef_alloc_func(rb_cRactorSelector);

-

- rb_define_singleton_method(rb_cRactorSelector, "new", ractor_selector_new , -1);

- rb_define_method(rb_cRactorSelector, "add", ractor_selector_add, 1);

- rb_define_method(rb_cRactorSelector, "remove", ractor_selector_remove, 1);

- rb_define_method(rb_cRactorSelector, "clear", ractor_selector_clear, 0);

- rb_define_method(rb_cRactorSelector, "empty?", ractor_selector_empty_p, 0);

- rb_define_method(rb_cRactorSelector, "wait", ractor_selector_wait, -1);

- rb_define_method(rb_cRactorSelector, "_wait", ractor_selector__wait, 4);

-}

-

- // internal

-

-#if USE_RACTOR_SELECTOR

- rb_init_ractor_selector();

-#endif

-// Ractor::Channel (emulate with Ractor)

-

-typedef rb_ractor_t rb_ractor_channel_t;

-

-static VALUE

-ractor_channel_func(RB_BLOCK_CALL_FUNC_ARGLIST(y, c))

-{

- rb_execution_context_t *ec = GET_EC();

- rb_ractor_t *cr = rb_ec_ractor_ptr(ec);

-

- while (1) {

- int state;

-

- EC_PUSH_TAG(ec);

- if ((state = EC_EXEC_TAG()) == TAG_NONE) {

- VALUE obj = ractor_receive(ec, cr);

- ractor_yield(ec, cr, obj, Qfalse);

- }

- EC_POP_TAG();

-

- if (state) {

- // ignore the error

- break;

- }

- }

-

- return Qnil;

-}

-

-static VALUE

-rb_ractor_channel_new(void)

-{

-#if 0

- return rb_funcall(rb_const_get(rb_cRactor, rb_intern("Channel")), rb_intern("new"), 0);

-#else

- // class Channel

- // def self.new

- // Ractor.new do # func body

- // while true

- // obj = Ractor.receive

- // Ractor.yield obj

- // end

- // rescue Ractor::ClosedError

- // nil

- // end

- // end

- // end

-

- return ractor_create_func(rb_cRactor, Qnil, rb_str_new2("Ractor/channel"), rb_ary_new(), ractor_channel_func);

-#endif

-}

-

-static VALUE

-rb_ractor_channel_yield(rb_execution_context_t *ec, VALUE vch, VALUE obj)

-{

- VM_ASSERT(ec == rb_current_ec_noinline());

- rb_ractor_channel_t *ch = RACTOR_PTR(vch);

-

- ractor_send(ec, (rb_ractor_t *)ch, obj, Qfalse);

- return Qnil;

-}

-

-static VALUE

-rb_ractor_channel_take(rb_execution_context_t *ec, VALUE vch)

-{

- VM_ASSERT(ec == rb_current_ec_noinline());

- rb_ractor_channel_t *ch = RACTOR_PTR(vch);

-

- return ractor_take(ec, (rb_ractor_t *)ch);

-}

-

-static VALUE

-rb_ractor_channel_close(rb_execution_context_t *ec, VALUE vch)

-{

- VM_ASSERT(ec == rb_current_ec_noinline());

- rb_ractor_channel_t *ch = RACTOR_PTR(vch);

-

- ractor_close_incoming(ec, (rb_ractor_t *)ch);

- return ractor_close_outgoing(ec, (rb_ractor_t *)ch);

-}

-

- VALUE ch;

- rb_ractor_channel_yield(GET_EC(), crr->ch, Qtrue);

-

- .ch = rb_ractor_channel_new(),

- rb_ractor_channel_take(ec, crr.ch);

- rb_ractor_channel_close(ec, crr.ch);

- .ch = rb_ractor_channel_new(),

- rb_ractor_channel_take(ec, crr.ch);

- rb_ractor_channel_close(ec, crr.ch);

-# r.take # wait for it to finish

-# r.take

-# There are two pairs of methods for sending/receiving messages:

-#

-# * Ractor#send and Ractor.receive for when the _sender_ knows the receiver (push);

-# * Ractor.yield and Ractor#take for when the _receiver_ knows the sender (pull);

-#

-# as if received by Ractor.receive, and the last block value is sent outside of the

-# ractor as if sent by Ractor.yield.

-#

-# A little demonstration of a classic ping-pong:

-#

-# server = Ractor.new(name: "server") do

-# puts "Server starts: #{self.inspect}"

-# puts "Server sends: ping"

-# Ractor.yield 'ping' # The server doesn't know the receiver and sends to whoever interested

-# received = Ractor.receive # The server doesn't know the sender and receives from whoever sent

-# puts "Server received: #{received}"

-# end

-#

-# client = Ractor.new(server) do |srv| # The server is sent to the client, and available as srv

-# puts "Client starts: #{self.inspect}"

-# received = srv.take # The client takes a message from the server

-# puts "Client received from " \

-# "#{srv.inspect}: #{received}"

-# puts "Client sends to " \

-# "#{srv.inspect}: pong"

-# srv.send 'pong' # The client sends a message to the server

-# end

-#

-# [client, server].each(&:take) # Wait until they both finish

-#

-# This will output something like:

-#

-# Server starts: #<Ractor:#2 server test.rb:1 running>

-# Server sends: ping

-# Client starts: #<Ractor:#3 test.rb:8 running>

-# Client received from #<Ractor:#2 server test.rb:1 blocking>: ping

-# Client sends to #<Ractor:#2 server test.rb:1 blocking>: pong

-# Server received: pong

-#

-# Ractors receive their messages via the <em>incoming port</em>, and send them

-# to the <em>outgoing port</em>. Either one can be disabled with Ractor#close_incoming and

-# Ractor#close_outgoing, respectively. When a ractor terminates, its ports are closed

-# automatically.

- # Ractor.select(*ractors, [yield_value:, move: false]) -> [ractor or symbol, obj]

- #

- # Wait for any ractor to have something in its outgoing port, read from this ractor, and

- # then return that ractor and the object received.

- #

- # r1 = Ractor.new {Ractor.yield 'from 1'}

- # r2 = Ractor.new {Ractor.yield 'from 2'}

- #

- # r, obj = Ractor.select(r1, r2)

- #

- # puts "received #{obj.inspect} from #{r.inspect}"

- # # Prints: received "from 1" from #<Ractor:#2 test.rb:1 running>

- # # But could just as well print "from r2" here, either prints could be first.

- #

- # If one of the given ractors is the current ractor, and it is selected, +r+ will contain

- # the +:receive+ symbol instead of the ractor object.

- #

- # r1 = Ractor.new(Ractor.current) do |main|

- # main.send 'to main'

- # Ractor.yield 'from 1'

- # end

- # r2 = Ractor.new do

- # Ractor.yield 'from 2'

- # end

- #

- # r, obj = Ractor.select(r1, r2, Ractor.current)

- # puts "received #{obj.inspect} from #{r.inspect}"

- # # Could print: received "to main" from :receive

- #

- # If +yield_value+ is provided, that value may be yielded if another ractor is calling #take.

- # In this case, the pair <tt>[:yield, nil]</tt> is returned:

- #

- # r1 = Ractor.new(Ractor.current) do |main|

- # puts "Received from main: #{main.take}"

- # end

- #

- # puts "Trying to select"

- # r, obj = Ractor.select(r1, Ractor.current, yield_value: 123)

- # wait

- # puts "Received #{obj.inspect} from #{r.inspect}"

- #

- # This will print:

- #

- # Trying to select

- # Received from main: 123

- # Received nil from :yield

- #

- # +move+ boolean flag defines whether yielded value will be copied (default) or moved.

- def self.select(*ractors, yield_value: yield_unspecified = true, move: false)

- raise ArgumentError, 'specify at least one ractor or `yield_value`' if yield_unspecified && ractors.empty?

-

- if ractors.delete Ractor.current

- do_receive = true

- else

- do_receive = false

- __builtin_ractor_select_internal ractors, do_receive, !yield_unspecified, yield_value, move

- # Ractor.receive -> msg

- #

- # Receive a message from the incoming port of the current ractor (which was

- # sent there by #send from another ractor).

- #

- # r = Ractor.new do

- # v1 = Ractor.receive

- # puts "Received: #{v1}"

- # end

- # r.send('message1')

- # r.take

- # # Here will be printed: "Received: message1"

- #

- # Alternatively, the private instance method +receive+ may be used:

- #

- # r = Ractor.new do

- # v1 = receive

- # puts "Received: #{v1}"

- # end

- # r.send('message1')

- # r.take

- # # This prints: "Received: message1"

- #

- # The method blocks if the queue is empty.

- #

- # r = Ractor.new do

- # puts "Before first receive"

- # v1 = Ractor.receive

- # puts "Received: #{v1}"

- # v2 = Ractor.receive

- # puts "Received: #{v2}"

- # end

- # wait

- # puts "Still not received"

- # r.send('message1')

- # wait

- # puts "Still received only one"

- # r.send('message2')

- # r.take

- #

- # Output:

- #

- # Before first receive

- # Still not received

- # Received: message1

- # Still received only one

- # Received: message2

- #

- # If close_incoming was called on the ractor, the method raises Ractor::ClosedError

- # if there are no more messages in the incoming queue:

- #

- # Ractor.new do

- # close_incoming

- # receive

- # end

- # wait

- # # in `receive': The incoming port is already closed => #<Ractor:#2 test.rb:1 running> (Ractor::ClosedError)

- __builtin_cexpr! %q{

- ractor_receive(ec, rb_ec_ractor_ptr(ec))

- }

- __builtin_cexpr! %q{

- ractor_receive(ec, rb_ec_ractor_ptr(ec))

- }

- # Ractor.receive_if {|msg| block } -> msg

- #

- # Receive only a specific message.

- #

- # Instead of Ractor.receive, Ractor.receive_if can be given a pattern (or any

- # filter) in a block and you can choose the messages to accept that are available in

- # your ractor's incoming queue.

- #

- # r = Ractor.new do

- # p Ractor.receive_if{|msg| msg.match?(/foo/)} #=> "foo3"

- # p Ractor.receive_if{|msg| msg.match?(/bar/)} #=> "bar1"

- # p Ractor.receive_if{|msg| msg.match?(/baz/)} #=> "baz2"

- # end

- # r << "bar1"

- # r << "baz2"

- # r << "foo3"

- # r.take

- #

- # This will output:

- #

- # foo3

- # bar1

- # baz2

- #

- # If the block returns a truthy value, the message is removed from the incoming queue

- # and returned.

- # Otherwise, the message remains in the incoming queue and the next messages are checked

- # by the given block.

- #

- # If there are no messages left in the incoming queue, the method will

- # block until new messages arrive.

- #

- # If the block is escaped by break/return/exception/throw, the message is removed from

- # the incoming queue as if a truthy value had been returned.

- #

- # r = Ractor.new do

- # val = Ractor.receive_if{|msg| msg.is_a?(Array)}

- # puts "Received successfully: #{val}"

- # end

- #

- # r.send(1)

- # r.send('test')

- # wait

- # puts "2 non-matching sent, nothing received"

- # r.send([1, 2, 3])

- # wait

- #

- # Prints:

- #

- # 2 non-matching sent, nothing received

- # Received successfully: [1, 2, 3]

- #

- # Note that you can not call receive/receive_if in the given block recursively.

- # You should not do any tasks in the block other than message filtration.

- #

- # Ractor.current << true

- # Ractor.receive_if{|msg| Ractor.receive}

- # #=> `receive': can not call receive/receive_if recursively (Ractor::Error)

- #

- def self.receive_if &b

- Primitive.ractor_receive_if b

- end

-

- # same as Ractor.receive_if

- private def receive_if &b

- Primitive.ractor_receive_if b

- end

-

- #

- # call-seq:

- # ractor.send(msg, move: false) -> self

- #

- # Send a message to a Ractor's incoming queue to be accepted by Ractor.receive.

- #

- # r = Ractor.new do

- # value = Ractor.receive

- # puts "Received #{value}"

- # end

- # r.send 'message'

- # # Prints: "Received: message"

- #

- # The method is non-blocking (will return immediately even if the ractor is not ready

- # to receive anything):

- #

- # r = Ractor.new {sleep(5)}

- # r.send('test')

- # puts "Sent successfully"

- # # Prints: "Sent successfully" immediately

- #

- # An attempt to send to a ractor which already finished its execution will raise Ractor::ClosedError.

- #

- # r = Ractor.new {}

- # r.take

- # p r

- # # "#<Ractor:#6 (irb):23 terminated>"

- # r.send('test')

- # # Ractor::ClosedError (The incoming-port is already closed)

- #

- # If close_incoming was called on the ractor, the method also raises Ractor::ClosedError.

- #

- # r = Ractor.new do

- # sleep(500)

- # receive

- # end

- # r.close_incoming

- # r.send('test')

- # # Ractor::ClosedError (The incoming-port is already closed)

- # # The error is raised immediately, not when the ractor tries to receive

- #

- # If the +obj+ is unshareable, by default it will be copied into the receiving ractor by deep cloning.

- # If <tt>move: true</tt> is passed, the object is _moved_ into the receiving ractor and becomes

- # inaccessible to the sender.

- #

- # r = Ractor.new {puts "Received: #{receive}"}

- # msg = 'message'

- # r.send(msg, move: true)

- # r.take

- # p msg

- #

- # This prints:

- #

- # Received: message

- # in `p': undefined method `inspect' for #<Ractor::MovedObject:0x000055c99b9b69b8>

- #

- # All references to the object and its parts will become invalid to the sender.

- #

- # r = Ractor.new {puts "Received: #{receive}"}

- # s = 'message'

- # ary = [s]

- # copy = ary.dup

- # r.send(ary, move: true)

- #

- # s.inspect

- # # Ractor::MovedError (can not send any methods to a moved object)

- # ary.class

- # # Ractor::MovedError (can not send any methods to a moved object)

- # copy.class

- # # => Array, it is different object

- # copy[0].inspect

- # # Ractor::MovedError (can not send any methods to a moved object)

- # # ...but its item was still a reference to `s`, which was moved

- #

- # If the object is shareable, <tt>move: true</tt> has no effect on it:

- #

- # r = Ractor.new {puts "Received: #{receive}"}

- # s = 'message'.freeze

- # r.send(s, move: true)

- # s.inspect #=> "message", still available

- def send(obj, move: false)

- __builtin_cexpr! %q{

- ractor_send(ec, RACTOR_PTR(self), obj, move)

- }

- #

- # call-seq:

- # Ractor.yield(msg, move: false) -> nil

- #

- # Send a message to the current ractor's outgoing port to be accepted by #take.

- #

- # r = Ractor.new {Ractor.yield 'Hello from ractor'}

- # puts r.take

- # # Prints: "Hello from ractor"

- #

- # This method is blocking, and will return only when somebody consumes the

- # sent message.

- #

- # r = Ractor.new do

- # Ractor.yield 'Hello from ractor'

- # puts "Ractor: after yield"

- # end

- # wait

- # puts "Still not taken"

- # puts r.take

- #

- # This will print:

- #

- # Still not taken

- # Hello from ractor

- # Ractor: after yield

- #

- # If the outgoing port was closed with #close_outgoing, the method will raise:

- #

- # r = Ractor.new do

- # close_outgoing

- # Ractor.yield 'Hello from ractor'

- # end

- # wait

- # # `yield': The outgoing-port is already closed (Ractor::ClosedError)

- #

- # The meaning of the +move+ argument is the same as for #send.

- def self.yield(obj, move: false)

- __builtin_cexpr! %q{

- ractor_yield(ec, rb_ec_ractor_ptr(ec), obj, move)

- }

- end

-

- #

- # call-seq:

- # ractor.take -> msg

- #

- # Get a message from the ractor's outgoing port, which was put there by Ractor.yield or at ractor's

- # termination.

- #

- # r = Ractor.new do

- # Ractor.yield 'explicit yield'

- # 'last value'

- # end

- # puts r.take #=> 'explicit yield'

- # puts r.take #=> 'last value'

- # puts r.take # Ractor::ClosedError (The outgoing-port is already closed)

- #

- # The fact that the last value is also sent to the outgoing port means that +take+ can be used

- # as an analog of Thread#join ("just wait until ractor finishes"). However, it will raise if

- # somebody has already consumed that message.

- #

- # If the outgoing port was closed with #close_outgoing, the method will raise Ractor::ClosedError.

- #

- # r = Ractor.new do

- # sleep(500)

- # Ractor.yield 'Hello from ractor'

- # end

- # r.close_outgoing

- # r.take

- # # Ractor::ClosedError (The outgoing-port is already closed)

- # # The error would be raised immediately, not when ractor will try to receive

- #

- # If an uncaught exception is raised in the Ractor, it is propagated by take as a

- # Ractor::RemoteError.

- #

- # r = Ractor.new {raise "Something weird happened"}

- #

- # begin

- # r.take

- # rescue => e

- # p e # => #<Ractor::RemoteError: thrown by remote Ractor.>

- # p e.ractor == r # => true

- # p e.cause # => #<RuntimeError: Something weird happened>

- # end

- #

- # Ractor::ClosedError is a descendant of StopIteration, so the termination of the ractor will break

- # out of any loops that receive this message without propagating the error:

- #

- # r = Ractor.new do

- # 3.times {|i| Ractor.yield "message #{i}"}

- # "finishing"

- # end

- #

- # loop {puts "Received: " + r.take}

- # puts "Continue successfully"

- #

- # This will print:

- #

- # Received: message 0

- # Received: message 1

- # Received: message 2

- # Received: finishing

- # Continue successfully

- def take

- __builtin_cexpr! %q{

- ractor_take(ec, RACTOR_PTR(self))

- }

- end

-

- # ractor.close_incoming -> true | false

- # Closes the incoming port and returns whether it was already closed. All further attempts

- # to Ractor.receive in the ractor, and #send to the ractor will fail with Ractor::ClosedError.

- # r = Ractor.new {sleep(500)}

- # r.close_incoming #=> false

- # r.close_incoming #=> true

- # r.send('test')

- # # Ractor::ClosedError (The incoming-port is already closed)

- def close_incoming

- __builtin_cexpr! %q{

- ractor_close_incoming(ec, RACTOR_PTR(self));

- }

- end

-

- #

- # call-seq:

- # ractor.close_outgoing -> true | false

- #

- # Closes the outgoing port and returns whether it was already closed. All further attempts

- # to Ractor.yield in the ractor, and #take from the ractor will fail with Ractor::ClosedError.

- #

- # r = Ractor.new {sleep(500)}

- # r.close_outgoing #=> false

- # r.close_outgoing #=> true

- # r.take

- # # Ractor::ClosedError (The outgoing-port is already closed)

- def close_outgoing

- __builtin_cexpr! %q{

- ractor_close_outgoing(ec, RACTOR_PTR(self));

- }

-enum rb_ractor_basket_type {

- // basket is empty

- basket_type_none,

-

- // value is available

- basket_type_ref,

- basket_type_copy,

- basket_type_move,

- basket_type_will,

-

- // basket should be deleted

- basket_type_deleted,

-

- // basket is reserved

- basket_type_reserved,

-

- // take_basket is available

- basket_type_take_basket,

-

- // basket is keeping by yielding ractor

- basket_type_yielding,

-};

-

-// per ractor taking configuration

-struct rb_ractor_selector_take_config {

- bool closed;

- bool oneshot;

-};

-

-struct rb_ractor_basket {

- union {

- enum rb_ractor_basket_type e;

- rb_atomic_t atomic;

- } type;

- VALUE sender; // Ractor object sending message

- rb_thread_t *sending_th;

-

- union {

- struct {

- VALUE v;

- bool exception;

- } send;

-

- struct {

- struct rb_ractor_basket *basket;

- struct rb_ractor_selector_take_config *config;

- } take;

- } p; // payload

-};

-

-static inline bool

-basket_type_p(struct rb_ractor_basket *b, enum rb_ractor_basket_type type)

-{

- return b->type.e == type;

-}

-

-static inline bool

-basket_none_p(struct rb_ractor_basket *b)

-{

- return basket_type_p(b, basket_type_none);

-}

-

-struct rb_ractor_queue {

- struct rb_ractor_basket *baskets;

- int start;

- int cnt;

- int size;

- unsigned int serial;

- unsigned int reserved_cnt;

-};

-

-enum rb_ractor_wait_status {

- wait_none = 0x00,

- wait_receiving = 0x01,

- wait_taking = 0x02,

- wait_yielding = 0x04,

- wait_moving = 0x08,

-};

-

-enum rb_ractor_wakeup_status {

- wakeup_none,

- wakeup_by_send,

- wakeup_by_yield,

- wakeup_by_take,

- wakeup_by_close,

- wakeup_by_interrupt,

- wakeup_by_retry,

-};

-

- bool incoming_port_closed;

- bool outgoing_port_closed;

- // All sent messages will be pushed into recv_queue

- struct rb_ractor_queue recv_queue;

- // The following ractors waiting for the yielding by this ractor

- struct rb_ractor_queue takers_queue;

- // Enabled if the ractor already terminated and not taken yet.

- struct rb_ractor_basket will_basket;

- struct ractor_wait {

- struct ccan_list_head waiting_threads;

- // each thread has struct ccan_list_node ractor_waiting.waiting_node

- } wait;

-

- // vm wide barrier synchronization

- rb_nativethread_cond_t barrier_wait_cond;

-

- ractor.take

- }.take

- d2, d3 = Ractor.new(d) { |d| [d, Date.parse(d.to_s)] }.take

- }.take

- }.take

- }.take

- }.take

- }.take

- }.take

- puts r.take == hexdigest

- end.each(&:take)

- end.take

- ractor = Ractor.new do

- Ractor.yield :sync

- Ractor.yield s.name

- ractor.take # => :sync

- name = ractor.take # => first name

- ractor.take # => :done

- end.each(&:take)

- assert_equal(Fiber, r.take)

- puts r.take

- puts console.class == r.take.class

- puts r.take

- actual_json = r.take

- obj = 'a' * 1024

- Ractor.yield obj

- r.take

- r.take

- ractors.each(&:take)

- ractors.each(&:take)

- assert_equal(Pathname("a/b/c"), r.take)

- puts(ignore_warnings { Ractor.new(*arguments, &block) }.take)

- end.take

- end.take

- end.take

- Ractor.new{}.take

- Ractor.yield #{exception_var}.class

- error_class = #{ractor_var}.take

- r = Ractor.new do

- Ractor.yield ENV['test']

- Ractor.yield ENV['TEST']

- Ractor.yield ENV['test']

- Ractor.yield ENV['TEST']

- Ractor.yield ENV['TEST']

- Ractor.yield ENV['test']

- assert_nil(r.take)

- assert_nil(r.take)

- assert_equal('foo', r.take)

- assert_equal('foo', r.take)

- assert_nil(r.take)

- assert_equal('bar', r.take)

- assert_equal('bar', r.take)

- assert_equal('foo', r.take)

- #{str_for_assert_raise_on_yielded_exception_class(TypeError, "r")}

- r = Ractor.new do

- #{str_for_assert_raise_on_yielded_exception_class(TypeError, "r")}

- r = Ractor.new do

- Ractor.yield(ENV.has_value?(val))

- Ractor.yield(ENV.has_value?(val.upcase))

- Ractor.yield(ENV.has_value?(val))

- Ractor.yield(ENV.has_value?(val.upcase))

- Ractor.yield ENV.has_value?(val)

- Ractor.yield ENV.has_value?(val.upcase)

- end

- assert_equal(false, r.take)

- assert_equal(false, r.take)

- assert_equal(true, r.take)

- assert_equal(false, r.take)

- assert_equal(false, r.take)

- assert_equal(true, r.take)

- r = Ractor.new do

- Ractor.yield ENV.key(val)

- Ractor.yield ENV.key(val.upcase)

- Ractor.yield ENV.key(val)

- Ractor.yield ENV.key(val.upcase)

- Ractor.yield ENV.key(val)

- Ractor.yield ENV.key(val.upcase)

- assert_nil(r.take)

- assert_nil(r.take)

- assert_equal('TEST', r.take.upcase)

- assert_equal('test', r.take)

- assert_nil(r.take)

- assert_nil(r.take)

- assert_equal('TEST', r.take.upcase)

- assert_equal('test', r.take)

- r = Ractor.new do

- Ractor.yield ENV.delete("TEST")

- Ractor.yield(ENV.delete("TEST"){|name| "NO "+name})

- #{str_to_receive_invalid_envvar_errors("r")}

- assert_nil(r.take)

- exception_class = r.take

- assert_equal("NO TEST", r.take)

- r = Ractor.new do

- Ractor.yield ENV["#{PATH_ENV}"]

- Ractor.yield ENV[""]

- #{str_to_receive_invalid_envvar_errors("r")}

- assert_equal("", r.take)

- assert_nil(r.take)

- r = Ractor.new do

- Ractor.yield ENV.fetch("test")

- Ractor.yield ex.receiver.object_id

- Ractor.yield ex.key

- Ractor.yield ENV.fetch("test", "foo")

- Ractor.yield(ENV.fetch("test"){"bar"})

- Ractor.yield ENV.fetch('#{PATH_ENV}')

- end

- assert_equal("foo", r.take)

- #{str_for_assert_raise_on_yielded_exception_class(KeyError, "r")}

- assert_equal(ENV.object_id, r.take)

- assert_equal("test", r.take)

- assert_equal("foo", r.take)

- assert_equal("bar", r.take)

- #{str_to_receive_invalid_envvar_errors("r")}

- exception_class = r.take

- assert_equal("", r.take)

- r = Ractor.new do

- Ractor.yield ENV["test"]

- exception_class = r.take

- assert_nil(r.take)

- #{str_to_receive_invalid_envvar_errors("r")}

- #{str_to_receive_invalid_envvar_errors("r")}

- r = Ractor.new do

- Ractor.yield a

- a = r.take

- r = Ractor.new do

- ENV.each_key {|k| Ractor.yield(k)}

- Ractor.yield "finished"

- while((x=r.take) != "finished")

- r = Ractor.new do

- Ractor.yield a

- a = r.take

- r = Ractor.new do

- ENV.each_value {|k| Ractor.yield(k)}

- Ractor.yield "finished"

- while((x=r.take) != "finished")

- r = Ractor.new do

- ENV.each_pair {|k, v| Ractor.yield([k,v])}

- Ractor.yield "finished"

- while((k,v=r.take) != "finished")

- r = Ractor.new do

- Ractor.yield [h1, h2]

- Ractor.yield(ENV.reject! {|k, v| #{ignore_case_str} ? k.upcase == "TEST" : k == "test" })

- h1, h2 = r.take

- assert_nil(r.take)

- r = Ractor.new do

- Ractor.yield [h1, h2]

- Ractor.yield (ENV.delete_if {|k, v| #{ignore_case_str} ? k.upcase == "TEST" : k == "test" })

- h1, h2 = r.take

- assert_same(ENV, r.take)

- r = Ractor.new do

- Ractor.yield [h1, h2]

- Ractor.yield(ENV.select! {|k, v| #{ignore_case_str} ? k.upcase != "TEST" : k != "test" })

- h1, h2 = r.take

- assert_nil(r.take)

- r = Ractor.new do

- Ractor.yield [h1, h2]

- Ractor.yield(ENV.filter! {|k, v| #{ignore_case_str} ? k.upcase != "TEST" : k != "test" })

- h1, h2 = r.take

- assert_nil(r.take)

- r = Ractor.new do

- Ractor.yield [h1, h2]

- Ractor.yield (ENV.keep_if {|k, v| #{ignore_case_str} ? k.upcase != "TEST" : k != "test" })

- h1, h2 = r.take

- assert_equal(ENV, r.take)

- r = Ractor.new do

- Ractor.yield ENV.values_at("test", "test")

- assert_equal(["foo", "foo"], r.take)

- r = Ractor.new do

- Ractor.yield h.size

- Ractor.yield [k, v]

- assert_equal(1, r.take)

- k, v = r.take

- r = Ractor.new do

- Ractor.yield(h.size)

- Ractor.yield [k, v]

- assert_equal(1, r.take)

- k, v = r.take

- r = Ractor.new do

- Ractor.yield(ENV.slice())

- Ractor.yield(ENV.slice(""))

- Ractor.yield(ENV.slice("unknown"))

- Ractor.yield(ENV.slice("foo", "baz"))

- end

- assert_equal({}, r.take)

- assert_equal({}, r.take)

- assert_equal({}, r.take)

- assert_equal({"foo"=>"bar", "baz"=>"qux"}, r.take)

- r = Ractor.new do

- Ractor.yield ENV.except()

- Ractor.yield ENV.except("")

- Ractor.yield ENV.except("unknown")

- Ractor.yield ENV.except("foo", "baz")

- end

- assert_equal({"bar"=>"rab", "baz"=>"qux", "foo"=>"bar"}, r.take)

- assert_equal({"bar"=>"rab", "baz"=>"qux", "foo"=>"bar"}, r.take)

- assert_equal({"bar"=>"rab", "baz"=>"qux", "foo"=>"bar"}, r.take)

- assert_equal({"bar"=>"rab"}, r.take)

- r = Ractor.new do

- Ractor.yield ENV.size

- assert_equal(0, r.take)

- assert_equal("ENV", r.take)

- r = Ractor.new do

- Ractor.yield s

- s = r.take

- r = Ractor.new do

- Ractor.yield a

- a = r.take

- assert_nil(r.take)

- r = Ractor.new do

- Ractor.yield [s, ENV.size]

- s, s2 = r.take

- r = Ractor.new do

- Ractor.yield ENV.empty?

- Ractor.yield ENV.empty?

- assert r.take

- assert !r.take

- r = Ractor.new do

- Ractor.yield ENV.has_key?("test")

- Ractor.yield ENV.has_key?("test")

- assert !r.take

- assert r.take

- #{str_to_receive_invalid_envvar_errors("r")}

- r = Ractor.new do

- Ractor.yield ENV.assoc("test")

- Ractor.yield ENV.assoc("test")

- assert_nil(r.take)

- k, v = r.take

- #{str_to_receive_invalid_envvar_errors("r")}

- r = Ractor.new do

- Ractor.yield ENV.has_value?("foo")

- Ractor.yield ENV.has_value?("foo")

- assert !r.take

- assert r.take

- r = Ractor.new do

- Ractor.yield ENV.rassoc("foo")

- Ractor.yield ENV.rassoc("foo")

- assert_nil(r.take)

- k, v = r.take

- r = Ractor.new do

- Ractor.yield [h, ENV.to_hash]

- h, h2 = r.take

- r = Ractor.new do

- Ractor.yield [ENV.to_hash, ENV.to_h]

- Ractor.yield [ENV.map {|k, v| ["$\#{k}", v.size]}.to_h, ENV.to_h {|k, v| ["$\#{k}", v.size]}]

- a, b = r.take

- c, d = r.take

- r = Ractor.new do

- Ractor.yield [h1, h2]

- h1, h2 = r.take

- r = Ractor.new do

- Ractor.yield [a,b]

- Ractor.yield ENV.shift

- a,b = r.take

- assert_nil(r.take)

- r = Ractor.new do

- Ractor.yield(ENV.invert)

- check(r.take.to_a, [%w(bar foo), %w(qux baz)])

- r = Ractor.new do

- Ractor.yield ENV.to_hash

- Ractor.yield ENV.to_hash

- check(r.take.to_a, [%w(foo bar), %w(baz qux)])

- check(r.take.to_a, [%w(Foo Bar), %w(Baz Qux)])

- r = Ractor.new do

- Ractor.yield ENV.to_hash

- Ractor.yield ENV.to_hash

- check(r.take.to_a, [%w(foo bar), %w(baz quux), %w(a b)])

- check(r.take.to_a, [%w(foo bar), %w(baz baz_qux_quux), %w(a b)])

- r = Ractor.new huge_value do |v|

- Ractor.yield ENV["foo"]

- #{str_for_assert_raise_on_yielded_exception_class(Errno::EINVAL, "r")}

- result = r.take

- exception_class = r.take

- result = r.take

- r = Ractor.new do

- #{str_for_assert_raise_on_yielded_exception_class(TypeError, "r")}

- r = Ractor.new do

- Ractor.yield [k.frozen?]

- Ractor.yield [v.frozen?]

- Ractor.yield [k.frozen?]

- Ractor.yield [v.frozen?]

- Ractor.yield [ENV[k].frozen?, "[\#{k.dump}]"]

- Ractor.yield [ENV.fetch(k).frozen?, "fetch(\#{k.dump})"]

- Ractor.yield "finished"

- while((params=r.take) != "finished")

- r = Ractor.new do

- Ractor.yield [n, e0, e1, bug12475]

- n, e0, e1, bug12475 = r.take

- assert_equal Ractor::IsolationError, r_get.take.class

- assert_equal Ractor::IsolationError, r_get.take.class

- assert_equal Ractor::IsolationError, r_set.take.class

- assert_equal :new, r.take

- p r.take

- ractor_id = Ractor.new { Thread.current.group.object_id }.take

- end.take

- assert_equal 3, Ractor.new(tc) { |x| Ractor.yield(x.very_unique) }.take

- assert_equal tc.instance_variables.sort, Ractor.new(tc) { |x| Ractor.yield(x.instance_variables) }.take.sort

- Ractor.yield(o)

- o = r.take

- Ractor.yield(o)

- o = r.take

- puts r.take

- puts r.take.compact

- sizeof_int, fixnum_max = r.take

- actual = Ractor.new { Time.rfc2822("Fri, 21 Nov 1997 09:55:06 -0600") }.take

- r = Ractor.new do

- Ractor.yield d

- dir = r.take

- r.take

- assert_equal(URI.parse("https://ruby-lang.org/").inspect, r.take)

-#ifndef RUBY_THREAD_PTHREAD_H

- rb_native_cond_destroy(&th->ractor_waiting.cond);

-#endif

-thread_sched_lock_(struct rb_thread_sched *sched, rb_thread_t *th, const char *file, int line)

- rb_native_mutex_lock(&sched->lock_);

-

-#if VM_CHECK_MODE

- RUBY_DEBUG_LOG2(file, line, "th:%u prev_owner:%u", rb_th_serial(th), rb_th_serial(sched->lock_owner));

-#else

- RUBY_DEBUG_LOG2(file, line, "th:%u", rb_th_serial(th));

-thread_sched_unlock_(struct rb_thread_sched *sched, rb_thread_t *th, const char *file, int line)

- RUBY_DEBUG_LOG2(file, line, "th:%u", rb_th_serial(th));

-

-#if VM_CHECK_MODE

-

- rb_native_mutex_unlock(&sched->lock_);

-thread_sched_set_lock_owner(struct rb_thread_sched *sched, rb_thread_t *th)

- RUBY_DEBUG_LOG("th:%u", rb_th_serial(th));

-#if VM_CHECK_MODE > 0

- sched->lock_owner = th;

-static void ractor_sched_barrier_join_wait_locked(rb_vm_t *vm, rb_thread_t *th);

- while (UNLIKELY(vm->ractor.sched.barrier_waiting)) {

- RUBY_DEBUG_LOG("barrier-wait");

-

- ractor_sched_barrier_join_signal_locked(vm);

- ractor_sched_barrier_join_wait_locked(vm, add_th);

- VM_ASSERT(!vm->ractor.sched.barrier_waiting);

- if (add_th && !del_th && UNLIKELY(vm->ractor.sync.lock_owner != NULL)) {

- // it can be after barrier synchronization by another ractor

- rb_thread_t *lock_owner = NULL;

-#if VM_CHECK_MODE

- lock_owner = sched->lock_owner;

-#endif

- thread_sched_unlock(sched, lock_owner);

- {

- RB_VM_LOCKING();

- }

- thread_sched_lock(sched, lock_owner);

- }

-

- VM_ASSERT(!ractor_sched_timeslice_threads_contain_p(ready_th->vm, sched->running));

- thread_sched_set_lock_owner(sched, NULL);

- thread_sched_set_lock_owner(sched, th);

- thread_sched_set_lock_owner(sched, NULL);

- thread_sched_set_lock_owner(sched, th);

- thread_sched_set_lock_owner(sched, NULL);

- thread_sched_set_lock_owner(sched, th);

-static void

-setup_ubf(rb_thread_t *th, rb_unblock_function_t *func, void *arg)

- setup_ubf(th, ubf_waiting, (void *)th);

- setup_ubf(th, NULL, NULL);

-rb_ractor_sched_sleep(rb_execution_context_t *ec, rb_ractor_t *cr, rb_unblock_function_t *ubf_schedule_ractor_th)

- // r is sleeping status

- struct ccan_list_node *waitn = &th->ractor_waiting.waiting_node;

- VM_ASSERT(waitn->next == waitn->prev && waitn->next == waitn); // it should be unlinked

- ccan_list_add(&cr->sync.wait.waiting_threads, waitn);

- setup_ubf(th, ubf_schedule_ractor_th, (void *)ec);

- if (RUBY_VM_INTERRUPTED(th->ec)) {

- RUBY_DEBUG_LOG("interrupted");

- }

- else if (th->ractor_waiting.wakeup_status != wakeup_none) {

- RUBY_DEBUG_LOG("awaken:%d", (int)th->ractor_waiting.wakeup_status);

- }

- else {

- // sleep

- RB_VM_SAVE_MACHINE_CONTEXT(th);

- th->status = THREAD_STOPPED_FOREVER;

-

- RB_INTERNAL_THREAD_HOOK(RUBY_INTERNAL_THREAD_EVENT_SUSPENDED, th);

-

- bool can_direct_transfer = !th_has_dedicated_nt(th);

- thread_sched_wakeup_next_thread(sched, th, can_direct_transfer);

- thread_sched_wait_running_turn(sched, th, can_direct_transfer);

- th->status = THREAD_RUNNABLE;

- // wakeup

- }

- setup_ubf(th, NULL, NULL);

- rb_ractor_lock_self(cr);

- ccan_list_del_init(waitn);

-rb_ractor_sched_wakeup(rb_ractor_t *r, rb_thread_t *th)

- // ractor lock of r is acquired

- struct rb_thread_sched *sched = TH_SCHED(th);

- VM_ASSERT(th->ractor_waiting.wakeup_status != 0);

- thread_sched_lock(sched, th);

- if (th->status == THREAD_STOPPED_FOREVER) {

- thread_sched_to_ready_common(sched, th, true, false);

- thread_sched_unlock(sched, th);

- {

- // interrupts all running threads

- rb_thread_t *ith;

- ccan_list_for_each(&vm->ractor.sched.running_threads, ith, sched.node.running_threads) {

- if (ith->ractor != cr) {

- RUBY_DEBUG_LOG("barrier int:%u", rb_th_serial(ith));

- RUBY_VM_SET_VM_BARRIER_INTERRUPT(ith->ec);

- }

- }

- // wait for other ractors

- while (!ractor_sched_barrier_completed_p(vm)) {

- ractor_sched_set_unlocked(vm, cr);

- rb_native_cond_wait(&vm->ractor.sched.barrier_complete_cond, &vm->ractor.sched.lock);

- ractor_sched_set_locked(vm, cr);

- }

- ractor_sched_unlock(vm, cr);

- // acquire VM lock

- rb_native_mutex_lock(&vm->ractor.sync.lock);

- vm->ractor.sync.lock_rec = lock_rec;

- vm->ractor.sync.lock_owner = cr;

- RUBY_DEBUG_LOG("completed seirial:%u", vm->ractor.sched.barrier_serial);

- ractor_sched_lock(vm, cr);

- {

- vm->ractor.sched.barrier_waiting = false;

- setup_ubf(th, ubf_event_waiting, (void *)th);

- setup_ubf(th, NULL, NULL); // TODO: maybe it is already NULL?

- thread_sched_set_lock_owner(sched, th);

- rb_vm_t *vm = th->vm;

- bool has_ready_ractor = vm->ractor.sched.grq_cnt > 0; // at least this ractor is not queued

- if (!has_ready_ractor && next_th && !next_th->nt) {

- thread_sched_set_lock_owner(sched, NULL);

- rb_vm_cond_wait(vm, &vm->ractor.sync.barrier_cond);

- ccan_list_for_each(&vm->ractor.set, r, vmlr_node) {

- rb_native_cond_signal(&r->barrier_wait_cond);

- }

- rb_native_cond_signal(&vm->ractor.sync.barrier_cond);

- vm->ractor.sync.lock_owner = NULL;

- rb_native_cond_wait(&cr->barrier_wait_cond, &vm->ractor.sync.lock);

- VM_ASSERT(vm->ractor.sync.lock_owner == NULL);

- vm->ractor.sync.lock_owner = cr;

- rb_gc_mark(th->ractor_waiting.receiving_mutex);

- ccan_list_node_init(&th->ractor_waiting.waiting_node);

-#ifndef RUBY_THREAD_PTHREAD_H

- rb_native_cond_initialize(&th->ractor_waiting.cond);

-#endif

- rb_native_cond_initialize(&vm->ractor.sync.barrier_cond);

- rb_nativethread_cond_t barrier_cond;

-struct rb_thread_ractor_waiting {

- //enum rb_ractor_wait_status wait_status;

- int wait_status;

- //enum rb_ractor_wakeup_status wakeup_status;

- int wakeup_status;

- struct ccan_list_node waiting_node; // the rb_thread_t

- VALUE receiving_mutex; // protects Ractor.receive_if

-#ifndef RUBY_THREAD_PTHREAD_H

- rb_nativethread_cond_t cond;

-#endif

-};

-

- struct rb_thread_ractor_waiting ractor_waiting;

-

-vm_lock_leave(rb_vm_t *vm, unsigned int *lev APPEND_LOCATION_ARGS)

- (unsigned int)rb_ractor_id(vm->ractor.sync.lock_owner),

- vm_lock_leave(GET_VM(), lev APPEND_LOCATION_PARAMS);

- vm_lock_leave(vm, &vm->ractor.sync.lock_rec APPEND_LOCATION_PARAMS);

-#define RB_VM_LOCK_LEAVE_NO_BARRIER() RB_VM_LOCK_LEAVE_LEV(&_lev); }

- vm_locking_do; RB_VM_LOCK_LEAVE_LEV(&vm_locking_level), vm_locking_do = 0)