From 614a3816f567e15a122d36596ef16d3230c1cb4f Mon Sep 17 00:00:00 2001 From: PraneethMerugu Date: Sat, 5 Sep 2026 18:07:56 -0400 Subject: [PATCH] Support tracker-safe bounded gather execution --- src/execution/candidate_stage.jl | 26 +++++++++++++----------- src/execution/mechanism_support.jl | 4 ++-- src/execution/reduce_stage.jl | 7 +++++-- src/stage_model.jl | 1 + test/test_direct_pointwise_stage.jl | 25 +++++++++++++++++++++++ test/test_reduce_stage.jl | 31 +++++++++++++++++++++++++++++ test/test_stage_model.jl | 2 ++ test/test_unique_stage.jl | 15 ++++++++++++++ 8 files changed, 95 insertions(+), 16 deletions(-) diff --git a/src/execution/candidate_stage.jl b/src/execution/candidate_stage.jl index d812ed7..1c6e5dd 100644 --- a/src/execution/candidate_stage.jl +++ b/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 diff --git a/src/execution/mechanism_support.jl b/src/execution/mechanism_support.jl index 2716b3e..d5e2526 100644 --- a/src/execution/mechanism_support.jl +++ b/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 || diff --git a/src/execution/reduce_stage.jl b/src/execution/reduce_stage.jl index d25d3d4..90a65e4 100644 --- a/src/execution/reduce_stage.jl +++ b/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, diff --git a/src/stage_model.jl b/src/stage_model.jl index 3292733..362738e 100644 --- a/src/stage_model.jl +++ b/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 diff --git a/test/test_direct_pointwise_stage.jl b/test/test_direct_pointwise_stage.jl index 3c3c84c..1fb8911 100644 --- a/test/test_direct_pointwise_stage.jl +++ b/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 diff --git a/test/test_reduce_stage.jl b/test/test_reduce_stage.jl index 26fe54c..64fc1f4 100644 --- a/test/test_reduce_stage.jl +++ b/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) diff --git a/test/test_stage_model.jl b/test/test_stage_model.jl index 5525493..c75e4fd 100644 --- a/test/test_stage_model.jl +++ b/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}}(), diff --git a/test/test_unique_stage.jl b/test/test_unique_stage.jl index a781c10..85c7f18 100644 --- a/test/test_unique_stage.jl +++ b/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