Repository navigation
Submit kernel launches in batches - #693
Merged
Merged
Conversation
Contributor
|
Your PR requires formatting changes to meet the project's style guidelines. Click here to view the suggested changes.diff --git a/src/array.jl b/src/array.jl
index eba51c8..11554c8 100644
--- a/src/array.jl
+++ b/src/array.jl
@@ -105,7 +105,7 @@ end
See also: [`oneVector`](@ref), [`oneMatrix`](@ref), [`is_device`](@ref), [`is_shared`](@ref)
"""
mutable struct oneArray{T,N,B} <: AbstractGPUArray{T,N}
- data::DataRef{Managed{B}}
+ data::DataRef{Managed{B}}
maxsize::Int # maximum data size; excluding any selector bytes
offset::Int # offset of the data in the buffer, in bytes
@@ -125,13 +125,14 @@ mutable struct oneArray{T,N,B} <: AbstractGPUArray{T,N}
dev = device()
alignment = Base.datatype_alignment(T)
data = GPUArrays.cached_alloc((oneArray, B, ctx, dev, bufsize, alignment)) do
- DataRef(release, alloc_managed(B, ctx, dev, bufsize, alignment))
+ DataRef(release, alloc_managed(B, ctx, dev, bufsize, alignment))
end
obj = new{T,N,B}(data, maxsize, 0, dims)
finalizer(unsafe_free!, obj)
end
- function oneArray{T,N}(data::DataRef{Managed{B}}, dims::Dims{N};
+ function oneArray{T, N}(
+ data::DataRef{Managed{B}}, dims::Dims{N};
maxsize::Int=prod(dims) * sizeof(T), offset::Int=0) where {T,N,B}
check_eltype(T)
if sizeof(T) == 0
@@ -224,11 +225,11 @@ Base.size(x::oneArray) = x.dims
Base.sizeof(x::oneArray) = Base.elsize(x) * length(x)
function context(A::oneArray)
- return oneL0.context(A.data[].mem)
+ return oneL0.context(A.data[].mem)
end
function device(A::oneArray)
- return oneL0.device(A.data[].mem)
+ return oneL0.device(A.data[].mem)
end
buftype(x::oneArray) = buftype(typeof(x))
@@ -353,22 +354,22 @@ function Base.unsafe_convert(::Type{Ptr{T}}, x::oneArray{T}) where {T}
if is_device(x)
throw(ArgumentError("cannot take the CPU address of a $(typeof(x))"))
end
- managed = x.data[]
- maybe_synchronize(managed)
- convert(Ptr{T}, managed.mem) + x.offset
+ managed = x.data[]
+ maybe_synchronize(managed)
+ return convert(Ptr{T}, managed.mem) + x.offset
end
# device pointers record the task's stream as the memory's user, assuming the pointer is
# used by work submitted there. operations that know their submission target take
# ownership themselves, and use `device_pointer` instead.
function Base.unsafe_convert(::Type{ZePtr{T}}, x::oneArray{T}) where {T}
- managed = x.data[]
- @lock managed.lock take_ownership!(managed, global_stream(context(x), device()))
- device_pointer(x)
+ managed = x.data[]
+ @lock managed.lock take_ownership!(managed, global_stream(context(x), device()))
+ return device_pointer(x)
end
-device_pointer(x::oneArray{T}, i::Integer=1) where {T} =
- convert(ZePtr{T}, x.data[].mem) + x.offset + (i-1) * Base.elsize(x)
+device_pointer(x::oneArray{T}, i::Integer = 1) where {T} =
+ convert(ZePtr{T}, x.data[].mem) + x.offset + (i - 1) * Base.elsize(x)
## indexing
@@ -401,7 +402,7 @@ function typetagdata(a::oneArray, i=1)
# for zero-size element types (e.g. singleton unions), the byte offset
# is always zero, so the corresponding element offset is also zero
elem_offset = iszero(Base.elsize(a)) ? 0 : a.offset ÷ Base.elsize(a)
- return convert(ZePtr{UInt8}, a.data[].mem) + a.maxsize + elem_offset + i - 1
+ return convert(ZePtr{UInt8}, a.data[].mem) + a.maxsize + elem_offset + i - 1
end
function Base.copyto!(dest::oneArray{T}, doffs::Integer, src::Array{T}, soffs::Integer,
@@ -449,14 +450,14 @@ Base.copyto!(dest::oneDenseArray{T}, src::oneDenseArray{T}) where {T} =
function Base.unsafe_copyto!(ctx::ZeContext, dev::ZeDevice,
dest::oneDenseArray{T}, doffs, src::Array{T}, soffs, n) where T
- stream = global_stream(ctx, dev)
+ stream = global_stream(ctx, dev)
GC.@preserve src dest begin
- with_ownership(stream, Managed[dest.data[]]) do
- unsafe_copyto!(ctx, dev, device_pointer(dest, doffs), pointer(src, soffs), n)
- end
+ with_ownership(stream, Managed[dest.data[]]) do
+ unsafe_copyto!(ctx, dev, device_pointer(dest, doffs), pointer(src, soffs), n)
+ end
# Keep pageable host memory alive until the queued copy completes.
- synchronize(stream)
+ synchronize(stream)
end
if Base.isbitsunion(T)
# copy selector bytes
@@ -468,30 +469,30 @@ end
function Base.unsafe_copyto!(ctx::ZeContext, dev::ZeDevice,
dest::Array{T}, doffs, src::oneDenseArray{T}, soffs, n) where T
- stream = global_stream(ctx, dev)
- GC.@preserve src dest begin
- with_ownership(stream, Managed[src.data[]]) do
- unsafe_copyto!(ctx, dev, pointer(dest, doffs), device_pointer(src, soffs), n)
+ stream = global_stream(ctx, dev)
+ GC.@preserve src dest begin
+ with_ownership(stream, Managed[src.data[]]) do
+ unsafe_copyto!(ctx, dev, pointer(dest, doffs), device_pointer(src, soffs), n)
+ end
end
- end
if Base.isbitsunion(T)
# copy selector bytes
error("oneArray does not yet support isbits-union arrays")
end
# copies to the host are synchronizing
- synchronize(stream)
+ synchronize(stream)
return dest
end
function Base.unsafe_copyto!(ctx::ZeContext, dev::ZeDevice,
dest::oneDenseArray{T}, doffs, src::oneDenseArray{T}, soffs, n) where T
- GC.@preserve src dest begin
- with_ownership(global_stream(ctx, dev), Managed[dest.data[], src.data[]]) do
- unsafe_copyto!(ctx, dev, device_pointer(dest, doffs), device_pointer(src, soffs), n)
+ GC.@preserve src dest begin
+ with_ownership(global_stream(ctx, dev), Managed[dest.data[], src.data[]]) do
+ unsafe_copyto!(ctx, dev, device_pointer(dest, doffs), device_pointer(src, soffs), n)
+ end
end
- end
if Base.isbitsunion(T)
# copy selector bytes
error("oneArray does not yet support isbits-union arrays")
@@ -502,30 +503,32 @@ end
# between Array and host-accessible oneArray
function Base.unsafe_copyto!(ctx::ZeContext, dev::ZeDevice,
- dest::oneDenseArray{T,<:Any,<:Union{oneL0.SharedBuffer,oneL0.HostBuffer}}, doffs, src::Array{T}, soffs, n) where T
+ dest::oneDenseArray{T, <:Any, <:Union{oneL0.SharedBuffer, oneL0.HostBuffer}}, doffs, src::Array{T}, soffs, n
+ ) where {T}
if Base.isbitsunion(T)
# copy selector bytes
error("oneArray does not yet support isbits-union arrays")
end
- # the host pointer waits for outstanding work; hold the lock so that no new work is
- # submitted while copying
- GC.@preserve src dest with_managed_locks(Managed[dest.data[]]) do
- unsafe_copyto!(pointer(dest, doffs; type=oneL0.HostBuffer), pointer(src, soffs), n)
+ # the host pointer waits for outstanding work; hold the lock so that no new work is
+ # submitted while copying
+ GC.@preserve src dest with_managed_locks(Managed[dest.data[]]) do
+ unsafe_copyto!(pointer(dest, doffs; type = oneL0.HostBuffer), pointer(src, soffs), n)
end
return dest
end
function Base.unsafe_copyto!(ctx::ZeContext, dev::ZeDevice,
- dest::Array{T}, doffs, src::oneDenseArray{T,<:Any,<:Union{oneL0.SharedBuffer,oneL0.HostBuffer}}, soffs, n) where T
+ dest::Array{T}, doffs, src::oneDenseArray{T, <:Any, <:Union{oneL0.SharedBuffer, oneL0.HostBuffer}}, soffs, n
+ ) where {T}
if Base.isbitsunion(T)
# copy selector bytes
error("oneArray does not yet support isbits-union arrays")
end
- # the host pointer waits for outstanding work; hold the lock so that no new work is
- # submitted while copying
- GC.@preserve src dest with_managed_locks(Managed[src.data[]]) do
- unsafe_copyto!(pointer(dest, doffs), pointer(src, soffs; type=oneL0.HostBuffer), n)
+ # the host pointer waits for outstanding work; hold the lock so that no new work is
+ # submitted while copying
+ GC.@preserve src dest with_managed_locks(Managed[src.data[]]) do
+ unsafe_copyto!(pointer(dest, doffs), pointer(src, soffs; type = oneL0.HostBuffer), n)
end
return dest
@@ -623,7 +626,7 @@ function Base.unsafe_wrap(::Type{Array},
arr::oneArray{T,N,<:Union{oneL0.SharedBuffer,oneL0.HostBuffer}}) where {T,N}
# TODO: can we make this more convenient by increasing the buffer's refcount and using
# a finalizer on the Array? does that work when taking views etc of the Array?
- ptr = pointer(arr; type=oneL0.HostBuffer)
+ ptr = pointer(arr; type = oneL0.HostBuffer)
unsafe_wrap(Array, ptr, size(arr))
end
@@ -651,8 +654,12 @@ function Base.resize!(a::oneVector{T}, n::Integer) where {T}
# as a result, we can safely support resizing unowned buffers.
ctx = context(a)
dev = device(a)
- new_data = DataRef(release, alloc_managed(buftype(a), ctx, dev, bufsize,
- Base.datatype_alignment(T)))
+ new_data = DataRef(
+ release, alloc_managed(
+ buftype(a), ctx, dev, bufsize,
+ Base.datatype_alignment(T)
+ )
+ )
m = min(length(a), n)
if m > 0
with_ownership(global_stream(ctx, dev), Managed[new_data[], a.data[]]) do
diff --git a/src/compiler/execution.jl b/src/compiler/execution.jl
index b5b3a18..33211f4 100644
--- a/src/compiler/execution.jl
+++ b/src/compiler/execution.jl
@@ -154,7 +154,7 @@ end
# known yet. instead, the allocations are collected, if requested, so that the launch can
# take ownership of them right before submitting the kernel.
struct KernelAdaptor
- refs::Union{Nothing, Vector{DataRef}}
+ refs::Union{Nothing, Vector{DataRef}}
end
KernelAdaptor() = KernelAdaptor(nothing)
@@ -162,10 +162,10 @@ KernelAdaptor() = KernelAdaptor(nothing)
Adapt.adapt_storage(to::KernelAdaptor, p::ZePtr{T}) where {T} = reinterpret(Ptr{T}, p)
# convert oneAPI host arrays to device arrays
-function Adapt.adapt_storage(to::KernelAdaptor, xs::oneArray{T,N}) where {T,N}
- to.refs === nothing || push!(to.refs, xs.data)
- ptr = reinterpret(LLVMPtr{T,AS.CrossWorkgroup}, device_pointer(xs))
- oneDeviceArray{T,N,AS.CrossWorkgroup}(size(xs), ptr, xs.maxsize - xs.offset)
+function Adapt.adapt_storage(to::KernelAdaptor, xs::oneArray{T, N}) where {T, N}
+ to.refs === nothing || push!(to.refs, xs.data)
+ ptr = reinterpret(LLVMPtr{T, AS.CrossWorkgroup}, device_pointer(xs))
+ return oneDeviceArray{T, N, AS.CrossWorkgroup}(size(xs), ptr, xs.maxsize - xs.offset)
end
# Base.RefValue isn't GPU compatible, so provide a compatible alternative.
@@ -206,7 +206,7 @@ input object `x` as-is.
Do not add methods to this function, but instead extend the underlying Adapt.jl package and
register methods for the the `oneAPI.KernelAdaptor` type.
"""
-kernel_convert(arg, refs=nothing) = adapt(KernelAdaptor(refs), arg)
+kernel_convert(arg, refs = nothing) = adapt(KernelAdaptor(refs), arg)
## abstract kernel functionality
@@ -255,10 +255,10 @@ struct HostKernel{F,TT} <: AbstractKernel{F,TT}
# by every callable that converts to `f`.
source::Any
end
-HostKernel{F,TT}(f, fun::ZeKernel) where {F,TT} = HostKernel{F,TT}(f, fun, nothing)
+HostKernel{F, TT}(f, fun::ZeKernel) where {F, TT} = HostKernel{F, TT}(f, fun, nothing)
-with_source(kernel::HostKernel{F,TT}, source) where {F,TT} =
- HostKernel{F,TT}(kernel.f, kernel.fun, source)
+with_source(kernel::HostKernel{F, TT}, source) where {F, TT} =
+ HostKernel{F, TT}(kernel.f, kernel.fun, source)
# Upper bound on the spill (scratch) memory a single work-group may require, in bytes.
#
@@ -374,8 +374,9 @@ end
end
@inline function onecall(kernel::ZeKernel, tt, args::Tuple; groups::ZeDim=1, items::ZeDim=1,
- queue::Union{oneStream, ZeCommandQueue}=global_stream(context(), device()),
- roots=nothing)
+ queue::Union{oneStream, ZeCommandQueue} = global_stream(context(), device()),
+ roots = nothing
+ )
Base.@lock kernel begin
# unrolled, as iterating a heterogeneous tuple would dispatch at run time
ntuple(Val(length(args))) do i
@@ -389,13 +390,14 @@ end
# kept out of `onecall`, which is compiled for every kernel signature, as none of this
# depends on it
@noinline function launch_kernel(kernel::ZeKernel, groups::ZeDim, items::ZeDim,
- queue::Union{oneStream, ZeCommandQueue}, roots)
+ queue::Union{oneStream, ZeCommandQueue}, roots
+ )
groupsize!(kernel, items)
- launch!(queue, kernel, groups, roots)
+ return launch!(queue, kernel, groups, roots)
end
# `roots` are the references to the memory the kernel uses, which a batch keeps alive
-@inline function launch!(s::oneStream, kernel::ZeKernel, groups::ZeDim, roots=nothing)
+@inline function launch!(s::oneStream, kernel::ZeKernel, groups::ZeDim, roots = nothing)
mkl_wait!(s)
# NEO allocates a stream's scratch buffer at the first submission of a kernel whose
@@ -430,7 +432,7 @@ end
# explicit-queue compatibility: `@oneapi queue=...` submits through a per-dispatch
# command list, as it always has
-@inline function launch!(queue::ZeCommandQueue, kernel::ZeKernel, groups::ZeDim, roots=nothing)
+@inline function launch!(queue::ZeCommandQueue, kernel::ZeKernel, groups::ZeDim, roots = nothing)
spill = oneL0.spill_mem_size(kernel)
spill > queue.scratch_hwm && scratch_hedge!(queue, spill)
execute!(queue) do list
@@ -472,15 +474,17 @@ end
# form, which the kernel holds, only contains pointers. ownership is taken with the
# allocations locked until the kernel has been submitted, so that other tasks never observe
# it without the work it describes.
-@inline function launch_managed(kernel::HostKernel, source, args::Tuple;
- queue::Union{oneStream, ZeCommandQueue}=global_stream(context(), device()),
- kwargs...)
+@inline function launch_managed(
+ kernel::HostKernel, source, args::Tuple;
+ queue::Union{oneStream, ZeCommandQueue} = global_stream(context(), device()),
+ kwargs...
+ )
refs = DataRef[]
source === nothing || kernel_convert(source, refs)
converted = map(arg -> kernel_convert(arg, refs), args)
managed = Managed[ref[] for ref in refs]
ret = GC.@preserve source args with_ownership(queue, managed) do
- call(kernel, converted; queue, roots=refs, kwargs...)
+ call(kernel, converted; queue, roots = refs, kwargs...)
end
# memory that batches kept alive and turned out to be unused, also while waiting for
# another stream above. this is the first point without locks held.
diff --git a/src/context.jl b/src/context.jl
index abbf3a4..33f81f4 100644
--- a/src/context.jl
+++ b/src/context.jl
@@ -230,18 +230,22 @@ end
const batching_support = Dict{Tuple{ZeContext, ZeDevice}, Bool}()
const batching_support_lock = ReentrantLock()
function batching_supported(ctx::ZeContext, dev::ZeDevice)
- @lock batching_support_lock get!(batching_support, (ctx, dev)) do
+ return @lock batching_support_lock get!(batching_support, (ctx, dev)) do
try
list = oneL0.ZeCommandList(ctx, dev; flags = oneL0.ZE_COMMAND_LIST_FLAG_IN_ORDER)
oneL0.append_barrier!(list)
close(list)
- immediate = oneL0.ZeImmediateCommandList(ctx, dev;
+ immediate = oneL0.ZeImmediateCommandList(
+ ctx, dev;
flags = oneL0.ZE_COMMAND_QUEUE_FLAG_IN_ORDER,
- mode = oneL0.ZE_COMMAND_QUEUE_MODE_ASYNCHRONOUS)
+ mode = oneL0.ZE_COMMAND_QUEUE_MODE_ASYNCHRONOUS
+ )
GC.@preserve list begin
- oneL0.zeCommandListImmediateAppendCommandListsExp(immediate, 1,
- Ref(list.handle),
- C_NULL, 0, C_NULL)
+ oneL0.zeCommandListImmediateAppendCommandListsExp(
+ immediate, 1,
+ Ref(list.handle),
+ C_NULL, 0, C_NULL
+ )
oneL0.synchronize(immediate)
end
true
@@ -312,9 +316,11 @@ function create_stream(ctx::ZeContext, dev::ZeDevice,
"with $(err.code)")
end
batching = BATCHING[] && batching_supported(ctx, dev)
- s = oneStream(ctx, dev, list, ReentrantLock(),
- batching, nothing, Batch[], Batch[], 0, nothing, true, nothing, DataRef[],
- Batch[], nothing, nothing, false, 0, priority)
+ s = oneStream(
+ ctx, dev, list, ReentrantLock(),
+ batching, nothing, Batch[], Batch[], 0, nothing, true, nothing, DataRef[],
+ Batch[], nothing, nothing, false, 0, priority
+ )
return register_stream!(ctx, dev, s)
end
@@ -324,14 +330,18 @@ check_error(s::oneStream) = (s.error === nothing || throw(s.error); nothing)
function new_batch!(s::oneStream)
if s.events === nothing
- s.events = oneL0.ZeEventPool(s.ctx, MAX_BATCHES, s.dev;
- flags = oneL0.ZE_EVENT_POOL_FLAG_HOST_VISIBLE)
+ s.events = oneL0.ZeEventPool(
+ s.ctx, MAX_BATCHES, s.dev;
+ flags = oneL0.ZE_EVENT_POOL_FLAG_HOST_VISIBLE
+ )
end
s.nbatches += 1
list = oneL0.ZeCommandList(s.ctx, s.dev; flags = oneL0.ZE_COMMAND_LIST_FLAG_IN_ORDER)
- event = oneL0.ZeEvent(s.events, s.nbatches;
- signal = oneL0.ZE_EVENT_SCOPE_FLAG_HOST,
- wait = oneL0.ZE_EVENT_SCOPE_FLAG_HOST)
+ event = oneL0.ZeEvent(
+ s.events, s.nbatches;
+ signal = oneL0.ZE_EVENT_SCOPE_FLAG_HOST,
+ wait = oneL0.ZE_EVENT_SCOPE_FLAG_HOST
+ )
return Batch(list, event, DataRef[], Base.IdSet{Any}(), 0)
end
@@ -374,8 +384,10 @@ function flush_batch!(s::oneStream)
s.batch = nothing
try
close(b.list)
- oneL0.zeCommandListImmediateAppendCommandListsExp(s.immediate, 1, Ref(b.list.handle),
- b.event, 0, C_NULL)
+ oneL0.zeCommandListImmediateAppendCommandListsExp(
+ s.immediate, 1, Ref(b.list.handle),
+ b.event, 0, C_NULL
+ )
catch err
# the batch may have been submitted, so keep it and the memory it uses alive
push!(s.failed, b)
@@ -466,7 +478,7 @@ end
# failed to submit, which may still execute, are kept around.
function maybe_unregister_batching!(s::oneStream)
if s.recorder !== nothing && s.batch === nothing && isempty(s.submitted) &&
- isempty(s.failed)
+ isempty(s.failed)
@lock batching_streams_lock delete!(batching_streams, s)
s.recorder = nothing
end
@@ -523,8 +535,10 @@ end
function stream_queue(s::oneStream)
q = Base.@atomic s.queue
q === nothing || return q
- q = ZeCommandQueue(s.ctx, s.dev; flags = oneL0.ZE_COMMAND_QUEUE_FLAG_IN_ORDER,
- priority = s.priority)
+ q = ZeCommandQueue(
+ s.ctx, s.dev; flags = oneL0.ZE_COMMAND_QUEUE_FLAG_IN_ORDER,
+ priority = s.priority
+ )
Base.@atomic s.queue = q
return q
end
@@ -667,7 +681,7 @@ function oneL0.synchronize(s::oneStream; blocking::Bool=false)
end
# synchronize a stream, returning the error of a failed submission
-function synchronize_stream!(s::oneStream; blocking::Bool=false)
+function synchronize_stream!(s::oneStream; blocking::Bool = false)
sync = blocking ? oneL0.synchronize : oneL0.nonblocking_synchronize
# a failed submission is recorded in `s.error`, and reported below
@lock s.lock try
@@ -696,7 +710,7 @@ function oneL0.synchronize(; blocking::Bool=false)
dev = device()
oneL0.synchronize(global_stream(ctx, dev); blocking)
adopt_orphans!(ctx, dev)
- release_pending!()
+ return release_pending!()
end
# Julia → MKL ordering: everything Julia appended to the task's immediate list must be
@@ -739,7 +753,7 @@ outstanding oneMKL work. Appends are submitted to the device as they happen.
"""
@inline function oneL0.execute!(f::Base.Callable, s::oneStream)
mkl_wait!(s)
- @lock s.lock begin
+ return @lock s.lock begin
check_error(s)
flush_batch!(s)
oneL0.execute!(f, s.immediate)
diff --git a/src/memory.jl b/src/memory.jl
index 4afeabd..398d62d 100644
--- a/src/memory.jl
+++ b/src/memory.jl
@@ -65,7 +65,7 @@ end
# An allocation together with the submission target that last used it, so that host
# accesses only wait when there may be outstanding work, and work from another task first
# waits for the previous user. Shared by all arrays derived from the same allocation.
-mutable struct Managed{B<:oneL0.AbstractBuffer}
+mutable struct Managed{B <: oneL0.AbstractBuffer}
const mem::B
const lock::ReentrantLock
@@ -76,7 +76,7 @@ mutable struct Managed{B<:oneL0.AbstractBuffer}
# accesses can check it without taking the lock.
Base.@atomic dirty::Bool
- Managed(mem::B, owner=nothing, dirty::Bool=false) where {B} =
+ Managed(mem::B, owner = nothing, dirty::Bool = false) where {B} =
new{B}(mem, ReentrantLock(), owner, dirty)
end
@@ -125,13 +125,13 @@ end
# wait for outstanding work before accessing the memory from the host
function maybe_synchronize(managed::Managed)
Base.@atomic(managed.dirty) || return
- @lock managed.lock synchronize_owner!(managed)
+ return @lock managed.lock synchronize_owner!(managed)
end
# lock the given allocations in a stable order, avoiding deadlocks between tasks that
# submit operations involving the same allocations
function with_managed_locks(f, managed::Vector{Managed})
- locked = sort!(unique!(managed); by=objectid)
+ locked = sort!(unique!(managed); by = objectid)
nlocked = 0
ret = try
for m in locked
@@ -149,7 +149,7 @@ end
# submit an operation using the memory of `managed` to `owner`, as a single transaction
function with_ownership(f, owner, managed::Vector{Managed})
- with_managed_locks(managed) do
+ return with_managed_locks(managed) do
for m in managed
take_ownership!(m, owner)
end
diff --git a/src/pool.jl b/src/pool.jl
index 0177392..3dc1b1b 100644
--- a/src/pool.jl
+++ b/src/pool.jl
@@ -124,7 +124,7 @@ function release(managed::Managed)
if mem isa oneL0.DeviceBuffer && 0 < sizeof(mem) <= MAX_CHUNK
pool_free(managed) && return
end
- release(mem)
+ return release(mem)
end
@@ -171,8 +171,10 @@ struct DevicePool
slabs::Dict{UInt, Slab}
end
const NCLASSES = trailing_zeros(MAX_CHUNK) - trailing_zeros(MIN_CHUNK) + 1
-DevicePool() = DevicePool([Slab[] for _ in 1:NCLASSES],
- [FreeChunk[] for _ in 1:NCLASSES], Dict{UInt, Slab}())
+DevicePool() = DevicePool(
+ [Slab[] for _ in 1:NCLASSES],
+ [FreeChunk[] for _ in 1:NCLASSES], Dict{UInt, Slab}()
+)
const pools = Dict{Tuple{ZeContext, ZeDevice}, DevicePool}()
@@ -183,7 +185,7 @@ const pool_lock = Threads.SpinLock()
function with_pool_lock(f)
GC.enable_finalizers(false)
lock(pool_lock)
- try
+ return try
f()
finally
unlock(pool_lock)
@@ -261,7 +263,7 @@ function pool_free(managed::Managed{oneL0.DeviceBuffer})
ptr = pointer(mem)
owner = managed.owner
dirty = Base.@atomic managed.dirty
- with_pool_lock() do
+ return with_pool_lock() do
pool = get(pools, (oneL0.context(mem), oneL0.device(mem)), nothing)
pool === nothing && return false
slab = get(pool.slabs, UInt(ptr) & ~UInt(SLAB_SIZE - 1), nothing)
diff --git a/test/array.jl b/test/array.jl
index 22bb29d..6cbefe2 100644
--- a/test/array.jl
+++ b/test/array.jl
@@ -160,142 +160,144 @@ end
# on the memory, also when it was submitted by another task. The kernel takes long enough
# for a missing wait to be observed.
function slow_increment(x, n)
- for _ in 1:n
- x = x * 1.0000001f0 + 1f0
- end
- return x
+ for _ in 1:n
+ x = x * 1.0000001f0 + 1.0f0
+ end
+ return x
end
function slow_increment!(a, n)
- i = get_global_id()
- @inbounds a[i] = slow_increment(a[i], n)
- return
+ i = get_global_id()
+ @inbounds a[i] = slow_increment(a[i], n)
+ return
end
@testset "host access to $B arrays waits for GPU work" for B in (oneL0.SharedBuffer, oneL0.HostBuffer)
- n = 1_000_000
- launch(a) = @oneapi items=256 groups=cld(length(a), 256) slow_increment!(a, n)
- once = slow_increment(0f0, n)
- twice = slow_increment(once, n)
-
- # broadcast, then scalar indexing
- a = oneArray{Float32, 1, B}(undef, 1024)
- a .= 42
- @test a[1] == a[end] == 42
-
- # a kernel submitted by another task
- a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
- wait(@async launch(a))
- @test a[1] ≈ a[end] ≈ once
-
- # wrapping the memory in an Array
- a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
- launch(a)
- A = unsafe_wrap(Array, a)
- @test A[1] ≈ A[end] ≈ once
-
- # copying to an Array, which happens on the CPU
- a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
- wait(@async launch(a))
- A = Array(a)
- @test A[1] ≈ A[end] ≈ once
-
- # work from another task has to wait for this task's work on the same memory
- a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
- launch(a)
- wait(@async launch(a))
- @test a[1] ≈ a[end] ≈ twice
-
- # a kernel compiled without launching, accessing memory it captures
- let a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
- f = () -> (slow_increment!(a, n); return)
- kernel = @oneapi launch=false f()
- kernel(; items=256, groups=cld(length(a), 256))
+ n = 1_000_000
+ launch(a) = @oneapi items = 256 groups = cld(length(a), 256) slow_increment!(a, n)
+ once = slow_increment(0.0f0, n)
+ twice = slow_increment(once, n)
+
+ # broadcast, then scalar indexing
+ a = oneArray{Float32, 1, B}(undef, 1024)
+ a .= 42
+ @test a[1] == a[end] == 42
+
+ # a kernel submitted by another task
+ a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
+ wait(@async launch(a))
@test a[1] ≈ a[end] ≈ once
- end
- # a kernel submitted to an explicitly created queue
- queue = oneL0.ZeCommandQueue(context(), device(); mode=oneL0.ZE_COMMAND_QUEUE_MODE_ASYNCHRONOUS)
- a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
- @oneapi items=256 groups=cld(length(a), 256) queue=queue slow_increment!(a, n)
- @test a[1] ≈ a[end] ≈ once
+ # wrapping the memory in an Array
+ a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
+ launch(a)
+ A = unsafe_wrap(Array, a)
+ @test A[1] ≈ A[end] ≈ once
+
+ # copying to an Array, which happens on the CPU
+ a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
+ wait(@async launch(a))
+ A = Array(a)
+ @test A[1] ≈ A[end] ≈ once
+
+ # work from another task has to wait for this task's work on the same memory
+ a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
+ launch(a)
+ wait(@async launch(a))
+ @test a[1] ≈ a[end] ≈ twice
+
+ # a kernel compiled without launching, accessing memory it captures
+ let a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
+ f = () -> (slow_increment!(a, n); return)
+ kernel = @oneapi launch = false f()
+ kernel(; items = 256, groups = cld(length(a), 256))
+ @test a[1] ≈ a[end] ≈ once
+ end
+
+ # a kernel submitted to an explicitly created queue
+ queue = oneL0.ZeCommandQueue(context(), device(); mode = oneL0.ZE_COMMAND_QUEUE_MODE_ASYNCHRONOUS)
+ a = oneArray{Float32, 1, B}(zeros(Float32, 1024))
+ @oneapi items = 256 groups = cld(length(a), 256) queue = queue slow_increment!(a, n)
+ @test a[1] ≈ a[end] ≈ once
end
@testset "sub-allocation" begin
- n = 1_000_000
- once = slow_increment(0f0, n)
- base(x) = pointer(oneL0.lookup_alloc(context(), pointer(x)))
-
- if oneAPI.POOL[]
- # small arrays share driver allocations, which every kernel launch would otherwise make
- # resident one by one
- a = oneArray{Float32}(undef, 16)
- b = oneArray{Float32}(undef, 16)
- @test base(a) == base(b)
-
- c = oneArray{Float32}(undef, 1 << 20)
- @test base(c) == pointer(c)
- end
+ n = 1_000_000
+ once = slow_increment(0.0f0, n)
+ base(x) = pointer(oneL0.lookup_alloc(context(), pointer(x)))
+
+ if oneAPI.POOL[]
+ # small arrays share driver allocations, which every kernel launch would otherwise make
+ # resident one by one
+ a = oneArray{Float32}(undef, 16)
+ b = oneArray{Float32}(undef, 16)
+ @test base(a) == base(b)
+
+ c = oneArray{Float32}(undef, 1 << 20)
+ @test base(c) == pointer(c)
+ end
- # memory reused on the same stream is only written after earlier work on it
- a = oneArray(zeros(Float32, 1024))
- @oneapi items=256 groups=4 slow_increment!(a, n)
- ptr = pointer(a)
- oneAPI.unsafe_free!(a)
- b = oneArray{Float32}(undef, 1024)
- oneAPI.POOL[] && @test pointer(b) == ptr
- copyto!(b, ones(Float32, 1024))
- @test Array(b) == ones(Float32, 1024)
-
- # and from another task, only after waiting for it. that task only gets the memory if
- # there are no other chunks of that size, so use them up.
- n_chunk = oneAPI.MAX_CHUNK ÷ sizeof(Float32)
- a = oneArray(zeros(Float32, n_chunk))
- @oneapi items=256 groups=4 slow_increment!(a, n)
- others = []
- if oneAPI.POOL[]
- class = oneAPI.chunk_class(oneAPI.MAX_CHUNK)
- available() = oneAPI.with_pool_lock() do
- pool = oneAPI.pools[(context(), device())]
- !isempty(pool.fresh[class]) || !isempty(pool.free[class])
+ # memory reused on the same stream is only written after earlier work on it
+ a = oneArray(zeros(Float32, 1024))
+ @oneapi items = 256 groups = 4 slow_increment!(a, n)
+ ptr = pointer(a)
+ oneAPI.unsafe_free!(a)
+ b = oneArray{Float32}(undef, 1024)
+ oneAPI.POOL[] && @test pointer(b) == ptr
+ copyto!(b, ones(Float32, 1024))
+ @test Array(b) == ones(Float32, 1024)
+
+ # and from another task, only after waiting for it. that task only gets the memory if
+ # there are no other chunks of that size, so use them up.
+ n_chunk = oneAPI.MAX_CHUNK ÷ sizeof(Float32)
+ a = oneArray(zeros(Float32, n_chunk))
+ @oneapi items = 256 groups = 4 slow_increment!(a, n)
+ others = []
+ if oneAPI.POOL[]
+ class = oneAPI.chunk_class(oneAPI.MAX_CHUNK)
+ available() = oneAPI.with_pool_lock() do
+ pool = oneAPI.pools[(context(), device())]
+ !isempty(pool.fresh[class]) || !isempty(pool.free[class])
+ end
+ while available()
+ push!(others, oneArray{Float32}(undef, n_chunk))
+ end
end
- while available()
- push!(others, oneArray{Float32}(undef, n_chunk))
+ ptr = oneAPI.device_pointer(a)
+ oneAPI.unsafe_free!(a)
+ stream = oneAPI.global_stream(context(), device())
+ inherited, b = fetch(
+ @async begin
+ b = oneArray{Float32}(undef, n_chunk)
+ # the new array is owned by the stream that used the memory, so that using it here
+ # waits for the kernel first (which this GPU might do anyway, executing work in order)
+ inherited = b.data[].owner === stream
+ copyto!(b, ones(Float32, n_chunk))
+ inherited, b
+ end
+ )
+ if oneAPI.POOL[]
+ @test oneAPI.device_pointer(b) == ptr
+ @test inherited
end
- end
- ptr = oneAPI.device_pointer(a)
- oneAPI.unsafe_free!(a)
- stream = oneAPI.global_stream(context(), device())
- inherited, b = fetch(@async begin
- b = oneArray{Float32}(undef, n_chunk)
- # the new array is owned by the stream that used the memory, so that using it here
- # waits for the kernel first (which this GPU might do anyway, executing work in order)
- inherited = b.data[].owner === stream
- copyto!(b, ones(Float32, n_chunk))
- inherited, b
- end)
- if oneAPI.POOL[]
- @test oneAPI.device_pointer(b) == ptr
- @test inherited
- end
- synchronize()
- @test Array(b) == ones(Float32, n_chunk)
+ synchronize()
+ @test Array(b) == ones(Float32, n_chunk)
- # many small arrays keep their contents, also when unused memory is given back
- xs = [oneArray(fill(Float32(i), i)) for i in 1:300]
- for i in 1:2:300
- oneAPI.unsafe_free!(xs[i])
- end
- if oneAPI.POOL[]
- before = oneAPI._allocated_bytes[]
- ys = [oneArray{Float32}(undef, 1 << 16) for _ in 1:8] # a slab of their own
- foreach(oneAPI.unsafe_free!, ys)
- oneAPI.trim_pools!() # (which also frees other empty slabs)
- @test oneAPI._allocated_bytes[] <= before
- end
- @test all(i -> Array(xs[i]) == fill(Float32(i), i), 2:2:300)
- for i in 2:2:300
- xs[i] = oneArray(fill(Float32(i), i))
- end
- @test all(i -> Array(xs[i]) == fill(Float32(i), i), 2:2:300)
+ # many small arrays keep their contents, also when unused memory is given back
+ xs = [oneArray(fill(Float32(i), i)) for i in 1:300]
+ for i in 1:2:300
+ oneAPI.unsafe_free!(xs[i])
+ end
+ if oneAPI.POOL[]
+ before = oneAPI._allocated_bytes[]
+ ys = [oneArray{Float32}(undef, 1 << 16) for _ in 1:8] # a slab of their own
+ foreach(oneAPI.unsafe_free!, ys)
+ oneAPI.trim_pools!() # (which also frees other empty slabs)
+ @test oneAPI._allocated_bytes[] <= before
+ end
+ @test all(i -> Array(xs[i]) == fill(Float32(i), i), 2:2:300)
+ for i in 2:2:300
+ xs[i] = oneArray(fill(Float32(i), i))
+ end
+ @test all(i -> Array(xs[i]) == fill(Float32(i), i), 2:2:300)
end
# https://github.com/JuliaGPU/oneAPI.jl/issues/661: kernels have to declare indirect access,
@@ -357,14 +359,14 @@ end
resize!(c, 0)
oneAPI.unsafe_free!(c)
- # (with buffers big enough not to be sub-allocated, which `_allocated_bytes` doesn't track)
+ # (with buffers big enough not to be sub-allocated, which `_allocated_bytes` doesn't track)
GC.enable(false)
try
- d = oneArray{Float32}(undef, 1 << 19)
+ d = oneArray{Float32}(undef, 1 << 19)
before = oneAPI._allocated_bytes[]
resize!(d, 1 << 20)
oneAPI.unsafe_free!(d)
- @test oneAPI._allocated_bytes[] == before - sizeof(Float32) * (1 << 19)
+ @test oneAPI._allocated_bytes[] == before - sizeof(Float32) * (1 << 19)
finally
GC.enable(true)
end
diff --git a/test/execution.jl b/test/execution.jl
index 28a674b..d629aef 100644
--- a/test/execution.jl
+++ b/test/execution.jl
@@ -869,7 +869,7 @@ end
@testset "command batching" begin
n = 100 # spans several batches
batching = oneAPI.global_stream(context(), device()).batching &&
- !oneL0.sync_each_submission()
+ !oneL0.sync_each_submission()
# dependent launches execute in order, across batches
a = oneArray(Int64[0])
@@ -896,30 +896,36 @@ end
# also when they were recorded by another task
a = oneArray(Int64[0])
- wait(@async begin
- for i in 1:n
- @oneapi recurrence_kernel(a, i)
+ wait(
+ @async begin
+ for i in 1:n
+ @oneapi recurrence_kernel(a, i)
+ end
end
- end)
+ )
@test Array(a)[1] == recurrence(n)
# synchronizing submits what tasks recorded before finishing, also when the memory it
# uses isn't tracked, and when another task records on the same stream later
a = oneArray(Int64[0])
ptr = reinterpret(Core.LLVMPtr{Int64, oneAPI.AS.CrossWorkgroup}, pointer(a))
- stream = fetch(@async begin
- for _ in 1:3
- @oneapi ptr_increment_kernel(ptr)
+ stream = fetch(
+ @async begin
+ for _ in 1:3
+ @oneapi ptr_increment_kernel(ptr)
+ end
+ oneAPI.global_stream(context(), device())
end
- oneAPI.global_stream(context(), device())
- end)
+ )
synchronize()
@test Array(a)[1] == 3
- wait(@async begin
- for _ in 1:3
- @oneapi queue=stream ptr_increment_kernel(ptr)
+ wait(
+ @async begin
+ for _ in 1:3
+ @oneapi queue = stream ptr_increment_kernel(ptr)
+ end
end
- end)
+ )
synchronize()
@test Array(a)[1] == 6
@@ -936,11 +942,13 @@ end
end
stream = take!(streams)
batching && @test stream.recorder === owner
- wait(@async begin
- for _ in 1:3
- @oneapi queue=stream ptr_increment_kernel(ptr)
+ wait(
+ @async begin
+ for _ in 1:3
+ @oneapi queue = stream ptr_increment_kernel(ptr)
+ end
end
- end)
+ )
synchronize()
@test Array(a)[1] == 11
put!(done, nothing)
@@ -954,7 +962,7 @@ end
bytes = 1 << 23 # not sub-allocated
before = oneAPI._allocated_bytes[]
b = oneArray{Int64}(undef, bytes ÷ sizeof(Int64))
- @oneapi items=256 groups=4 fill_kernel(b, 1)
+ @oneapi items = 256 groups = 4 fill_kernel(b, 1)
oneAPI.unsafe_free!(b)
batching && @test oneAPI._allocated_bytes[] == before + bytes
synchronize()
@@ -978,7 +986,7 @@ end
# with all batches executing, launches are submitted directly instead of waiting for
# one, which may not finish before this task does something (here, open the gate)
if batching
- pool = oneL0.ZeEventPool(context(), 1, device(); flags=oneL0.ZE_EVENT_POOL_FLAG_HOST_VISIBLE)
+ pool = oneL0.ZeEventPool(context(), 1, device(); flags = oneL0.ZE_EVENT_POOL_FLAG_HOST_VISIBLE)
gate = pool[1]
a = oneArray(Int64[0])
# (only a few more: many submissions waiting for the gate make the driver stall, and
@@ -1005,43 +1013,61 @@ end
for _ in 1:40
x .+= 1
end
- LinearAlgebra.axpy!(2f0, x, y)
+ LinearAlgebra.axpy!(2.0f0, x, y)
for _ in 1:40
y .+= 1
end
- @test Array(y) == fill(2f0 * 41 + 40, 1024)
+ @test Array(y) == fill(2.0f0 * 41 + 40, 1024)
# a failed submission is reported until the stream is synchronized
- @test fetch(@async begin
- a = oneArray(Int64[0])
- stream = oneAPI.global_stream(context(), device())
- err = ErrorException("submission failed")
- stream.error = err
- caught(f) = try f(); nothing catch e; e end
- results = [caught(() -> @oneapi recurrence_kernel(a, 1)),
- caught(() -> copyto!(a, [1])),
- caught(() -> oneAPI.flush!(stream)),
- caught(() -> oneAPI.drain(stream)),
- caught(() -> synchronize())]
- # also when crossing a new spill tier, which synchronizes first
- stream.error = err
- stream.scratch_hwm = -1
- push!(results, caught(() -> @oneapi recurrence_kernel(a, 1)))
- stream.error = nothing
- all(r -> r === err, results) && caught(() -> synchronize()) === nothing
- end)
+ @test fetch(
+ @async begin
+ a = oneArray(Int64[0])
+ stream = oneAPI.global_stream(context(), device())
+ err = ErrorException("submission failed")
+ stream.error = err
+ caught(f) = try
+ f(); nothing
+ catch e
+ e
+ end
+ results = [
+ caught(() -> @oneapi recurrence_kernel(a, 1)),
+ caught(() -> copyto!(a, [1])),
+ caught(() -> oneAPI.flush!(stream)),
+ caught(() -> oneAPI.drain(stream)),
+ caught(() -> synchronize()),
+ ]
+ # also when crossing a new spill tier, which synchronizes first
+ stream.error = err
+ stream.scratch_hwm = -1
+ push!(results, caught(() -> @oneapi recurrence_kernel(a, 1)))
+ stream.error = nothing
+ all(r -> r === err, results) && caught(() -> synchronize()) === nothing
+ end
+ )
# and reported once by whoever synchronizes when the task that recorded didn't
err = ErrorException("submission failed")
a = oneArray(Int64[0])
- wait(@async begin
- for i in 1:2
- @oneapi recurrence_kernel(a, i)
+ wait(
+ @async begin
+ for i in 1:2
+ @oneapi recurrence_kernel(a, i)
+ end
+ oneAPI.global_stream(context(), device()).error = err
end
- oneAPI.global_stream(context(), device()).error = err
- end)
- @test try synchronize(); nothing catch e; e end === (batching ? err : nothing)
- @test try synchronize(); nothing catch e; e end === nothing
+ )
+ @test try
+ synchronize(); nothing
+ catch e
+ e
+ end === (batching ? err : nothing)
+ @test try
+ synchronize(); nothing
+ catch e
+ e
+ end === nothing
end
@testset "cooperative synchronize" begin |
maleadt
force-pushed
the
tb/launch-batching
branch
from
October 13, 2026 18:20
7a00cd3 to
d6e5f60
Compare
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## tb/usm-pool #693 +/- ##
===============================================
+ Coverage 81.18% 81.78% +0.59%
===============================================
Files 51 51
Lines 4119 4292 +173
===============================================
+ Hits 3344 3510 +166
- Misses 775 782 +7 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Kernel launches and `execute!` appended to the stream's immediate list directly, and other code reached into it as well. Give the stream a lock that every append holds, and rename the list so that code can't append to it without going through `launch!` or `execute!`. This prepares for batching launches, which needs all submissions to a stream to be ordered with the batch.
Every append to an immediate command list is a submission, which costs a system call: on an Iris Xe ~9 µs of host time per kernel launch, more than the rest of the launch path combined. Kernels launched on a stream are now recorded in an in-order regular command list, which is submitted to the stream's immediate list with zeCommandListImmediateAppendCommandListsExp once it holds 32 launches, or before anything else needs the stream's work: other operations on the stream, synchronization, host accesses to memory the stream uses (through `drain`), and oneMKL calls. Recording costs ~1.5 µs per launch, and a submission also pays the driver's per-allocation residency cost once per batch instead of once per launch. The first launch after synchronizing is submitted directly, so a lone kernel starts right away. `oneAPI.flush!` submits recorded launches explicitly. Each batch keeps the memory its kernels use alive until it has executed, so freeing an array can't race with a batch that hasn't been submitted yet. `synchronize()` also submits the batches of tasks that finished without doing so. A failed submission is reported right away, and again by the stream's next synchronization. Drivers that can't submit regular lists to an immediate one, which is an experimental extension, keep submitting every launch directly, as does ONEAPI_COMMAND_BATCHING=0.
maleadt
force-pushed
the
tb/launch-batching
branch
from
October 15, 2026 12:45
d6e5f60 to
a484227
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Each append to an immediate command list is a submission, and on i915 a submission is an
execbuffersystem call. On an Iris Xe that is ~9 µs of host time per kernel launch, most of the ~11 µs a launch takes; a@cudalaunch takes ~2 µs.Kernels launched on a stream are now recorded in an in-order regular command list, which is submitted to the stream's immediate list with
zeCommandListImmediateAppendCommandListsExponce it holds 32 launches, or before anything needs the stream's work: copies and other operations on the stream,synchronize, host accesses to memory used by the stream or its use by another task (through #686's ownership tracking), and oneMKL calls. Metal.jl batches the same way. The first launch after synchronizing is submitted right away, so a single kernel doesn't wait, andoneAPI.flush!submits explicitly, e.g. before host work that should overlap with the GPU. A submission also makes every live allocation resident only once per batch.Each batch keeps the memory its kernels use alive until it has executed, so freeing an array can't race with a batch that hasn't been submitted, which the driver's blocking free doesn't know about.
synchronize()also submits what tasks recorded before finishing without synchronizing. A failed submission is reported right away, and again by the stream's next synchronization. When all eight batches of a stream are executing, launches are submitted directly rather than waiting. Drivers without the experimental extension keep submitting every launch directly, as doesONEAPI_COMMAND_BATCHING=0.Launching a trivial kernel in a loop on an Iris Xe (thread CPU time per launch, median, ranges over three interleaved runs on a loaded machine):
k(a::oneVector, x::Float32)k(a::oneVector, b::oneVector)Recording a launch costs ~1.5 µs in the driver; the rest is Julia-side.
The batch size and number of batches are constants for now. Explicit
queue=::ZeCommandQueuelaunches aren't batched.The test suite passes on an Iris Xe with and without batching (and sub-allocation). The Aurora LTS stack only gets tested by CI, which synchronizes after every submission there, so launches aren't batched.