Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions bootstraptest/test_ractor.rb
Original file line number Diff line number Diff line change
Expand Up @@ -2848,3 +2848,18 @@ def st.m; :strct end
[o, s, st].each { |x| r.send(x, move: true) }
r.value.inspect
}

# A port that is still reachable keeps its queue, whether or not it was closed. (Dropping
# the queue of a port that is gone is not fixed for mmtk, so that case is a test-all one
# that omits itself: see test_port_queue_dropped_when_port_unreachable.)
assert_equal '[[0, 1, 2], [:a, :b]]', %q{
taken = 3.times.map do |i|
port = Ractor::Port.new
Ractor.new(port, i) { |p, n| p << n; nil }.join
port
end
closed = Ractor::Port.new
closed.send(:a); closed.send(:b); closed.close
6.times { GC.start }
[taken.map(&:receive), [closed.receive, closed.receive]]
}
15 changes: 5 additions & 10 deletions compile.c
Original file line number Diff line number Diff line change
Expand Up @@ -1506,7 +1506,7 @@ new_child_iseq(rb_iseq_t *iseq, const NODE *const node,

// The child AST wrapper does not carry the source hash, so copy it from
// the enclosing iseq before compiling, for grandchildren to inherit it.
if (ISEQ_BODY(iseq)->has_source_hash) {
if (ISEQ_BODY(iseq)->source_hash) {
rb_ast_t *child_ast = rb_ruby_ast_data_get(ast_value);
child_ast->body.source_hash = ISEQ_BODY(iseq)->source_hash;
child_ast->body.has_source_hash = 1;
Expand Down Expand Up @@ -12481,7 +12481,6 @@ rb_iseq_build_from_ary(rb_iseq_t *iseq, VALUE misc, VALUE locals, VALUE params,
VALUE source_hash = rb_hash_aref(misc, ID2SYM(rb_intern("source_hash")));
if (!NIL_P(source_hash)) {
ISEQ_BODY(iseq)->source_hash = NUM2ULL(source_hash);
ISEQ_BODY(iseq)->has_source_hash = true;
}

VALUE node_ids = Qfalse;
Expand Down Expand Up @@ -13790,12 +13789,10 @@ ibf_dump_iseq_each(struct ibf_dump *dump, const rb_iseq_t *iseq)
ibf_dump_write_small_value(dump, location_label_index);
ibf_dump_write_small_value(dump, body->location.first_lineno);
ibf_dump_write_small_value(dump, body->location.node_id);
/* Dump the source hash in two 32-bit halves, because VALUE may be
* 32 bits wide. */
uint64_t source_hash = body->has_source_hash ? body->source_hash : 0;
ibf_dump_write_small_value(dump, (VALUE)(uint32_t)(source_hash >> 32));
ibf_dump_write_small_value(dump, (VALUE)(uint32_t)source_hash);
ibf_dump_write_small_value(dump, body->has_source_hash ? 1 : 0);
/* Dump the source hash (0 if unavailable) in two 32-bit halves, because
* VALUE may be 32 bits wide. */
ibf_dump_write_small_value(dump, (VALUE)(uint32_t)(body->source_hash >> 32));
ibf_dump_write_small_value(dump, (VALUE)(uint32_t)body->source_hash);
ibf_dump_write_small_value(dump, body->location.code_location.beg_pos.lineno);
ibf_dump_write_small_value(dump, body->location.code_location.beg_pos.column);
ibf_dump_write_small_value(dump, body->location.code_location.end_pos.lineno);
Expand Down Expand Up @@ -13911,7 +13908,6 @@ ibf_load_iseq_each(struct ibf_load *load, rb_iseq_t *iseq, ibf_offset_t offset)
const uint64_t source_hash_hi = (uint64_t)ibf_load_small_value(load, &reading_pos);
const uint64_t source_hash_lo = (uint64_t)ibf_load_small_value(load, &reading_pos);
const uint64_t source_hash = (source_hash_hi << 32) | (uint32_t)source_hash_lo;
const bool has_source_hash = ibf_load_small_value(load, &reading_pos) != 0;
const int location_code_location_beg_pos_lineno = (int)ibf_load_small_value(load, &reading_pos);
const int location_code_location_beg_pos_column = (int)ibf_load_small_value(load, &reading_pos);
const int location_code_location_end_pos_lineno = (int)ibf_load_small_value(load, &reading_pos);
Expand Down Expand Up @@ -14013,7 +14009,6 @@ ibf_load_iseq_each(struct ibf_load *load, rb_iseq_t *iseq, ibf_offset_t offset)
load_body->location.first_lineno = location_first_lineno;
load_body->location.node_id = location_node_id;
load_body->source_hash = source_hash;
load_body->has_source_hash = has_source_hash;
load_body->location.code_location.beg_pos.lineno = location_code_location_beg_pos_lineno;
load_body->location.code_location.beg_pos.column = location_code_location_beg_pos_column;
load_body->location.code_location.end_pos.lineno = location_code_location_end_pos_lineno;
Expand Down
4 changes: 2 additions & 2 deletions gc/default/default.c
Original file line number Diff line number Diff line change
Expand Up @@ -7113,7 +7113,7 @@ gc_marks_finish(rb_objspace_t *objspace)
}

// TODO: refactor so we don't need to call this
rb_ractor_finish_marking();
rb_ractor_finish_marking(is_full_marking(objspace));

gc_event_hook(objspace, RUBY_INTERNAL_EVENT_GC_END_MARK);
}
Expand Down Expand Up @@ -9065,7 +9065,7 @@ gc_start_global(rb_objspace_t *driver, unsigned int reason, bool compact, bool a
/* This cycle's root pass over every Ractor has swept the deleted ractor-local keys out of
* each storage. Free the key structs while still inside the barrier (a local GC never
* can; see rb_ractor_finish_marking). */
rb_ractor_finish_marking();
rb_ractor_finish_marking(true);

gc_marking_exit(driver);

Expand Down
2 changes: 1 addition & 1 deletion gc/gc.h
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ MODULAR_GC_FN void rb_gc_print_backtrace();
RUBY_SYMBOL_EXPORT_END
#endif

void rb_ractor_finish_marking(void);
void rb_ractor_finish_marking(bool full_mark);

// -------------------Private section begin------------------------
// Functions in this section are private to the default GC and gc.c
Expand Down
8 changes: 4 additions & 4 deletions iseq.c
Original file line number Diff line number Diff line change
Expand Up @@ -1140,7 +1140,7 @@ rb_iseq_new_with_opt(VALUE ast_value, VALUE name, VALUE path, VALUE realpath,

if (body && body->has_source_hash) {
ISEQ_BODY(iseq)->source_hash = body->source_hash;
ISEQ_BODY(iseq)->has_source_hash = true;
RUBY_ASSERT(ISEQ_BODY(iseq)->source_hash != 0);
}

rb_iseq_compile_node(iseq, node);
Expand All @@ -1164,7 +1164,7 @@ pm_iseq_build(pm_scope_node_t *node, VALUE name, VALUE path, VALUE realpath,
ISEQ_BODY(iseq)->prism = true;

ISEQ_BODY(iseq)->source_hash = node->source_hash;
ISEQ_BODY(iseq)->has_source_hash = true;
RUBY_ASSERT(ISEQ_BODY(iseq)->source_hash != 0);

rb_compile_option_t next_option;
if (!option) option = &COMPILE_OPTION_DEFAULT;
Expand Down Expand Up @@ -3764,7 +3764,7 @@ iseq_data_to_ary(const rb_iseq_t *iseq)
rb_hash_aset(misc, ID2SYM(rb_intern("local_size")), INT2FIX(iseq_body->local_table_size));
rb_hash_aset(misc, ID2SYM(rb_intern("stack_max")), INT2FIX(iseq_body->stack_max));
rb_hash_aset(misc, ID2SYM(rb_intern("node_id")), INT2FIX(iseq_body->location.node_id));
rb_hash_aset(misc, ID2SYM(rb_intern("source_hash")), iseq_body->has_source_hash ? ULL2NUM(iseq_body->source_hash) : Qnil);
rb_hash_aset(misc, ID2SYM(rb_intern("source_hash")), iseq_body->source_hash ? ULL2NUM(iseq_body->source_hash) : Qnil);
rb_hash_aset(misc, ID2SYM(rb_intern("code_location")),
rb_ary_new_from_args(4,
INT2FIX(iseq_body->location.code_location.beg_pos.lineno),
Expand Down Expand Up @@ -4607,7 +4607,7 @@ static VALUE
iseqw_source_hash(VALUE self)
{
const rb_iseq_t *iseq = iseqw_check(self);
if (!ISEQ_BODY(iseq)->has_source_hash) return Qnil;
if (!ISEQ_BODY(iseq)->source_hash) return Qnil;
return ULL2NUM(ISEQ_BODY(iseq)->source_hash);
}

Expand Down
14 changes: 13 additions & 1 deletion ractor.c
Original file line number Diff line number Diff line change
Expand Up @@ -3671,7 +3671,7 @@ rb_ractor_local_storage_ptr_set(rb_ractor_local_key_t key, void *ptr)
#define DEFAULT_KEYS_CAPA 0x10

void
rb_ractor_finish_marking(void)
rb_ractor_finish_marking(bool full_mark)
{
/* A freed key's struct may only be released by a collection that purged every
* Ractor's storage with no other marker running: a global GC, or a single objspace.
Expand All @@ -3684,6 +3684,8 @@ rb_ractor_finish_marking(void)
* zombie_objspaces only marks the join slot): purge here, under the barrier, before
* the struct is freed, or a later ractor_free reads a freed key. */
rb_vm_t *vm = GET_VM();
rb_ractor_t *r;

for (size_t zi = 0; zi < vm->gc.zombie_objspaces_count; zi++) {
rb_ractor_t *owner = vm->gc.zombie_objspaces[zi].owner;
if (owner == NULL || owner->local_storage == NULL) continue;
Expand All @@ -3698,6 +3700,16 @@ rb_ractor_finish_marking(void)
freed_ractor_local_keys.capa = DEFAULT_KEYS_CAPA;
SIZED_REALLOC_N(freed_ractor_local_keys.keys, rb_ractor_local_key_t, DEFAULT_KEYS_CAPA, freed_ractor_local_keys.capa);
}

/* Under a minor mark an unmarked port is not a dead one. */
if (full_mark) {
ccan_list_for_each(&vm->ractor.set, r, vmlr_node) {
rb_ractor_reap_dead_ports(r);
}
if (vm->ractor.cnt == 0 && vm->ractor.main_ractor) {
rb_ractor_reap_dead_ports(vm->ractor.main_ractor);
}
}
}

static VALUE
Expand Down
3 changes: 2 additions & 1 deletion ractor_core.h
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,7 @@ struct rb_ractor_struct {
/* Mark the GC roots held in Ractor r's C structs (from the root scan in gc.c). */
void rb_ractor_mark_local_roots(rb_ractor_t *r);
void rb_ractor_mark_terminated_join_value(rb_ractor_t *r);
void rb_ractor_reap_dead_ports(rb_ractor_t *r);

/* Move src's registered_marks to dst and leave src empty (on join or when an orphan
* is absorbed). An absorb can run during a GC sweep, so the implementation uses raw
Expand Down Expand Up @@ -229,7 +230,7 @@ VALUE rb_ractor_ensure_shareable(VALUE obj, VALUE name);
st_table *rb_ractor_targeted_hooks(rb_ractor_t *cr);

RUBY_SYMBOL_EXPORT_BEGIN
void rb_ractor_finish_marking(void);
void rb_ractor_finish_marking(bool full_mark);

bool rb_ractor_shareable_p_continue(VALUE obj);

Expand Down
50 changes: 50 additions & 0 deletions ractor_sync.c
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,22 @@ static void ractor_off_queue_remove(struct ractor_basket *b);
void rb_ractor_courier_mark(struct rb_ractor_courier *c);
struct rb_ractor_courier *rb_ractor_courier_build_copy(VALUE obj, struct rb_ractor_courier **slot);

static void ractor_port_note_alive(const struct ractor_port *rp);

static void
ractor_port_mark(void *ptr)
{
const struct ractor_port *rp = (struct ractor_port *)ptr;

if (rp->r) {
rb_gc_mark(rp->r->pub.self);

/* Only a mark that covers every objspace can call a port dead. Ask the single
* objspace first: it answers without the VM, which a GC worker thread cannot
* reach (mmtk marks from several of them). */
if (rb_gc_single_objspace_p() || rb_gc_during_global_gc_p()) {
ractor_port_note_alive(rp);
}
}
}

Expand Down Expand Up @@ -342,13 +351,15 @@ ractor_mark_off_queue_baskets(rb_ractor_t *r)
struct ractor_queue {
struct ccan_list_head set;
bool closed;
bool alive; /* its Ractor::Port is still reachable; see the reap below */
};

static void
ractor_queue_init(struct ractor_queue *rq)
{
ccan_list_head_init(&rq->set);
rq->closed = false;
rq->alive = true;
}

static struct ractor_queue *
Expand All @@ -359,6 +370,19 @@ ractor_queue_new(void)
return rq;
}

static void
ractor_port_note_alive(const struct ractor_port *rp)
{
struct ractor_queue *rq;

if (rp->r->sync.ports && st_lookup(rp->r->sync.ports, rp->id_, (st_data_t *)&rq)) {
/* Several markers can reach the same port at once (mmtk marks from its GC worker
* threads), but they all store the same value and the reap reads it once marking
* is over. */
rq->alive = true;
}
}

static void
ractor_queue_mark(const struct ractor_queue *rq)
{
Expand Down Expand Up @@ -607,6 +631,32 @@ ractor_close_port(rb_execution_context_t *ec, rb_ractor_t *cr, const struct ract
return rq != NULL;
}

/* A port is the only way to receive from its queue, so a queue whose port is gone is
* unreachable -- but the table is keyed by id, so no sweep finds it. Mark and sweep the
* table itself: ractor_port_mark sets the flag, this clears it for the next cycle. */
static int
ractor_reap_dead_ports_i(st_data_t port_id, st_data_t val, st_data_t dat)
{
struct ractor_queue *rq = (struct ractor_queue *)val;

if (rq->alive) {
rq->alive = false;
return ST_CONTINUE;
}
else {
ractor_queue_free(rq);
return ST_DELETE;
}
}

void
rb_ractor_reap_dead_ports(rb_ractor_t *r)
{
if (r->sync.ports) {
st_foreach(r->sync.ports, ractor_reap_dead_ports_i, 0);
}
}

static int
ractor_free_all_ports_i(st_data_t port_id, st_data_t val, st_data_t dat)
{
Expand Down
3 changes: 2 additions & 1 deletion ruby_parser.c
Original file line number Diff line number Diff line change
Expand Up @@ -1119,7 +1119,8 @@ rb_source_hash_update(rb_source_hash_state_t *state, const uint8_t *ptr, size_t
uint64_t
rb_source_hash_finalize(const rb_source_hash_state_t *state)
{
return state->hash;
/* A hash of 0 means no source hash, so remap it to another value. */
return state->hash == 0 ? 1 : state->hash;
}

VALUE
Expand Down
16 changes: 16 additions & 0 deletions test/ruby/test_ractor.rb
Original file line number Diff line number Diff line change
Expand Up @@ -794,4 +794,20 @@ def test_io_priority_wait_on_mn_thread
assert_equal :ok, r.value
RUBY
end
def test_port_queue_dropped_when_port_unreachable
omit 'not fixed for mmtk: it never calls rb_ractor_finish_marking, where the reap runs' unless GC.config[:implementation] == 'default'
assert_ractor(<<~'RUBY')
200.times do
port = Ractor::Port.new
Ractor.new(port) { |p| p << Ractor::Port.new; nil }.join
end
8.times { GC.start }
# A dropped message holding a port used to root the sending Ractor, and with it
# that Ractor's whole objspace, for the life of the process: every one of the 200
# survived. A few of the last still can -- the reap needs a second full mark, and
# a conservative stack scan holds whatever it holds -- so this is not exact.
assert_operator ObjectSpace.each_object(Ractor).count, :<, 20
RUBY
end

end
12 changes: 12 additions & 0 deletions test/ruby/test_weakmap.rb
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,18 @@ def test_size
end
alias test_length test_size

def test_size_reflects_number_of_live_entries
was_disabled = GC.disable
m = ObjectSpace::WeakMap.new
Thread.new { m[Object.new] = Object.new }.join
GC.start
assert_equal(0, m.size, 'https://bugs.ruby-lang.org/issues/22251')
assert_equal(0, m.keys.size)
assert_equal(0, m.size)
ensure
GC.enable unless was_disabled
end

def test_frozen_object
o = Object.new.freeze
assert_nothing_raised(FrozenError) {@wm[o] = 'foo'}
Expand Down
2 changes: 1 addition & 1 deletion vm_backtrace.c
Original file line number Diff line number Diff line change
Expand Up @@ -601,7 +601,7 @@ location_source_range_m(VALUE self)
if (node_id == -1) {
rb_raise(rb_eRuntimeError, "cannot get source range for location without a node ID");
}
if (!ISEQ_BODY(iseq)->has_source_hash) {
if (!ISEQ_BODY(iseq)->source_hash) {
rb_raise(rb_eRuntimeError, "cannot get source range because the source hash is unavailable");
}
uint64_t source_hash = ISEQ_BODY(iseq)->source_hash;
Expand Down
6 changes: 3 additions & 3 deletions vm_core.h
Original file line number Diff line number Diff line change
Expand Up @@ -580,10 +580,10 @@ struct rb_iseq_constant_body {
void *zjit_payload;
#endif

// Hash of the source this iseq was compiled from. Meaningful only when
// has_source_hash is set.
// Hash of the source this iseq was compiled from, or 0 if it is
// unavailable. A computed hash of 0 is remapped to another value, so
// 0 never denotes a real hash.
uint64_t source_hash;
bool has_source_hash;
};

/* T_IMEMO/iseq */
Expand Down
16 changes: 16 additions & 0 deletions zjit/src/asm/arm64/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -912,6 +912,8 @@ pub fn stp_pre(cb: &mut CodeBlock, rt1: A64Opnd, rt2: A64Opnd, rn: A64Opnd) {
(A64Opnd::Reg(rt1), A64Opnd::Reg(rt2), A64Opnd::Mem(rn)) => {
assert!(rt1.num_bits == rt2.num_bits, "Expected source registers to be the same size");
assert!(imm_fits_bits(rn.disp.into(), 10), "The displacement must be 10 bits or less.");
assert!( !(rn.base_reg_no != 31 && (rn.base_reg_no == rt1.reg_no || rn.base_reg_no == rt2.reg_no)),
"Behavior is unpredictable when storing and writing back to the same register ({})", rn.base_reg_no);

RegisterPair::stp_pre(rt1.reg_no, rt2.reg_no, rn.base_reg_no, rn.disp as i16, rt1.num_bits).into()
},
Expand All @@ -929,6 +931,8 @@ pub fn stp_post(cb: &mut CodeBlock, rt1: A64Opnd, rt2: A64Opnd, rn: A64Opnd) {
(A64Opnd::Reg(rt1), A64Opnd::Reg(rt2), A64Opnd::Mem(rn)) => {
assert!(rt1.num_bits == rt2.num_bits, "Expected source registers to be the same size");
assert!(imm_fits_bits(rn.disp.into(), 10), "The displacement must be 10 bits or less.");
assert!( !(rn.base_reg_no != 31 && (rn.base_reg_no == rt1.reg_no || rn.base_reg_no == rt2.reg_no)),
"Behavior is unpredictable when storing and writing back to the same register ({})", rn.base_reg_no);

RegisterPair::stp_post(rt1.reg_no, rt2.reg_no, rn.base_reg_no, rn.disp as i16, rt1.num_bits).into()
},
Expand Down Expand Up @@ -1835,6 +1839,18 @@ mod tests {
assert_snapshot!(cb.hexdump(), @"ff7fbfa9");
}

#[test]
#[should_panic]
fn test_stp_pre_write_back_to_source() {
compile(|cb| stp_pre(cb, X0, X0, A64Opnd::new_mem(64, X0, -16)));
}

#[test]
#[should_panic]
fn test_stp_post_write_back_to_source() {
compile(|cb| stp_post(cb, X0, X0, A64Opnd::new_mem(64, X0, -16)));
}

#[test]
fn test_str_post() {
let cb = compile(|cb| str_post(cb, X10, A64Opnd::new_mem(64, X11, -16)));
Expand Down