1berry/ruby
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix timing bug

Browse Source 
passing will and closing notification can conflict and
`Ractor::Selector#empty?` can return wrong answer.
This patch fix it.

```
     s = Ractor::Selector.new
     s.add Ractor.new{10}
     s.add Ractor.new{20}
     r, v = s.wait
     vs = []
     vs << v
     r, v = s.wait
     vs << v
     [*vs.sort, s.empty?]
  #=> "[10, 20, false]" (expected "[10, 20, true]")
```
This commit is contained in:
Koichi Sasada 2023-03-03 03:24:59 +09:00
parent ddd3c92bf6
commit 941d36d195
Notes: git 2023-03-03 01:19:56 +00:00 
Merged: https://github.com/ruby/ruby/pull/7420
1 changed files with 16 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
ractor.c
View File

@ -518,6 +518,7 @@ basket_type_name(enum rb_ractor_basket_type type)
case basket_type_deleted: return "deleted"; case basket_type_deleted: return "deleted";
case basket_type_reserved: return "reserved"; case basket_type_reserved: return "reserved";
case basket_type_take_basket: return "take_basket"; case basket_type_take_basket: return "take_basket";
case basket_type_yielding: return "yielding";
} }
VM_ASSERT(0); VM_ASSERT(0);
return NULL; return NULL;
@ -1532,7 +1533,7 @@ ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t dat)
int ret; int ret;
if (!basket_none_p(tb)) { if (!basket_none_p(tb)) {
RUBY_DEBUG_LOG("already taken:%s", basket_type_name(tb->type)); RUBY_DEBUG_LOG("already taken:%s", basket_type_name(tb->type.e));
return ST_STOP; return ST_STOP;
} }
@ -1546,7 +1547,7 @@ ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t dat)
ret = ST_STOP; ret = ST_STOP;
} }
else { else {
RUBY_DEBUG_LOG("has will, but already taken"); RUBY_DEBUG_LOG("has will, but already taken (%s)", basket_type_name(tb->type.e));
ret = ST_CONTINUE; ret = ST_CONTINUE;
} }
} }
@ -1558,7 +1559,7 @@ ractor_selector_wait_i(st_data_t key, st_data_t val, st_data_t dat)
ret = ST_STOP; ret = ST_STOP;
} }
else { else {
RUBY_DEBUG_LOG("closed, but already taken"); RUBY_DEBUG_LOG("closed, but already taken (%s)", basket_type_name(tb->type.e));
ret = ST_CONTINUE; ret = ST_CONTINUE;
} }
} }
@ -1644,8 +1645,8 @@ ractor_selector_wait(rb_execution_context_t *ec, VALUE selv, VALUE do_receivev,
retry_waiting: retry_waiting:
while (1) { while (1) {
if (!basket_none_p(tb)) { if (!basket_none_p(tb)) {
RUBY_DEBUG_LOG("taken:%s from r:%u", basket_type_name(tb->type), RUBY_DEBUG_LOG("taken:%s from r:%u", basket_type_name(tb->type.e),
rb_ractor_id(RACTOR_PTR(tb->sender))); tb->sender ? rb_ractor_id(RACTOR_PTR(tb->sender)) : 0);
break; break;
} }
if (do_receive && !ractor_queue_empty_p(cr, rq)) { if (do_receive && !ractor_queue_empty_p(cr, rq)) {
@ -1687,11 +1688,16 @@ ractor_selector_wait(rb_execution_context_t *ec, VALUE selv, VALUE do_receivev,
case basket_type_yielding: case basket_type_yielding:
rb_bug("unreachable"); rb_bug("unreachable");
case basket_type_deleted: { case basket_type_deleted: {
ractor_selector_remove(ec, selv, taken_basket.sender);
rb_ractor_t *r = RACTOR_PTR(taken_basket.sender); rb_ractor_t *r = RACTOR_PTR(taken_basket.sender);
if (!ractor_take_will_lock(r, &taken_basket)) { 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"); // rb_raise(rb_eRactorClosedError, "The outgoing-port is already closed");
// remove and retry wait // remove and retry wait
ractor_selector_remove(ec, selv, taken_basket.sender);
goto retry; goto retry;
} }
break; break;
@ -1704,6 +1710,8 @@ ractor_selector_wait(rb_execution_context_t *ec, VALUE selv, VALUE do_receivev,
break; break;
} }
RUBY_DEBUG_LOG("taken_basket:%s", basket_type_name(taken_basket.type.e));
ret_v = ractor_basket_accept(&taken_basket); ret_v = ractor_basket_accept(&taken_basket);
ret_r = taken_basket.sender; ret_r = taken_basket.sender;
success: success:
@ -1768,6 +1776,7 @@ ractor_close_outgoing(rb_execution_context_t *ec, rb_ractor_t *r)
if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_yielding, basket_type_deleted) != basket_type_yielding) { if (RUBY_ATOMIC_CAS(tb->type.atomic, basket_type_yielding, basket_type_deleted) != basket_type_yielding) {
rb_bug("unreachable"); rb_bug("unreachable");
} }
RUBY_DEBUG_LOG("set delete for r:%u", rb_ractor_id(RACTOR_PTR(b.sender)));
} }
if (b.p.take.config) { if (b.p.take.config) {