Submit work through per-task immediate command lists

docs/src/lts.md

@@ -93,9 +93,10 @@ references it. A garbage-collected free of a dead array whose last kernel has no
 then faults on the GPU, which bans the kernel context — after which *every* later submission
 fails with `ZE_RESULT_ERROR_UNKNOWN`.
 
-On LTS, oneAPI.jl keeps a registry of the command queues in use and drains those that could
-reference a buffer before freeing it. Command queues are likewise drained before being
-destroyed; a queue still busy after 10 s is deliberately leaked, since destroying it would
+On LTS, oneAPI.jl keeps a registry of the streams in use — each task's immediate command
+list plus its companion command queue for oneMKL work — and drains those that could
+reference a buffer before freeing it. Lists and queues are likewise drained before being
+destroyed; one still busy after 10 s is deliberately leaked, since destroying it would
 trigger the very fault the drain prevents. The visible cost is that a GC-driven free can
 block until outstanding work completes.
 
@@ -107,7 +108,10 @@ earlier, separately submitted command list. The result is a silent *dropped tail
 work-items of a kernel, or the last elements of a copy, never land.
 
 Synchronizing after every submission eliminates it, at roughly a 3× throughput cost. It is
-off by default and enabled with:
+off by default and enabled with the setting below. With the current immediate-command-list
+submission path it host-synchronizes the stream after every append; the dropped-tail
+failure was only ever observed on the earlier queue-submission path, so this workaround may
+no longer be needed — it is kept until that is re-established under oversubscription.
 
 ```bash
 export ONEAPI_SYNC_EACH_SUBMISSION=1

docs/src/troubleshooting.md

@@ -86,3 +86,16 @@ Enable debug mode in oneAPI.jl to use debug builds of underlying toolchains (if
 oneAPI.set_debug!(true)
 ```
 
+
+### Scratch hedge
+
+Before the first submission of a kernel whose register spill exceeds the stream's
+high-water mark, oneAPI.jl drains the stream and runs finalizers: the driver performs its
+scratch-buffer allocation at that moment, and (at least through NEO 26.x) aborts the
+process instead of erroring when it fails. Making the allocation happen at a clean moment
+removes that failure mode for workloads under memory pressure. The drain costs one
+synchronization per (stream, spill tier) — once per workload in practice. Disable it with:
+
+```bash
+export ONEAPI_SCRATCH_HEDGE=0
+```

lib/level-zero/barrier.jl

@@ -1,6 +1,6 @@
 export append_barrier!, device_barrier
 
-append_barrier!(list::ZeCommandList, signal_event=nothing, wait_events::ZeEvent...) =
+append_barrier!(list::AbstractZeCommandList, signal_event=nothing, wait_events::ZeEvent...) =
     zeCommandListAppendBarrier(list, something(signal_event, C_NULL),
                                length(wait_events), [wait_events...])
 

lib/level-zero/cmdlist.jl

@@ -1,8 +1,10 @@
 # list
 
-export ZeCommandList, execute!
+export ZeCommandList, ZeImmediateCommandList, AbstractZeCommandList, execute!
 
-mutable struct ZeCommandList
+abstract type AbstractZeCommandList end
+
+mutable struct ZeCommandList <: AbstractZeCommandList
     handle::ze_command_list_handle_t
 
     context::ZeContext
@@ -22,10 +24,10 @@ mutable struct ZeCommandList
     end
 end
 
-Base.unsafe_convert(::Type{ze_command_list_handle_t}, list::ZeCommandList) = list.handle
+Base.unsafe_convert(::Type{ze_command_list_handle_t}, list::AbstractZeCommandList) = list.handle
 
-Base.:(==)(a::ZeCommandList, b::ZeCommandList) = a.handle == b.handle
-Base.hash(e::ZeCommandList, h::UInt) = hash(e.handle, h)
+Base.:(==)(a::AbstractZeCommandList, b::AbstractZeCommandList) = a.handle == b.handle
+Base.hash(e::AbstractZeCommandList, h::UInt) = hash(e.handle, h)
 
 Base.close(list::ZeCommandList) = zeCommandListClose(list)
 
@@ -47,6 +49,62 @@ function ZeCommandList(f::Base.Callable, args...; kwargs...)
     return list
 end
 
+"""
+    ZeImmediateCommandList(ctx::ZeContext, dev::ZeDevice, ordinal=1, index=1;
+                           flags=0, mode=ZE_COMMAND_QUEUE_MODE_DEFAULT,
+                           priority=ZE_COMMAND_QUEUE_PRIORITY_NORMAL)
+
+Create an immediate command list: appended commands are submitted to the device as they
+are appended, with no separate close/execute step and no per-submission list object.
+The descriptor is a command *queue* descriptor; pass
+`flags=ZE_COMMAND_QUEUE_FLAG_IN_ORDER` and `mode=ZE_COMMAND_QUEUE_MODE_ASYNCHRONOUS`
+for an asynchronous stream with in-order semantics (requires Level Zero >= 1.9).
+Synchronize with [`synchronize`](@ref).
+"""
+mutable struct ZeImmediateCommandList <: AbstractZeCommandList
+    handle::ze_command_list_handle_t
+
+    context::ZeContext
+    device::ZeDevice
+    ordinal::Int
+
+    function ZeImmediateCommandList(ctx::ZeContext, dev::ZeDevice, ordinal=1, index=1;
+                                    flags=0,
+                                    mode::ze_command_queue_mode_t=ZE_COMMAND_QUEUE_MODE_DEFAULT,
+                                    priority::ze_command_queue_priority_t=ZE_COMMAND_QUEUE_PRIORITY_NORMAL)
+        desc_ref = Ref(ze_command_queue_desc_t(;
+            ordinal=ordinal-1, index=index-1, flags, mode, priority
+        ))
+        handle_ref = Ref{ze_command_list_handle_t}()
+        zeCommandListCreateImmediate(ctx, dev, desc_ref, handle_ref)
+        obj = new(handle_ref[], ctx, dev, ordinal)
+        finalizer(obj) do obj
+            # unlike a regular list, an immediate list can have work in flight at
+            # finalization on any stack, and destroying it then is illegal. Bounded
+            # unchecked drain, leaking the list on timeout — same rationale as the
+            # queue finalizer in cmdqueue.jl: an infinite wait on event-gated work
+            # whose event is never signaled would hang GC and process exit.
+            if unchecked_zeCommandListHostSynchronize(obj, FINALIZER_SYNC_TIMEOUT_NS) == RESULT_NOT_READY
+                @warn "Leaking an immediate command list still busy after $(FINALIZER_SYNC_TIMEOUT_NS ÷ 1_000_000_000)s to avoid blocking finalization" maxlog = 1
+                return
+            end
+            zeCommandListDestroy(obj)
+            # mark destroyed: the stream registry can still reach this list, and
+            # synchronizing a destroyed handle crashes in the driver
+            obj.handle = ze_command_list_handle_t(C_NULL)
+        end
+        obj
+    end
+end
+
+"""
+    synchronize(list::ZeImmediateCommandList, timeout=typemax(UInt64))
+
+Block the host until all commands appended to the immediate command list have completed.
+"""
+synchronize(list::ZeImmediateCommandList, timeout::Number=typemax(UInt64)) =
+    zeCommandListHostSynchronize(list, timeout)
+
 # Opt-in workaround for the Aurora LTS NEO stack (set ONEAPI_SYNC_EACH_SUBMISSION=1).
 # Under heavy multi-process oversubscription of a single tile, a whole-queue
 # `zeCommandQueueSynchronize` does not reliably retire the tail of an earlier,
@@ -112,3 +170,18 @@ function execute!(f::Base.Callable, queue::ZeCommandQueue, fence=nothing; kwargs
     list = ZeCommandList(f, queue.context, queue.device, queue.ordinal; kwargs...)
     execute!(queue, [list], fence)
 end
+
+"""
+    execute!(list::ZeImmediateCommandList) do list
+        append_...!(list)
+    end
+
+Append operations to an immediate command list. Each append is submitted to the device
+as it happens, so there is no separate close/execute step; the return value is that of
+the do block.
+"""
+function execute!(f::Base.Callable, list::ZeImmediateCommandList)
+    ret = f(list)
+    sync_each_submission() && synchronize(list)
+    return ret
+end

lib/level-zero/cmdqueue.jl

@@ -50,7 +50,7 @@ mutable struct ZeCommandQueue
             zeCommandQueueDestroy(obj)
             if LTS[]
                 # mark the queue as destroyed: it can still be weakly reachable (e.g. from
-                # the queue registry used by `synchronize_all_queues`), and synchronizing a
+                # the stream registry used by `synchronize_all_streams`), and synchronizing a
                 # destroyed handle crashes in the driver.
                 obj.handle = ze_command_queue_handle_t(C_NULL)
             end

lib/level-zero/copy.jl

@@ -2,22 +2,22 @@
 
 export append_copy!, append_fill!, append_prefetch!, append_advise!
 
-append_copy!(list::ZeCommandList, dst::Union{Ptr,ZePtr}, src::Union{Ptr,ZePtr},
+append_copy!(list::AbstractZeCommandList, dst::Union{Ptr,ZePtr}, src::Union{Ptr,ZePtr},
              size::Integer, signal_event::Union{ZeEvent,Nothing}=nothing,
              wait_events::ZeEvent...) =
     zeCommandListAppendMemoryCopy(list, dst, src, size, something(signal_event, C_NULL),
                                    length(wait_events), [wait_events...])
 
-append_fill!(list::ZeCommandList, ptr::Union{Ptr,ZePtr}, pattern::Union{Ptr,ZePtr},
+append_fill!(list::AbstractZeCommandList, ptr::Union{Ptr,ZePtr}, pattern::Union{Ptr,ZePtr},
              pattern_size::Integer, size::Integer,
              signal_event::Union{ZeEvent,Nothing}=nothing, wait_events::ZeEvent...) =
     zeCommandListAppendMemoryFill(list, ptr, pattern, pattern_size, size,
                                   something(signal_event, C_NULL), length(wait_events),
                                   [wait_events...])
 
-append_prefetch!(list::ZeCommandList, ptr::Union{Ptr,ZePtr}, size::Integer) =
+append_prefetch!(list::AbstractZeCommandList, ptr::Union{Ptr,ZePtr}, size::Integer) =
     zeCommandListAppendMemoryPrefetch(list, ptr, size)
 
-append_advise!(list::ZeCommandList, dev::ZeDevice, ptr::Union{Ptr,ZePtr}, size::Integer,
+append_advise!(list::AbstractZeCommandList, dev::ZeDevice, ptr::Union{Ptr,ZePtr}, size::Integer,
                advise::ze_memory_advice_t) =
     zeCommandListAppendMemAdvise(list, dev, ptr, size, advise)

lib/level-zero/event.jl

@@ -54,15 +54,15 @@ Base.:(==)(a::ZeEvent, b::ZeEvent) = a.handle == b.handle
 Base.hash(e::ZeEvent, h::UInt) = hash(e.handle, h)
 
 signal(event::ZeEvent) = zeEventHostSignal(event)
-append_signal!(list::ZeCommandList, event::ZeEvent) = zeCommandListAppendSignalEvent(list, event)
+append_signal!(list::AbstractZeCommandList, event::ZeEvent) = zeCommandListAppendSignalEvent(list, event)
 
 Base.wait(event::ZeEvent, timeout::Number=typemax(UInt64)) =
     zeEventHostSynchronize(event, timeout)
-append_wait!(list::ZeCommandList, events::ZeEvent...) =
+append_wait!(list::AbstractZeCommandList, events::ZeEvent...) =
     zeCommandListAppendWaitOnEvents(list, length(events), [events...])
 
 Base.reset(event::ZeEvent) = zeEventHostReset(event)
-append_reset!(list::ZeCommandList, event::ZeEvent) = zeCommandListAppendEventReset(list, event)
+append_reset!(list::AbstractZeCommandList, event::ZeEvent) = zeCommandListAppendEventReset(list, event)
 
 function Base.isdone(event::ZeEvent)
     res = unchecked_zeEventQueryStatus(event)

lib/mkl/fft.jl

@@ -523,22 +523,31 @@ end
 # Execution helpers
 _rawptr(a::oneAPI.oneArray{T}) where T = reinterpret(Ptr{Cvoid}, pointer(a))
 
+# NOTE: plans capture their SYCL queue at construction and thus bypass `sycl_queue` —
+# the usual Julia → MKL ordering boundary — at execution time, so every _exec! method
+# must run `mkl_boundary!` itself before handing work to oneMKL.
+
 function _exec!(p::cMKLFFTPlan{T,MKLFFT_FORWARD,true}, X::oneAPI.oneArray{T}) where T
+    oneAPI.mkl_boundary!()
     st = onemklDftComputeForward(p.handle, _rawptr(X)); st==0 || error("forward FFT failed ($st)"); X
 end
 function _exec!(p::cMKLFFTPlan{T,MKLFFT_INVERSE,true}, X::oneAPI.oneArray{T}) where T
+    oneAPI.mkl_boundary!()
     st = onemklDftComputeBackward(p.handle, _rawptr(X)); st==0 || error("inverse FFT failed ($st)"); X
 end
 function _exec!(p::cMKLFFTPlan{T,K,false}, X::oneAPI.oneArray{T}, Y::oneAPI.oneArray{T}) where {T,K}
+    oneAPI.mkl_boundary!()
     st = (K==MKLFFT_FORWARD ? onemklDftComputeForwardOutOfPlace : onemklDftComputeBackwardOutOfPlace)(p.handle, _rawptr(X), _rawptr(Y)); st==0 || error("FFT failed ($st)"); Y
 end
 
 # Real forward
 function _exec!(p::rMKLFFTPlan{T,MKLFFT_FORWARD,false}, X::oneAPI.oneArray{T}, Y::oneAPI.oneArray{Complex{T}}) where T
+    oneAPI.mkl_boundary!()
     st = onemklDftComputeForwardOutOfPlace(p.handle, _rawptr(X), _rawptr(Y)); st==0 || error("rfft failed ($st)"); Y
 end
 # Real inverse (complex -> real)
 function _exec!(p::rMKLFFTPlan{T,MKLFFT_INVERSE,false}, X::oneAPI.oneArray{T}, Y::oneAPI.oneArray{R}) where {R,T<:Complex{R}}
+    oneAPI.mkl_boundary!()
     st = onemklDftComputeBackwardOutOfPlace(p.handle, _rawptr(X), _rawptr(Y)); st==0 || error("brfft failed ($st)"); Y
 end
 

src/array.jl

@@ -462,7 +462,7 @@ function Base.unsafe_copyto!(ctx::ZeContext, dev::ZeDevice,
   end
 
   # copies to the host are synchronizing
-  synchronize(global_queue(context(src), device()))
+  synchronize(global_stream(context(src), device()))
 
   return dest
 end
@@ -482,7 +482,7 @@ end
 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
   # maintain queue-ordered semantics
-  synchronize(global_queue(ctx, dev))
+  synchronize(global_stream(ctx, dev))
 
   if Base.isbitsunion(T)
     # copy selector bytes
@@ -503,7 +503,7 @@ 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
   # maintain queue-ordered semantics
-  synchronize(global_queue(ctx, dev))
+  synchronize(global_stream(ctx, dev))
 
   if Base.isbitsunion(T)
     # copy selector bytes

src/compiler/execution.jl

@@ -29,7 +29,9 @@ launches the kernel on the GPU.
 ## Launch Keywords (runtime)
 - `groups`: Number of workgroups (required). Can be an integer or tuple.
 - `items`: Number of work-items per workgroup (required). Can be an integer or tuple.
-- `queue::ZeCommandQueue=global_queue(...)`: Command queue to submit to.
+- `queue=global_stream(...)`: Submission target — the task's `oneStream` by default. An
+  explicit `ZeCommandQueue` is also accepted and submits through a per-dispatch command
+  list.
 
 # Examples
 
@@ -322,38 +324,53 @@ end
 const _kernel_instances = Dict{UInt, Any}()
 
 @inline function onecall(kernel::ZeKernel, tt, args...; groups::ZeDim=1, items::ZeDim=1,
-                         queue::ZeCommandQueue=global_queue(context(), device()))
+                         queue::Union{oneStream, ZeCommandQueue}=global_stream(context(), device()))
     for (i, arg) in enumerate(args)
         oneL0.arguments(kernel)[i] = arg
     end
 
     groupsize!(kernel, items)
+    launch!(queue, kernel, groups)
+end
+
+@inline function launch!(s::oneStream, kernel::ZeKernel, groups::ZeDim)
+    mkl_wait!(s)
 
-    # NEO allocates a queue's scratch buffer at the first submission of a kernel whose
+    # NEO allocates a stream's scratch buffer at the first submission of a kernel whose
     # spill exceeds what is already allocated, and that allocation aborts the process on
     # failure (no null check). Cross each new spill high-water mark deliberately, at the
     # cleanest reachable moment, instead of at a GC-lottery-determined one.
     spill = oneL0.spill_mem_size(kernel)
-    spill > queue.scratch_hwm && scratch_hedge!(queue, spill)
+    spill > s.scratch_hwm && scratch_hedge!(s, spill)
+
+    append_launch!(s.list, kernel, groups)
+    oneL0.sync_each_submission() && oneL0.synchronize(s.list)
+    return
+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)
+    spill = oneL0.spill_mem_size(kernel)
+    spill > queue.scratch_hwm && scratch_hedge!(queue, spill)
     execute!(queue) do list
         append_launch!(list, kernel, groups)
     end
 end
 
-# Slow path of the scratch hedge, firing once per (queue, spill tier): retire in-flight
+# Slow path of the scratch hedge, firing once per (stream, spill tier): retire in-flight
 # work, flush deferred releases, and run finalizers so dead driver objects and arrays are
 # destroyed before NEO performs its null-check-free scratch allocation. Opt out with
 # ONEAPI_SCRATCH_HEDGE=0; the high-water mark is maintained regardless, so the toggle
 # only skips the drain.
-@noinline function scratch_hedge!(queue::ZeCommandQueue, spill::Int)
+@noinline function scratch_hedge!(target::Union{oneStream, ZeCommandQueue}, spill::Int)
     if oneL0.SCRATCH_HEDGE[]
-        oneL0.synchronize(queue)
+        oneL0.synchronize(target)
         oneL0._run_reclaim_callbacks()
         GC.gc(false)
         Threads.atomic_add!(oneL0.SCRATCH_HEDGE_COUNT, 1)
     end
-    queue.scratch_hwm = spill
+    target.scratch_hwm = spill
     return
 end