Support tracker-safe bounded gather execution

src/execution/candidate_stage.jl

@@ -689,18 +689,20 @@ end
         if item <= stage.source_count && prefix_valid
             _, active = _stage_control_state(
                 stage, qualified.parameters, item)
-            access_valid = _stage_accesses_valid(
-                stage.accesses, stage.fields, item)
-            access_valid || _candidate_fail!(
-                diagnostics, _CANDIDATE_FAILURE_RELATION)
-            if active && access_valid
-                result = _call_stage_evaluator(qualified, item,
-                    _stage_reads(stage, item,
-                        _CandidateEvaluationValidation(diagnostics)),
-                    qualified.parameters)
-                _claim_publications!(
-                    publications, workspaces,
-                    result, stage.fields, item, 1, diagnostics)
+            if active
+                access_valid = _stage_accesses_valid(
+                    stage.accesses, stage.fields, item)
+                access_valid || _candidate_fail!(
+                    diagnostics, _CANDIDATE_FAILURE_RELATION)
+                if access_valid
+                    result = _call_stage_evaluator(qualified, item,
+                        _stage_reads(stage, item,
+                            _CandidateEvaluationValidation(diagnostics)),
+                        qualified.parameters)
+                    _claim_publications!(
+                        publications, workspaces,
+                        result, stage.fields, item, 1, diagnostics)
+                end
             end
         end
     end

src/execution/mechanism_support.jl

@@ -138,8 +138,8 @@ function _centrally_qualified_resolved_record(backend, type::Type)
     return qualified && offset == sizeof(type)
 end
 
-const _WIDE_RESOLVED_RECORD_MAX_FIELDS = 12
-const _WIDE_RESOLVED_RECORD_MAX_BYTES = 48
+const _WIDE_RESOLVED_RECORD_MAX_FIELDS = 16
+const _WIDE_RESOLVED_RECORD_MAX_BYTES = 64
 const _WIDE_RESOLVED_RECORD_MAX_ALIGNMENT = 8
 
 _requires_wide_component_record(type::Type) = fieldcount(type) > 8 ||

src/execution/reduce_stage.jl

@@ -455,8 +455,11 @@ function _candidate_workspace_from_tree(tree, spec,
 end
 
 function _require_reduce_value_capabilities(backend, ::Type{T}) where {T}
-    all(operation -> _centrally_qualified_value_capability(
-        backend, T, operation, :global), (:load, :store)) || throw(
+    all((:load, :store)) do operation
+        _centrally_qualified_value_capability(
+            backend, T, operation, :global) ||
+            _centrally_qualified_stage_record(backend, T, operation)
+    end || throw(
         LocalMathValidationError(
             "the backend lacks the reviewed Reduce value capability";
             stage = :prepare, contract = :reduce_backend_capability,

src/stage_model.jl

@@ -61,6 +61,7 @@ function _device_type_parameter(parameter)
     # `_device_evaluator_capture`, and the closed-callable effect analysis is
     # still the final authority over the method body.
     parameter isa Symbol && return true
+    parameter === nothing && return true
     parameter isa Union{UUIDs.UUID,Val,Ptr,Ref,NamedTuple} && return false
     parameter isa Bool && return true
     parameter isa Enum && return true

test/test_direct_pointwise_stage.jl

@@ -304,6 +304,31 @@ end
     @test LMDP._logical_lowering_entries(
         strict_plan.lowering)[1].executor.layout isa
         LMDP._GroupedCandidateLayout
+
+    prefix = LMDP.Parameter(:prefix, Int32; bounds = (Int32(0), Int32(3)))
+    partial = LMDP.Publication((LMDP.FieldPublication(
+        output, identity, LMDP.PublicationValue(:value)),),
+        LMDP.Unique(Int32; coverage = LMDP.PartialCoverage(),
+            onempty = LMDP.PreserveEmpty()))
+    prefixed_law = LMDP.LocalLaw(LMDP.Stage(
+        source, (values = LMDP.Access(values, strict; required = false),),
+        (partial,), LMDP.Evaluator(OptionalIndexEvaluator()),
+        LMDP.Control(; prefix),
+        LMDP.SourceOrigin(:direct_pointwise_inactive_relation, 1));
+        parameters = LMDP.ParameterSchema(prefix))
+    prefixed_storage = fill(Int32(-1), 3)
+    prefixed_bound = _direct_pointwise_bound(prefixed_law, (
+        keys => Int32[2, 0, 0],
+        values => Int32[10, 20, 30],
+        output => prefixed_storage,
+    ), (identity, strict))
+    prefixed_plan = LMDP.plan(prefixed_bound; backend)
+    @test LMDP._logical_lowering_entries(
+        prefixed_plan.lowering)[1].executor.layout isa
+        LMDP._GroupedCandidateLayout
+    wait(LMDP.execute!(
+        LMDP.prepare(prefixed_plan); parameters = (prefix = Int32(1),)))
+    @test prefixed_storage == Int32[20, -1, -1]
 end
 
 @testset "pointwise lowering rejects off-item destination reads" begin

test/test_reduce_stage.jl

@@ -20,6 +20,17 @@ struct MixedConflictEvaluator end
     unique = LMR.UniqueValue(item),
     reduction = LMR.Contribution(item),
 )
+struct RecordContributionValue
+    unsigned::UInt32
+    signed::Int32
+end
+struct RecordContribution end
+@inline (::RecordContribution)(item::Int32, reads, parameters) =
+    (value = LMR.Contribution(RecordContributionValue(
+        UInt32(item), -item)),)
+@inline _combine_record_contributions(left::RecordContributionValue,
+    right::RecordContributionValue) = RecordContributionValue(
+        left.unsigned + right.unsigned, left.signed + right.signed)
 
 function _reduce_test_stage(source, output, relation, law, evaluator;
         role = :value, control = LMR.Control())
@@ -47,6 +58,26 @@ function _run_test_candidate!(prepared)
     return only(prepared.runtime.launches).stage
 end
 
+@testset "canonical Reduce admits reviewed isbits record storage" begin
+    source = LMR.Space(ReduceStageNode, 3)
+    destination = LMR.Space(ReduceStageNode, 1)
+    output = LMR.Field(destination, RecordContributionValue)
+    relation = LMR.FixedRelation(source => destination; degree = 1)
+    seed = RecordContributionValue(UInt32(0), Int32(0))
+    law = LMR.Reduce(RecordContributionValue, _combine_record_contributions;
+        seed = LMR.IdentitySeed(seed), order = LMR.CanonicalLeftFold())
+    stage = _reduce_test_stage(
+        source, output, relation, law, RecordContribution())
+    storage = fill(seed, 1)
+    endpoints = reshape(fill(Int32(1), 3), 1, 3)
+    bound = LMR._bind_law(LMR.LocalLaw(stage), LMR._StructuralBinding(
+        (LMR._field_storage_binding(output, storage),),
+        (LMR._relation_storage_binding(relation, (
+            endpoints, counts = fill(Int32(1), 3))),)))
+    _run_test_candidate!(_prepare_test_candidate(bound))
+    @test storage == [RecordContributionValue(UInt32(6), Int32(-6))]
+end
+
 @testset "canonical Reduce is the exact item-major lane-minor left fold" begin
     source = LMR.Space(ReduceStageNode, 3)
     destination = LMR.Space(ReduceStageNode, 2)

test/test_stage_model.jl

@@ -280,6 +280,8 @@ struct SMForeignBounds <: LMM._ParameterBounds end
     # cannot hide in an otherwise empty/isbits callable.
     symbol_identity = SMHostileParameterEvaluator{:label}()
     @test LMM.Evaluator(symbol_identity).evaluator === symbol_identity
+    absent_identity = SMHostileParameterEvaluator{nothing}()
+    @test LMM.Evaluator(absent_identity).evaluator === absent_identity
     for hostile in (
             SMHostileParameterEvaluator{LMM.UUIDs.uuid4()}(),
             SMHostileParameterEvaluator{Val{1}}(),

test/test_unique_stage.jl

@@ -169,6 +169,21 @@ end
     ))
     _run_test_unique!(_prepare_test_unique(bound))
     @test collect(storage) == fill(record, 2)
+
+    lane_record = ntuple(Int32, 16)
+    lane_output = LMU.Field(destination, typeof(lane_record))
+    lane_stage = _unique_test_stage(source, lane_output, relation,
+        LMU.Unique(typeof(lane_record)), UniqueConstantEvaluator(lane_record))
+    lane_storage = LMU.StructArrays.StructArray(fill(lane_record, 2))
+    lane_bound = LMU._bind_law(LMU.LocalLaw(lane_stage), LMU._StructuralBinding(
+        (LMU._field_storage_binding(lane_output, lane_storage),),
+        (LMU._relation_storage_binding(relation, (
+            endpoints = reshape(Int32[1, 2], 1, 2),
+            counts = Int32[1, 1],
+        )),),
+    ))
+    _run_test_unique!(_prepare_test_unique(lane_bound))
+    @test collect(lane_storage) == fill(lane_record, 2)
 end
 
 @testset "static relation borrowing and dynamic relation validation" begin