Merge adjacent implicit binding partitions

After:
```
julia> convert(Core.Binding, GlobalRef(Base, :Intrinsics))
Binding Base.Intrinsics
   617:∞ - implicit `using` resolved to constant Core.Intrinsics
   0:616 - undefined binding - guard entry

julia> convert(Core.Binding, GlobalRef(Base, :Math))
Binding Base.Math
   22128:∞ - constant binding to Base.Math
   0:22127 - backdated constant binding to Base.Math
```

There is a bit of trickiness here. In particular, the question is,
"when do we check" whether the partition next to the one we currently
looked at happens to have the same implicit resolution as our current
one. The most obvious answer is that we should do it on access, but in
practice that would require essentially scanning back and considering
every possible world age state at every lookup. This is undesirable -
the lookup is not crazy expensive, but it can add up and most world
ages we never touch, so it is also wasteful.

This instead implements a different approach where we only perform
the resolution for world ages that somebody actually asked about,
but can then subsequently merge partitions if we do find that they
are identical. The logic for that is a bit involved, since we need
to be careful to keep the datastructure valid at every point, but
does address the issue.

Fixes #57923

Author: Keno

src/jltypes.c

@@ -3266,8 +3266,10 @@ void jl_init_types(void) JL_GC_DISABLED
                         jl_svec(5, jl_any_type,
                         jl_ulong_type, jl_ulong_type, jl_any_type/*jl_binding_partition_type*/, jl_ulong_type),
                         jl_emptysvec, 0, 1, 0);
-    const static uint32_t binding_partition_atomicfields[] = { 0b01101 }; // Set fields 1, 3, 4 as atomic
+    const static uint32_t binding_partition_atomicfields[] = { 0b01111 }; // Set fields 1, 2, 3, 4 as atomic
     jl_binding_partition_type->name->atomicfields = binding_partition_atomicfields;
+    const static uint32_t binding_partition_constfields[] = { 0x100001 }; // Set fields 1, 5 as constant
+    jl_binding_partition_type->name->constfields = binding_partition_constfields;
 
     jl_binding_type =
         jl_new_datatype(jl_symbol("Binding"), core, jl_any_type, jl_emptysvec,

src/julia.h

@@ -760,7 +760,7 @@ typedef struct JL_ALIGNED_ATTR(8) _jl_binding_partition_t {
      * } restriction;
      */
     jl_value_t *restriction;
-    size_t min_world;
+    _Atomic(size_t) min_world;
     _Atomic(size_t) max_world;
     _Atomic(struct _jl_binding_partition_t *) next;
     size_t kind;

src/module.c

@@ -126,13 +126,70 @@ static void update_implicit_resolution(struct implicit_search_resolution *to_upd
 
 static jl_binding_partition_t *jl_implicit_import_resolved(jl_binding_t *b, struct implicit_search_gap gap, struct implicit_search_resolution resolution)
 {
+    size_t new_kind = resolution.ultimate_kind | gap.inherited_flags;
+    size_t new_max_world = gap.max_world < resolution.max_world ? gap.max_world : resolution.max_world;
+    size_t new_min_world = gap.min_world > resolution.min_world ? gap.min_world : resolution.min_world;
+    jl_binding_partition_t *next = gap.replace;
+    if (jl_is_binding_partition(gap.parent)) {
+        // Check if we can merge this into the previous binding partition
+        jl_binding_partition_t *prev = (jl_binding_partition_t *)gap.parent;
+        size_t expected_prev_min_world = new_max_world + 1;
+        if (prev->restriction == resolution.binding_or_const && prev->kind == new_kind) {
+            if (!jl_atomic_cmpswap(&prev->min_world, &expected_prev_min_world, new_min_world)) {
+                if (expected_prev_min_world <= new_min_world) {
+                    return prev;
+                }
+                else if (expected_prev_min_world <= new_max_world) {
+                    // Concurrent modification by another thread - bail.
+                    return NULL;
+                }
+                // There remains a gap - proceed
+            } else {
+                if (next) {
+                    size_t next_min_world = jl_atomic_load_relaxed(&next->min_world);
+                    expected_prev_min_world = new_min_world;
+                    for (;;) {
+                        // We've updated the previous partition - check if we've closed a gap
+                        size_t next_max_world = jl_atomic_load_relaxed(&next->max_world);
+                        if (next_max_world == expected_prev_min_world-1 && next->kind == new_kind && next->restriction == resolution.binding_or_const) {
+                            if (jl_atomic_cmpswap(&prev->min_world, &expected_prev_min_world, next_min_world)) {
+                                jl_binding_partition_t *nextnext = jl_atomic_load_relaxed(&next->next);
+                                if (!jl_atomic_cmpswap(&prev->next, &next, nextnext)) {
+                                    // `next` may have been merged into its subsequent partition - we need to retry
+                                    assert(next);
+                                    continue;
+                                }
+                                // N.B.: This can lose modifications to next->{min_world, next}.
+                                // However, those modifications could only have been for another implicit
+                                // partition, so we are ok to lose them and recompute them later if necessary.
+                            }
+                            assert(expected_prev_min_world <= new_min_world);
+                        }
+                        break;
+                    }
+                }
+                return prev;
+            }
+        }
+    }
     jl_binding_partition_t *new_bpart = new_binding_partition();
-    jl_atomic_store_relaxed(&new_bpart->max_world, gap.max_world < resolution.max_world ? gap.max_world : resolution.max_world);
-    new_bpart->min_world = gap.min_world > resolution.min_world ? gap.min_world : resolution.min_world;
-    new_bpart->kind = resolution.ultimate_kind | gap.inherited_flags;
+    jl_atomic_store_relaxed(&new_bpart->max_world, new_max_world);
+    new_bpart->kind = new_kind;
     new_bpart->restriction = resolution.binding_or_const;
     jl_gc_wb_fresh(new_bpart, new_bpart->restriction);
-    jl_atomic_store_relaxed(&new_bpart->next, gap.replace);
+
+    if (next) {
+        // See if we can merge the next partition into this one
+        size_t next_max_world = jl_atomic_load_relaxed(&next->max_world);
+        if (next_max_world == new_min_world - 1 && next->kind == new_kind && next->restriction == resolution.binding_or_const) {
+            // See above for potentially losing modifications to next.
+            new_min_world = jl_atomic_load_acquire(&next->min_world);
+            next = jl_atomic_load_relaxed(&next->next);
+        }
+    }
+
+    new_bpart->min_world = new_min_world;
+    jl_atomic_store_relaxed(&new_bpart->next, next);
     if (!jl_atomic_cmpswap(gap.insert, &gap.replace, new_bpart))
         return NULL;
     jl_gc_wb(gap.parent, new_bpart);
@@ -170,13 +227,11 @@ struct implicit_search_resolution jl_resolve_implicit_import(jl_binding_t *b, mo
         struct _jl_module_using data = *module_usings_getidx(m, i);
         JL_UNLOCK(&m->lock);
         if (data.min_world > world) {
-            if (max_world > data.min_world)
-                max_world = data.min_world - 1;
+            max_world = WORLDMIN(max_world, data.min_world - 1);
             continue;
         }
         if (data.max_world < world) {
-            if (min_world < data.max_world)
-                min_world = data.max_world + 1;
+            min_world = WORLDMAX(min_world, data.max_world + 1);
             continue;
         }
 
@@ -279,8 +334,13 @@ JL_DLLEXPORT jl_binding_partition_t *jl_maybe_reresolve_implicit(jl_binding_t *b
         assert(bpart == jl_atomic_load_relaxed(&b->partitions));
         assert(bpart);
         struct implicit_search_resolution resolution = jl_resolve_implicit_import(b, NULL, new_max_world+1, 0);
+        int resolution_unchanged = bpart->restriction == resolution.binding_or_const && jl_binding_kind(bpart) == resolution.ultimate_kind;
         if (resolution.min_world == bpart->min_world) {
-            assert(bpart->restriction == resolution.binding_or_const && jl_binding_kind(bpart) == resolution.ultimate_kind);
+            // The resolution has the same world bounds - it must be unchanged
+            assert(resolution_unchanged);
+            return bpart;
+        } else if (resolution_unchanged) {
+            // If the resolution is unchanged, we can still keep the bpart
             return bpart;
         }
         assert(resolution.min_world == new_max_world+1 && "Missed an invalidation or bad resolution bounds");

test/rebinding.jl

@@ -318,3 +318,68 @@ module UndefinedTransitions
         @test Base.Compiler.is_nothrow(Base.Compiler.decode_effects(ci.ipo_purity_bits))
     end
 end
+
+# Identical implicit partitions should be merge (#57923)
+for binding in (convert(Core.Binding, GlobalRef(Base, :Math)),
+                convert(Core.Binding, GlobalRef(Base, :Intrinsics)))
+    # Test that these both only have two partitions
+    @test isdefined(binding, :partitions)
+    @test isdefined(binding.partitions, :next)
+    @test !isdefined(binding.partitions.next, :next)
+end
+
+# Test various scenarios for implicit partition merging
+module MergeStress
+    for i = 1:5
+        @eval module $(Symbol("M$i"))
+            export x, y
+            const x = 1
+            const y = 2
+        end
+    end
+    const before = Base.get_world_counter()
+    using .M1
+    const afterM1 = Base.get_world_counter()
+    using .M2
+    const afterM2 = Base.get_world_counter()
+    using .M3
+    const afterM3 = Base.get_world_counter()
+    using .M4
+    const afterM4 = Base.get_world_counter()
+    using .M5
+    const afterM5 = Base.get_world_counter()
+end
+
+function count_partitions(b::Core.Binding)
+    n = 0
+    isdefined(b, :partitions) || return n
+    bpart = b.partitions
+    while true
+        n += 1
+        isdefined(bpart, :next) || break
+        bpart = bpart.next
+    end
+    return n
+end
+using Base: invoke_in_world
+
+const xbinding = convert(Core.Binding, GlobalRef(MergeStress, :x))
+function access_and_count(point)
+    invoke_in_world(getglobal(MergeStress, point), getglobal, MergeStress, :x)
+    count_partitions(xbinding)
+end
+
+@test count_partitions(xbinding) == 0
+@test access_and_count(:afterM1) == 1
+# M2 is the first change to the `usings` table after M1. The partitions
+# can and should be merged
+@test access_and_count(:afterM2) == 1
+
+# There is a gap between M2 and M5 - the partitions should not be merged
+@test access_and_count(:afterM5) == 2
+
+# M4 and M5 are adjacent, these partitions should also be merged (in the opposite direction)
+@test access_and_count(:afterM4) == 2
+
+# M3 connects all, so we should have a single partition
+@test access_and_count(:afterM3) == 1