diff --git a/lib/level-zero/cmdlist.jl b/lib/level-zero/cmdlist.jl index 0e00f0d4..e75497b1 100644 --- a/lib/level-zero/cmdlist.jl +++ b/lib/level-zero/cmdlist.jl @@ -152,25 +152,6 @@ function sync_each_submission(f::Base.Callable, enable::Bool) end end -function execute!(queue::ZeCommandQueue, lists::Vector{ZeCommandList}, fence = nothing) - r = zeCommandQueueExecuteCommandLists(queue, length(lists), lists, something(fence, C_NULL)) - sync_each_submission() && synchronize(queue) - return r -end - -""" - execute!(queue::ZeCommandQueue, ...) do list - append_...!(list) - end - -Create a command list for the device that owns `queue`, passing in a do block that appends -operations. The list is then closed and executed on the queue. -""" -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) diff --git a/lib/level-zero/cmdqueue.jl b/lib/level-zero/cmdqueue.jl index a308255a..cd00566e 100644 --- a/lib/level-zero/cmdqueue.jl +++ b/lib/level-zero/cmdqueue.jl @@ -19,6 +19,15 @@ mutable struct ZeCommandQueue # maintained by the scratch hedge (see `scratch_hedge!` in src/compiler/execution.jl) scratch_hwm::Int + # the command lists `execute!` submitted, oldest first, each with the fence signaling + # its completion, and those that are free to reuse. creating a list per submission and + # leaving its destruction to the GC piles up driver memory the GC doesn't know about. + # lists that failed to submit may still execute, so they are kept but never reused. + const lock::ReentrantLock + const submitted::Vector{Tuple{ZeCommandList, ze_fence_handle_t}} + const spare::Vector{Tuple{ZeCommandList, ze_fence_handle_t}} + const failed::Vector{Tuple{ZeCommandList, ze_fence_handle_t}} + function ZeCommandQueue(ctx::ZeContext, dev::ZeDevice, ordinal=1, index=1; flags=0, mode::ze_command_queue_mode_t=ZE_COMMAND_QUEUE_MODE_DEFAULT, @@ -28,9 +37,11 @@ mutable struct ZeCommandQueue )) handle_ref = Ref{ze_command_queue_handle_t}() zeCommandQueueCreate(ctx, dev, desc_ref, handle_ref) - obj = new(handle_ref[], ctx, dev, ordinal, 0) + obj = new(handle_ref[], ctx, dev, ordinal, 0, ReentrantLock(), [], [], []) finalizer(obj) do obj - if LTS[] + # the fences of the lists `execute!` submitted have to be destroyed before the + # queue, and not while in use: drain the queue first if any may still execute. + if LTS[] || !isempty(obj.submitted) || !isempty(obj.failed) # the queue may still have work in flight (nothing requires a task to # synchronize before dying), and zeCommandQueueDestroy does not wait for # it: on the LTS NEO stack the still-running work then faults as soon as @@ -47,6 +58,9 @@ mutable struct ZeCommandQueue return end end + for (_, fence) in Iterators.flatten((obj.submitted, obj.spare, obj.failed)) + zeFenceDestroy(fence) + end zeCommandQueueDestroy(obj) if LTS[] # mark the queue as destroyed: it can still be weakly reachable (e.g. from @@ -67,6 +81,69 @@ Base.hash(e::ZeCommandQueue, h::UInt) = hash(e.handle, h) synchronize(queue::ZeCommandQueue, timeout::Number=typemax(UInt64)) = zeCommandQueueSynchronize(queue, timeout) +function execute!(queue::ZeCommandQueue, lists::Vector{ZeCommandList}, fence = nothing) + r = zeCommandQueueExecuteCommandLists(queue, length(lists), lists, something(fence, C_NULL)) + sync_each_submission() && synchronize(queue) + return r +end + +""" + execute!(queue::ZeCommandQueue, ...) do list + append_...!(list) + end + +Pass a do block that appends operations to a command list for the device that owns +`queue`. The list is then closed and executed on the queue. Lists are reused once they have +executed, unless a `fence` or keyword arguments for the list are passed. +""" +function execute!(f::Base.Callable, queue::ZeCommandQueue, fence=nothing; kwargs...) + if fence !== nothing || !isempty(kwargs) + list = ZeCommandList(f, queue.context, queue.device, queue.ordinal; kwargs...) + return execute!(queue, [list], fence) + end + + list, fence = take_list!(queue) + try + f(list) + close(list) + catch + @lock queue.lock push!(queue.spare, (list, fence)) + rethrow() + end + r = try + zeCommandQueueExecuteCommandLists(queue, 1, [list], fence) + catch + @lock queue.lock push!(queue.failed, (list, fence)) + rethrow() + end + @lock queue.lock push!(queue.submitted, (list, fence)) + sync_each_submission() && synchronize(queue) + return r +end + +function take_list!(queue::ZeCommandQueue) + @lock queue.lock begin + while !isempty(queue.submitted) + _, fence = first(queue.submitted) + res = unchecked_zeFenceQueryStatus(fence) + res == RESULT_NOT_READY && break + res == RESULT_SUCCESS || throw_api_error(res) + push!(queue.spare, popfirst!(queue.submitted)) + end + if !isempty(queue.spare) + list, fence = last(queue.spare) + reset(list) + zeFenceReset(fence) + return pop!(queue.spare) + end + end + + list = ZeCommandList(queue.context, queue.device, queue.ordinal) + fence_ref = Ref{ze_fence_handle_t}() + zeFenceCreate(queue, Ref(ze_fence_desc_t()), fence_ref) + return list, fence_ref[] +end + ## groups diff --git a/lib/level-zero/oneL0.jl b/lib/level-zero/oneL0.jl index f8e32e8a..565391f3 100644 --- a/lib/level-zero/oneL0.jl +++ b/lib/level-zero/oneL0.jl @@ -122,8 +122,8 @@ function Base.showerror(io::IO, err::OutOfGPUMemoryError) end include("context.jl") -include("cmdqueue.jl") include("cmdlist.jl") +include("cmdqueue.jl") include("fence.jl") include("event.jl") include("synchronization.jl") diff --git a/src/compiler/execution.jl b/src/compiler/execution.jl index b5b3a180..34bfc7bf 100644 --- a/src/compiler/execution.jl +++ b/src/compiler/execution.jl @@ -41,8 +41,7 @@ launches the kernel on the GPU. - `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=global_stream(...)`: Submission target — the task's `oneStream` by default. An - explicit `ZeCommandQueue` is also accepted and submits through a per-dispatch command - list. + explicit `ZeCommandQueue` is also accepted and submits each launch as a command list. # Examples @@ -428,8 +427,7 @@ end return end -# explicit-queue compatibility: `@oneapi queue=...` submits through a per-dispatch -# command list, as it always has +# explicit-queue compatibility: `@oneapi queue=...` submits each launch as a command list @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) diff --git a/test/level-zero.jl b/test/level-zero.jl index 76048d11..d994f63f 100644 --- a/test/level-zero.jl +++ b/test/level-zero.jl @@ -95,6 +95,26 @@ execute!(queue) do list @test list isa ZeCommandList end +# lists are reused once they have executed +synchronize(queue) +handles = Set{oneL0.ze_command_list_handle_t}() +for _ in 1:100 + execute!(queue) do list + push!(handles, list.handle) + end + synchronize(queue) +end +@test length(handles) == 1 + +# including when appending fails +@test_throws ErrorException execute!(queue) do list + error("append failed") +end +execute!(queue) do list + @test list.handle in handles +end +synchronize(queue) + end group = first(compute_groups(dev)) @@ -154,6 +174,22 @@ end # @test kernel_timestamp(timed_event).global.start != nothing # @test kernel_timestamp(timed_event).context.start != nothing +# a command list that hasn't executed yet is not reused. this submits work that waits on an +# event signaled afterwards, which the ONEAPI_SYNC_EACH_SUBMISSION workaround would deadlock +event = ZeEventPool(ctx, 1)[1] +oneL0.sync_each_submission(false) do + gated = Ref{oneL0.ze_command_list_handle_t}() + execute!(queue) do list + append_wait!(list, event) + gated[] = list.handle + end + execute!(queue) do list + @test list.handle != gated[] + end + signal(event) + synchronize(queue) +end + end