[mlir][OpenACC] Atomicize contended shared array reduction updates (#212971)

Example:
```fortran
!$acc parallel loop gang reduction(+:a)
do i = 1, N
  !$acc loop worker reduction(+:a)
  do j = 1, M
    a(i) = a(i) + b(j,i)
  end do
end do
```

In this code, the worker accumulator is block-shared, so every worker
updates
the same element with a plain read-modify-write and all but one partial
is lost.

Fix: make in-place updates of a block-shared array accumulator atomic,
unless
the element index varies per thread, in which case each thread owns its
element
and the plain update is already race-free.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matsu 2026-07-30 14:02:45 -07:00 committed by GitHub
parent 7650a43aa8
commit c836b82d08
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 301 additions and 35 deletions

View file

@ -382,6 +382,21 @@ static Value castPointerLikeTypeIfNeeded(OpBuilder &builder, Location loc,
/// looking through view/cast ops.
static acc::PrivateLocalOp getPrivateLocalForMemref(Value memref);
/// Returns the dimensions that own \p privateLocal.
static GPUParallelDimsAttr
getPrivateParDims(acc::PrivateLocalOp privateLocal,
acc::ComputeRegionOp computeRegion);
/// True when the storage backing \p privateLocal is thread_x-private. An
/// unknown scope conservatively counts as per-thread.
static bool storageHasThreadX(acc::PrivateLocalOp privateLocal,
acc::ComputeRegionOp computeRegion) {
GPUParallelDimsAttr dims = getPrivateParDims(privateLocal, computeRegion);
return !dims || llvm::any_of(dims.getArray(), [](GPUParallelDimAttr d) {
return d.isThreadX();
});
}
/// Returns the sole user of \p v, or null if it has zero or multiple uses.
static Operation *getOnlyUser(Value v) {
if (!v.hasOneUse())
@ -777,8 +792,11 @@ static void initPerThreadArrayAccum(OpBuilder &b, Location loc, Value alloca,
std::optional<int64_t>
ACCCGToGPULowering::isEligibleForSharedMemory(acc::PrivateLocalOp privateLocal,
MemRefType baseTy) {
// Cross-thread array reduction accumulators must stay per-thread.
if (perThreadArrayReductionAccum(privateLocal.getResult()))
// Cross-thread array reduction accumulators must stay per-thread when their
// storage scope includes thread_x. Gang-/worker-scoped array temps remain
// eligible for shared memory.
if (perThreadArrayReductionAccum(privateLocal.getResult()) &&
storageHasThreadX(privateLocal, computeRegion))
return std::nullopt;
ModuleOp module = computeRegion->getParentOfType<ModuleOp>();
FailureOr<bool> isCandidate = isPrivateLocalSharedMemoryCandidate(
@ -1225,13 +1243,14 @@ static bool isRedundantChainAccumulate(acc::ReductionAccumulateOp op) {
return false;
}
/// Returns the dimensions that own \p privateLocal.
static GPUParallelDimsAttr
getPrivateParDims(acc::PrivateLocalOp privateLocal,
acc::ComputeRegionOp computeRegion) {
if (GPUParallelDimsAttr parDims = acc::getParDimsAttr(privateLocal))
return parDims;
return getPrivatizeOp(privateLocal, computeRegion).getParDimsAttr();
if (acc::PrivatizeOp privatize = getPrivatizeOp(privateLocal, computeRegion))
return privatize.getParDimsAttr();
return {};
}
/// True when \p privateLocal has one private slot per ThreadY row.
@ -2707,10 +2726,12 @@ void ACCCGToGPULowering::processPrivateLocal(
} else {
// Hoisted acc.privatize: allocate per-thread stack storage in the launch
// body. Cross-thread array reduction accumulators are per-thread too, so
// the accumulate can reduce each element across threads.
// the accumulate can reduce each element across threads. Skip when storage
// par_dims lack thread_x (gang-/worker-scoped array temp).
acc::ReductionAccumulateArrayOp arrayAccum =
perThreadArrayReductionAccum(privateLocal.getResult());
if ((isThreadXPrivatize(privatizeOp) || arrayAccum) &&
if ((isThreadXPrivatize(privatizeOp) ||
(arrayAccum && storageHasThreadX(privateLocal, computeRegion))) &&
canUseStackAlloca(baseTy, loc, options.maxThreadPrivateStack)) {
Value alloca = memref::AllocaOp::create(rewriter, loc, baseTy);
if (arrayAccum) {
@ -2799,7 +2820,8 @@ void ACCCGToGPULowering::processPrivateLocal(
acc::ReductionAccumulateArrayOp arrayAccum =
perThreadArrayReductionAccum(privateLocal.getResult());
for (mlir::acc::GPUParallelDimAttr parDim : parDimsPair.first) {
if ((parDim.isThreadX() || arrayAccum) &&
if ((parDim.isThreadX() ||
(arrayAccum && storageHasThreadX(privateLocal, computeRegion))) &&
canUseStackAlloca(baseTy, loc, options.maxThreadPrivateStack)) {
Value alloca = memref::AllocaOp::create(rewriter, loc, baseTy);
if (arrayAccum) {
@ -3471,6 +3493,122 @@ void ACCCGToGPULowering::processAccumulateOp(acc::ReductionAccumulateOp op) {
}
}
/// True when \p v takes a distinct value per thread: it derives from a thread
/// id, either directly or through the bounds of an enclosing loop. By this
/// point a thread-mapped loop carries its mapping in its bounds rather than in
/// `acc.par_dims`, so the bounds are what must be inspected.
static bool isThreadVarying(Value v, ArrayRef<Value> threadIds,
DenseSet<Value> &visited) {
if (!v || !visited.insert(v).second)
return false;
if (llvm::is_contained(threadIds, v))
return true;
if (auto arg = dyn_cast<BlockArgument>(v)) {
Operation *owner = arg.getOwner()->getParentOp();
unsigned dim = arg.getArgNumber();
if (auto loop = dyn_cast<scf::ParallelOp>(owner)) {
if (dim >= loop.getLowerBound().size())
return false;
return isThreadVarying(loop.getLowerBound()[dim], threadIds, visited) ||
isThreadVarying(loop.getStep()[dim], threadIds, visited);
}
if (auto loop = dyn_cast<scf::ForOp>(owner))
return dim == 0 &&
(isThreadVarying(loop.getLowerBound(), threadIds, visited) ||
isThreadVarying(loop.getStep(), threadIds, visited));
return false;
}
Operation *def = v.getDefiningOp();
if (!def)
return false;
if (isa<gpu::ThreadIdOp, gpu::LaneIdOp>(def))
return true;
return llvm::any_of(def->getOperands(), [&](Value o) {
return isThreadVarying(o, threadIds, visited);
});
}
/// Strips view and memory-space-cast chains to the underlying buffer.
static Value accumulatorRoot(Value v) {
while (Operation *op = v.getDefiningOp()) {
if (auto cast = dyn_cast<memref::MemorySpaceCastOp>(op)) {
v = cast.getSource();
continue;
}
if (auto viewLike = dyn_cast<ViewLikeOpInterface>(op)) {
if (isa<MemRefType>(viewLike.getViewSource().getType())) {
v = viewLike.getViewSource();
continue;
}
}
break;
}
return v;
}
/// Matches `%l = load %m[%i]` / `%c = combine(%l, %x)` / `store %c, %m[%i]` on
/// the accumulator \p accum and returns the contributed value `%x`.
static Value matchAccumulatorUpdate(memref::StoreOp store, Value accum) {
if (accumulatorRoot(store.getMemRef()) != accum)
return {};
Operation *combine = store.getValueToStore().getDefiningOp();
if (!combine || combine->getNumOperands() != 2)
return {};
for (unsigned i = 0; i != 2; ++i) {
auto load = combine->getOperand(i).getDefiningOp<memref::LoadOp>();
if (!load || accumulatorRoot(load.getMemRef()) != accum)
continue;
if (!llvm::equal(load.getIndices(), store.getIndices()))
continue;
return combine->getOperand(1 - i);
}
return {};
}
/// A block-shared accumulator is updated in place by the loop body, so several
/// threads may hit the same element. Make those updates atomic unless the
/// element index provably varies across the participating threads.
static void atomicizeSharedAccumulatorUpdates(Value accum,
arith::AtomicRMWKind kind,
ArrayRef<Value> threadIds,
RewriterBase &rewriter) {
OpBuilder::InsertionGuard guard(rewriter);
SmallVector<memref::StoreOp> stores;
SmallVector<Value> worklist{accum};
DenseSet<Value> seen;
while (!worklist.empty()) {
Value cur = worklist.pop_back_val();
if (!seen.insert(cur).second)
continue;
for (Operation *user : cur.getUsers()) {
if (auto store = dyn_cast<memref::StoreOp>(user))
stores.push_back(store);
else if (isa<ViewLikeOpInterface, memref::MemorySpaceCastOp>(user))
llvm::append_range(worklist, user->getResults());
}
}
for (memref::StoreOp store : stores) {
Value contribution = matchAccumulatorUpdate(store, accum);
if (!contribution)
continue;
// A thread-varying index means each thread owns its element, so the
// existing plain update is already race-free.
if (llvm::any_of(store.getIndices(), [&](Value idx) {
DenseSet<Value> visited;
return isThreadVarying(idx, threadIds, visited);
}))
continue;
Operation *combine = store.getValueToStore().getDefiningOp();
rewriter.setInsertionPoint(store);
memref::AtomicRMWOp::create(rewriter, store.getLoc(), kind, contribution,
store.getMemRef(), store.getIndices());
rewriter.eraseOp(store);
if (combine && combine->use_empty())
rewriter.eraseOp(combine);
}
}
void ACCCGToGPULowering::processAccumulateArrayOp(
acc::ReductionAccumulateArrayOp op) {
LLVM_DEBUG(llvm::dbgs() << "processing accumulate array op: " << *op << "\n");
@ -3526,35 +3664,55 @@ void ACCCGToGPULowering::processAccumulateArrayOp(
// Per-element gpu.all_reduce is only correct when each thread owns its own
// accumulator copy. For a statically-shaped accumulator, classify from the
// operand: an explicit shared allocation is block-shared regardless of size,
// and anything else (a per-thread stack alloca, or a view over one) is
// per-thread when it fits the per-thread stack budget and block-shared when
// it is too large. For a dynamically-shaped accumulator the type conveys no
// size, so classify from par_dims (which the producer sets to the reduction's
// actual parallel scope): a thread dimension means per-thread storage.
bool isPerThreadPrivate;
if (memrefTy.hasStaticShape()) {
Operation *rootOp = unwrapMemRefConversion(memref).getDefiningOp();
isPerThreadPrivate =
!isa_and_nonnull<memref::AllocOp>(rootOp) &&
canUseStackAlloca(memrefTy, loc, options.maxThreadPrivateStack);
} else {
isPerThreadPrivate = llvm::any_of(
op.getParDims().getArray(),
[](mlir::acc::GPUParallelDimAttr d) { return d.isThreadX(); });
}
if (!isPerThreadPrivate) {
// Block-shared accumulator: no-op only when the accumulate spans a block
// dim (threads distribute distinct elements, so the block partial is in
// place and the atomic combine finishes it). A thread-only shared
// reduction, where several threads reduce into the same element, is not yet
// supported.
if (hasBlockDim) {
eraseDeadBounds();
} else {
(void)accSupport.emitNYI(
loc, "reduction: shared-memory array reduction accumulate");
// operand: an explicit shared/heap allocation is block-shared regardless of
// size, and a stack alloca (or a view over one) is per-thread when it fits
// the per-thread stack budget. Storage `acc.par_dims` without thread_x means
// gang-/worker-scoped privacy (shared among vector lanes) even if the type
// would fit on the stack. For a dynamically-shaped accumulator the type
// conveys no size, so classify from storage/accumulate thread_x dims.
auto storageIsThreadXPrivate = [&](Value v) -> bool {
acc::PrivateLocalOp privateLocal = getPrivateLocalForMemref(v);
GPUParallelDimsAttr dims =
privateLocal ? getPrivateParDims(privateLocal, computeRegion)
: GPUParallelDimsAttr();
if (!dims) {
if (Operation *root = unwrapMemRefConversion(v).getDefiningOp())
dims = getParDimsAttr(root);
}
return !dims ||
llvm::any_of(dims.getArray(), [](auto d) { return d.isThreadX(); });
};
Operation *rootOp = unwrapMemRefConversion(memref).getDefiningOp();
bool isSharedStorage = isa_and_nonnull<memref::AllocOp>(rootOp) ||
isa_and_nonnull<acc::GPUSharedMemoryOp>(rootOp);
if (auto addrSpace = dyn_cast_if_present<gpu::AddressSpaceAttr>(
memrefTy.getMemorySpace())) {
isSharedStorage |=
addrSpace.getValue() == gpu::GPUDialect::getWorkgroupAddressSpace();
}
bool isPerThreadPrivate =
!isSharedStorage && storageIsThreadXPrivate(op.getMemref()) &&
(memrefTy.hasStaticShape()
? canUseStackAlloca(memrefTy, loc, options.maxThreadPrivateStack)
: llvm::any_of(op.getParDims().getArray(),
[](mlir::acc::GPUParallelDimAttr d) {
return d.isThreadX();
}));
if (!isPerThreadPrivate) {
// Block-shared accumulator: the body already updated it in place, so the
// block partial is complete and the atomic combine finishes across blocks.
// Threads that share an element must not race, so their in-place updates
// become atomic.
SmallVector<Value> threadIds;
if (Value xId = getGPUThreadIdFor(gpu::Processor::ThreadX))
threadIds.push_back(xId);
if (Value yId = getGPUThreadIdFor(gpu::Processor::ThreadY))
threadIds.push_back(yId);
if (Value zId = getGPUThreadIdFor(gpu::Processor::ThreadZ))
threadIds.push_back(zId);
atomicizeSharedAccumulatorUpdates(accumulatorRoot(memref), kind, threadIds,
rewriter);
eraseDeadBounds();
return;
}

View file

@ -5,9 +5,12 @@
// all_reduce would scale the block partial by the thread count. The block
// partial is already in place and the atomic combine merges across blocks, so
// the accumulate lowers away to nothing (like the block-only case).
// Here every thread updates element 0, so the in-place update must be atomic
// or the threads race and partials are lost.
// CHECK-LABEL: func.func @array_reduction_shared
// CHECK: gpu.launch
// CHECK: memref.atomic_rmw addi
// CHECK-NOT: gpu.all_reduce
// CHECK-NOT: acc.reduction_accumulate_array
// CHECK-NOT: acc.bounds
@ -58,3 +61,108 @@ func.func @array_reduction_shared(%arg0: memref<8192xi32>) {
acc.copyout accPtr(%0 : memref<8192xi32>) to varPtr(%arg0 : memref<8192xi32>) {dataClause = #acc<data_clause acc_reduction>, implicit = true, name = "r"}
return
}
// The same block-shared accumulator, but each thread updates its own element.
// The update is already race-free, so it must stay a plain store.
// CHECK-LABEL: func.func @array_reduction_shared_partitioned
// CHECK: gpu.launch
// CHECK-NOT: memref.atomic_rmw
// CHECK-NOT: gpu.all_reduce
// CHECK-NOT: acc.reduction_accumulate_array
func.func @array_reduction_shared_partitioned(%arg0: memref<8192xi32>) {
%0 = acc.copyin varPtr(%arg0 : memref<8192xi32>) -> memref<8192xi32> {dataClause = #acc<data_clause acc_reduction>, implicit = true, name = "r"}
acc.kernel_environment dataOperands(%0 : memref<8192xi32>) {
%c1_pw = arith.constant 1 : index
%c128 = arith.constant 128 : index
%bx = acc.par_width %c1_pw {par_dim = #acc.par_dim<block_x>}
%tx = acc.par_width %c128 {par_dim = #acc.par_dim<thread_x>}
acc.compute_region launch(%kbx = %bx, %ktx = %tx) ins(%arg2 = %0) : (memref<8192xi32>) {
%c8192 = arith.constant 8192 : index
%c0_i32 = arith.constant 0 : i32
%c1_i32 = arith.constant 1 : i32
%c1 = arith.constant 1 : index
%c0 = arith.constant 0 : index
%2 = acc.reduction_init %arg2 <add> : memref<8192xi32> {
%alloc = memref.alloc() : memref<8192xi32>
scf.parallel (%i) = (%c0) to (%c8192) step (%c1) {
memref.store %c0_i32, %alloc[%i] : memref<8192xi32>
scf.reduce
} {acc.par_dims = #acc<par_dims[thread_x]>}
acc.yield %alloc : memref<8192xi32>
}
scf.parallel (%bx_iv) = (%c0) to (%kbx) step (%c1) {
scf.parallel (%tx_iv) = (%c0) to (%c8192) step (%c1) {
%3 = memref.load %2[%tx_iv] : memref<8192xi32>
%4 = arith.addi %3, %c1_i32 : i32
memref.store %4, %2[%tx_iv] : memref<8192xi32>
scf.reduce
} {acc.par_dims = #acc<par_dims[thread_x]>}
scf.reduce
} {acc.par_dims = #acc<par_dims[block_x]>}
%b = acc.bounds extent(%c8192 : index)
acc.reduction_accumulate_array %2 bounds(%b) <add> : memref<8192xi32> {par_dims = #acc<par_dims[block_x, thread_x]>}
acc.reduction_combine_region %2 into %arg2 : memref<8192xi32> {
scf.for %i = %c0 to %c8192 step %c1 {
%3 = memref.load %2[%i] : memref<8192xi32>
%4 = memref.load %arg2[%i] : memref<8192xi32>
%5 = arith.addi %3, %4 : i32
memref.store %5, %arg2[%i] : memref<8192xi32>
}
}
acc.yield
} {origin = "acc.parallel"}
}
acc.copyout accPtr(%0 : memref<8192xi32>) to varPtr(%arg0 : memref<8192xi32>) {dataClause = #acc<data_clause acc_reduction>, implicit = true, name = "r"}
return
}
// A gang-scoped array private_local (block-only storage dims) feeding a
// worker-level accumulate must not take the per-thread all_reduce path. Every
// worker updates element 0 of the gang-private slot, so the update must be
// atomic.
// CHECK-LABEL: func.func @array_reduction_gang_storage_thread_accum
// CHECK: gpu.launch
// CHECK: memref.atomic_rmw addi
// CHECK-NOT: gpu.all_reduce
// CHECK-NOT: acc.reduction_accumulate_array
func.func @array_reduction_gang_storage_thread_accum(%arg0: memref<4xi32>) {
%0 = acc.copyin varPtr(%arg0 : memref<4xi32>) -> memref<4xi32> {dataClause = #acc<data_clause acc_reduction>, implicit = true, name = "r"}
acc.kernel_environment dataOperands(%0 : memref<4xi32>) {
%c2 = arith.constant 2 : index
%c4 = arith.constant 4 : index
%c32 = arith.constant 32 : index
%bx = acc.par_width %c2 {par_dim = #acc.par_dim<block_x>}
%wy = acc.par_width %c4 {par_dim = #acc.par_dim<thread_y>}
%tx = acc.par_width %c32 {par_dim = #acc.par_dim<thread_x>}
%private = acc.privatize [#acc<par_dims[block_x]>] : () -> !acc.private_type<memref<4xi32>>
acc.compute_region launch(%kbx = %bx, %kwy = %wy, %ktx = %tx) ins(%arg2 = %0, %priv = %private) : (memref<4xi32>, !acc.private_type<memref<4xi32>>) {
%c0 = arith.constant 0 : index
%c1 = arith.constant 1 : index
%c0_i32 = arith.constant 0 : i32
%c1_i32 = arith.constant 1 : i32
%c4_idx = arith.constant 4 : index
scf.parallel (%bx_iv) = (%c0) to (%kbx) step (%c1) {
%local = acc.private_local %priv {acc.par_dims = #acc<par_dims[block_x]>} : (!acc.private_type<memref<4xi32>>) -> memref<4xi32>
scf.for %i = %c0 to %c4_idx step %c1 {
memref.store %c0_i32, %local[%i] : memref<4xi32>
}
scf.parallel (%wy_iv) = (%c0) to (%kwy) step (%c1) {
%3 = memref.load %local[%c0] : memref<4xi32>
%4 = arith.addi %3, %c1_i32 : i32
memref.store %4, %local[%c0] : memref<4xi32>
scf.reduce
} {acc.par_dims = #acc<par_dims[thread_y]>}
%b = acc.bounds extent(%c4_idx : index)
acc.reduction_accumulate_array %local bounds(%b) <add> : memref<4xi32> {par_dims = #acc<par_dims[thread_y]>}
scf.reduce
} {acc.par_dims = #acc<par_dims[block_x]>}
acc.yield
} {origin = "acc.parallel"}
}
acc.copyout accPtr(%0 : memref<4xi32>) to varPtr(%arg0 : memref<4xi32>) {dataClause = #acc<data_clause acc_reduction>, implicit = true, name = "r"}
return
}