From d86a982ab21d9014c35501f4fe110e148b210655 Mon Sep 17 00:00:00 2001 From: Tim Besard Date: Wed, 21 Oct 2026 11:52:34 +0200 Subject: [PATCH 1/4] Add a `debug_level` keyword to `@oneapi` Kernels were always compiled at Julia's own debug level, which decides how much GPUCompiler reports about exceptions. The keyword makes it possible to get more detail for a single kernel without restarting Julia with -g2. --- docs/src/api/kernels.md | 2 ++ src/compiler/compilation.jl | 5 +++-- src/compiler/execution.jl | 4 +++- test/execution.jl | 14 ++++++++++++++ 4 files changed, 22 insertions(+), 3 deletions(-) diff --git a/docs/src/api/kernels.md b/docs/src/api/kernels.md index 8a525fe7..512e6172 100644 --- a/docs/src/api/kernels.md +++ b/docs/src/api/kernels.md @@ -20,6 +20,8 @@ launches the kernel on the GPU. - `kernel::Bool=false`: Whether to compile as a kernel or device function - `name::Union{String,Nothing}=nothing`: Explicit name for the kernel - `always_inline::Bool=false`: Whether to always inline device functions +- `debug_level::Int`: How much to report about exceptions thrown by the kernel (defaults to + Julia's `-g` level) **Launch Keywords (runtime):** - `groups`: Number of workgroups (required). Can be an integer or tuple. diff --git a/src/compiler/compilation.jl b/src/compiler/compilation.jl index a78b01e3..ef456c87 100644 --- a/src/compiler/compilation.jl +++ b/src/compiler/compilation.jl @@ -201,7 +201,8 @@ function _atomics_capabilities(dev, properties; supports_fp16, supports_fp64) end @noinline function _compiler_config(dev; kernel=true, name=nothing, always_inline=false, - sub_group_size=nothing, kwargs...) + sub_group_size=nothing, + debug_level=Base.JLOptions().debug_level, kwargs...) properties = oneL0.module_properties(dev) support = eltype_support(dev) supports_fp16 = support.float16 @@ -245,7 +246,7 @@ end supports_fp16, supports_fp64, supports_bfloat16, driver = :intel, kwargs...) params = oneAPICompilerParams(; sub_group_size) - CompilerConfig(target, params; kernel, name, always_inline) + CompilerConfig(target, params; kernel, name, always_inline, debug_level) end # run inference + LLVM codegen + SPIR-V emission. returns `(image, entry)`, both diff --git a/src/compiler/execution.jl b/src/compiler/execution.jl index f9660c68..0851c7d1 100644 --- a/src/compiler/execution.jl +++ b/src/compiler/execution.jl @@ -4,7 +4,7 @@ export @oneapi, zefunction, kernel_convert ## high-level @oneapi interface const MACRO_KWARGS = [:launch] -const COMPILER_KWARGS = [:kernel, :name, :always_inline, :atomics, :sub_group_size] +const COMPILER_KWARGS = [:kernel, :name, :always_inline, :atomics, :sub_group_size, :debug_level] const LAUNCH_KWARGS = [:groups, :items, :queue] """ @@ -36,6 +36,8 @@ launches the kernel on the GPU. - `sub_group_size::Union{Int,Nothing}=nothing`: The sub-group size the kernel has to be compiled for, one of the device's `oneL0.compute_properties(dev).subGroupSizes`. By default, the compiler chooses one. +- `debug_level::Int`: How much to report about exceptions thrown by the kernel. Defaults to + Julia's debug level (the `-g` option). ## Launch Keywords (runtime) - `groups`: Number of workgroups (required). Can be an integer or tuple. diff --git a/test/execution.jl b/test/execution.jl index 42a0fef5..c9563c4e 100644 --- a/test/execution.jl +++ b/test/execution.jl @@ -5,6 +5,11 @@ using StaticArrays dummy() = return +function store_checked(a, i) + a[i] = 1 + return +end + @testset "@oneapi" begin @test_throws UndefVarError @oneapi undefined() @@ -57,6 +62,15 @@ end @test_throws ArgumentError @oneapi sub_group_size=maximum(sizes)+1 dummy() end +@testset "debug level" begin + T = Tuple{oneDeviceVector{Int, oneAPI.AS.CrossWorkgroup}, Int} + ir(level) = sprint(io -> oneAPI.code_llvm(io, store_checked, T; kernel = true, + debug_level = level)) + @test !occursin("report_exception", ir(0)) + @test occursin("report_exception", ir(1)) && !occursin("report_exception_frame", ir(1)) + @test occursin("report_exception_frame", ir(2)) +end + @testset "inference" begin foo() = @oneapi dummy() From 000a1cd9712ee64f0f7d72aa1d56d13bf1d75190 Mon Sep 17 00:00:00 2001 From: Tim Besard Date: Wed, 21 Oct 2026 11:57:29 +0200 Subject: [PATCH 2/4] Report device exceptions to the host A work-item that threw only printed a message, if anything, and exited, so the host carried on with incomplete results. Kernels now report exceptions to their stream, and synchronizing it, or waiting for memory it owns, throws a `KernelException` that carries the exception's name and reason, and at -g2 the throwing work-item and its backtrace. Each stream owns a report in host memory, checked when synchronizing without any added submission or wait, and a claim word in device memory. The first work-item to throw claims the report by storing its id in the claim word, which makes it the report's only writer, as devices need not support atomics on host memory. Reports are carved out of slabs shared by a device's streams, as every allocation adds to the cost of every launch. Kernels keep no per-work-item state, which would take private memory and slow down every kernel. Strings are reported by the address of the constant holding them, and read by the host once an exception was reported, only for addresses within an allocation, such as a module's constants, and up to its end. Copying them on the device puts a loop in every reporting function, which GPUCompiler calls for every frame of every throw site, and made IGC compile bounds-checked kernels up to 14 times slower. Reporting is kept out of line, which keeps those kernels fast, except at -g2, where IGC compiles all the frames faster inlined. `@gputhrow` now records its name and reason in the report instead of printing them, which also removes the printf from bounds-checked kernels. --- docs/src/api/kernels.md | 11 +- docs/src/kernels.md | 34 +++++- src/compiler/exceptions.jl | 213 +++++++++++++++++++++++++++++++++++++ src/compiler/execution.jl | 15 +-- src/context.jl | 18 +++- src/device/quirks.jl | 8 -- src/device/runtime.jl | 156 +++++++++++++++++++++++++-- src/memory.jl | 9 +- src/oneAPI.jl | 1 + test/execution.jl | 186 +++++++++++++++++++++++++++++++- 10 files changed, 618 insertions(+), 33 deletions(-) create mode 100644 src/compiler/exceptions.jl diff --git a/docs/src/api/kernels.md b/docs/src/api/kernels.md index 512e6172..2c3d6478 100644 --- a/docs/src/api/kernels.md +++ b/docs/src/api/kernels.md @@ -21,14 +21,21 @@ launches the kernel on the GPU. - `name::Union{String,Nothing}=nothing`: Explicit name for the kernel - `always_inline::Bool=false`: Whether to always inline device functions - `debug_level::Int`: How much to report about exceptions thrown by the kernel (defaults to - Julia's `-g` level) + Julia's `-g` level; see `KernelException`) **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::Union{oneStream,ZeCommandQueue}`: Submission target (defaults to the task's `global_stream`). An explicit `ZeCommandQueue` submits through a per-dispatch command - list and is not ordered with respect to the task's stream. + list and is not ordered with respect to the task's stream. Exceptions thrown by such + kernels are not reported. + +### `KernelException` + +Thrown when synchronizing a stream on which a kernel threw an exception. It carries the +exception's `name` and `reason` (from debug level 1), and the `work_item` and `work_group` +that threw it along with a `backtrace` (from debug level 2). ### `zefunction(f, tt; kwargs...)` diff --git a/docs/src/kernels.md b/docs/src/kernels.md index a9eeb46f..afe7f9a5 100644 --- a/docs/src/kernels.md +++ b/docs/src/kernels.md @@ -71,8 +71,8 @@ private to each work-item. Each allocation is rounded up to 16 bytes, and memory reclaimed when the work-item exits. Allocated objects must not be shared with other work-items or retained across kernel launches. -When the heap is exhausted, the work-item prints an error and exits without completing -its work. This does not raise a host-side exception, and kernel output may be incomplete: +When the heap is exhausted, the work-item prints an error and throws an exception (see +[Exceptions](@ref)), so kernel output may be incomplete: ``` ERROR: Out of dynamic GPU memory (trying to allocate 4 bytes) @@ -82,3 +82,33 @@ Kernels without remaining allocations do not reserve an arena. The device compil optimize away some heap storage, but allocations can increase private-memory use and reduce performance. Avoid repeated allocations in loops: even short-lived objects consume heap space until the work-item exits. + + +## Exceptions + +A work-item that throws an exception, e.g. on an out-of-bounds access, exits without +completing its work. The other work-items keep running. The exception is reported as a +`KernelException` when the stream the kernel was launched on is synchronized, e.g. by +`synchronize()` or by copying an array to the host: + +```julia +julia> function kernel(a) + a[get_global_id() + 1] = 1 + return + end + +julia> a = oneArray{Int}(undef, 1); + +julia> @oneapi kernel(a) + +julia> synchronize() +ERROR: KernelException: A BoundsError was thrown on Intel(R) Arc(TM) A770 Graphics: Out-of-bounds array access +For more details, run Julia with `-g2`, or launch the kernel with `@oneapi debug_level=2`. +``` + +The amount of detail depends on the debug level the kernel is compiled with, which defaults +to Julia's `-g` level. At level 0 only the fact that an exception was thrown is reported; at +level 1 (the default) also its type and reason; at level 2 also the work-item that threw it +and a backtrace. Only one exception is reported per synchronization. Exceptions in kernels +launched on an explicitly created `ZeCommandQueue` are not reported. + diff --git a/src/compiler/exceptions.jl b/src/compiler/exceptions.jl new file mode 100644 index 00000000..94a3f14d --- /dev/null +++ b/src/compiler/exceptions.jl @@ -0,0 +1,213 @@ +# reporting of exceptions thrown by kernels + +export KernelException + +""" + KernelException + +An exception thrown by a kernel on device `dev`, reported when synchronizing the stream it +was launched on (e.g. by `synchronize()`, or a copy to the host such as `Array(x)`). + +What is known about the exception depends on the debug level the kernel was compiled with +(Julia's `-g` level, or `@oneapi debug_level=...`): + +- `0`: only that an exception was thrown; +- `1`: also the `name` of its type and the `reason`, for exceptions thrown by Julia's + runtime (e.g. bounds errors, or domain errors in math functions); +- `2`: also the position of the work-item that threw (`work_item` is its local id, + `work_group` the id of its work-group), and its `backtrace` as `(function, file, line)` + tuples. + +Fields that were not recorded are empty, or zero. Only one exception is reported per +synchronization, and the other work-items of the kernel keep running. +""" +struct KernelException <: Exception + dev::ZeDevice + name::String + reason::String + work_item::NTuple{3, Int} + work_group::NTuple{3, Int} + backtrace::Vector{Tuple{String, String, Int}} +end + +function Base.showerror(io::IO, err::KernelException) + name = isempty(err.name) ? "exception" : err.name + article = first(uppercase(name)) in ('A', 'E', 'I', 'O', 'U') ? "An" : "A" + print(io, "KernelException: $article $name was thrown") + # positions are 1-based, so zero coordinates mean they were not recorded + if err.work_item != (0, 0, 0) + print(io, " by work-item ", join(err.work_item, '×'), + " in work-group ", join(err.work_group, '×')) + end + print(io, " on ", oneL0.properties(err.dev).name) + isempty(err.reason) || print(io, ": ", err.reason) + if err.work_item == (0, 0, 0) + print(io, "\nFor more details, run Julia with `-g2`, or launch the kernel with `@oneapi debug_level=2`.") + else + print(io, "\nStacktrace:") + for (i, (func, file, line)) in enumerate(err.backtrace) + print(io, "\n [", i, "] ", func) + isempty(file) || print(io, " at ", file, ":", line) + end + end +end + +const EXCEPTION_STRING_MAX = 4096 + +# buffers that copies may still be writing to, as their list could not be synchronized +const unsynchronized_string_buffers = Any[] +const unsynchronized_string_buffers_lock = ReentrantLock() + +# Read the strings at `addresses`, as reported by a kernel, returning a dictionary that maps +# the addresses that could be read to their contents. The addresses come from memory the +# kernel could have corrupted, so they are only read if they fall within an allocation, +# such as a module's constants, and only up to the end of it. Freed modules are not read. +function read_device_strings(list, ctx::ZeContext, addresses) + buffers = Dict{UInt64, Vector{UInt8}}() + for address in addresses + (address == 0 || haskey(buffers, address)) && continue + base = Ref{Ptr{Cvoid}}() + size = Ref{Csize_t}() + res = oneL0.unchecked_zeMemGetAddressRange(ctx, Ptr{Cvoid}(address), base, size) + res == oneL0.RESULT_SUCCESS || continue + available = UInt64(UInt(base[]) + size[]) - address + available > 0 || continue + buffers[address] = Vector{UInt8}(undef, min(available, EXCEPTION_STRING_MAX)) + end + isempty(buffers) && return Dict{UInt64, String}() + copied = false + GC.@preserve buffers try + for (address, buffer) in buffers + append_copy!(list, pointer(buffer), ZePtr{UInt8}(address), length(buffer)) + end + copied = true + catch err + bt = catch_backtrace() + @error "Failed to read the strings of a kernel exception" exception=(err, bt) + finally + # copies appended before a failure still write to the buffers + if oneL0.unchecked_zeCommandListHostSynchronize(list, typemax(UInt64)) != + oneL0.RESULT_SUCCESS + @lock unsynchronized_string_buffers_lock begin + push!(unsynchronized_string_buffers, buffers) + end + copied = false + end + end + copied || return Dict{UInt64, String}() + return Dict(address => String(buffer[1:something(findfirst(iszero, buffer), + length(buffer) + 1) - 1]) + for (address, buffer) in buffers) +end + + +## per-stream report + +# Reports are carved out of slabs shared by a device's streams, as kernels access memory +# indirectly, which makes every allocation add to the cost of every launch (see `pool.jl`). +const REPORTS_PER_SLAB = 64 + +const free_report_slots = Dict{Tuple{ZeContext, ZeDevice}, Vector{ReportSlot}}() + +KernelState(s::oneStream) = KernelState(UInt(s.exceptions.report) % UInt64, + UInt(s.exceptions.claim) % UInt64) +# explicitly created queues have no report +KernelState(::ZeCommandQueue) = KernelState() + +# the pattern resetting a claim; a global, as Level Zero does not specify how long a fill's +# pattern has to remain valid +const unclaimed = Ref{UInt32}(0) + +reset_claim!(list, slot::ReportSlot) = + GC.@preserve unclaimed append_fill!(list, slot.claim, + Base.unsafe_convert(Ptr{UInt32}, unclaimed), + sizeof(UInt32), sizeof(UInt32)) + +clear_report!(slot::ReportSlot) = + fill!(unsafe_wrap(Array, convert(Ptr{UInt8}, slot.report), sizeof(ExceptionReport)), 0x00) + +function new_report_slab(ctx::ZeContext, dev::ZeDevice) + reports = allocate(oneL0.HostBuffer, ctx, dev, + REPORTS_PER_SLAB * sizeof(ExceptionReport), sizeof(UInt64)) + claims = try + allocate(oneL0.DeviceBuffer, ctx, dev, + REPORTS_PER_SLAB * sizeof(UInt32), sizeof(UInt32)) + catch + release(reports) + rethrow() + end + return [ReportSlot(convert(Ptr{ExceptionReport}, reports) + + (i - 1) * sizeof(ExceptionReport), + convert(ZePtr{UInt32}, pointer(claims)) + (i - 1) * sizeof(UInt32)) + for i in 1:REPORTS_PER_SLAB] +end + +# The claim is reset on `list`, in order before any kernel uses it. Slabs are never freed. +function allocate_report_slot(ctx::ZeContext, dev::ZeDevice, list) + slot = with_pool_lock() do + free = get(free_report_slots, (ctx, dev), nothing) + (free === nothing || isempty(free)) ? nothing : pop!(free) + end + if slot === nothing + # outside of the lock, as allocating memory may collect garbage, freeing slots + slots = new_report_slab(ctx, dev) + slot = pop!(slots) + with_pool_lock() do + append!(get!(Vector{ReportSlot}, free_report_slots, (ctx, dev)), slots) + end + end + try + clear_report!(slot) + reset_claim!(list, slot) + catch + release_report_slot(ctx, dev, slot) + rethrow() + end + return slot +end + +release_report_slot(ctx::ZeContext, dev::ZeDevice, slot::ReportSlot) = + with_pool_lock(() -> push!(free_report_slots[(ctx, dev)], slot)) + +# Called when a stream is collected. Its kernels may still be executing, and writing the +# report, in which case the slot is leaked, as is the command list (see its finalizer). +function release_report_slot(s::oneStream) + list = s.immediate + # the list's finalizer may have run already, destroying it once it was idle + if list.handle == C_NULL || + oneL0.unchecked_zeCommandListHostSynchronize(list, 0) == oneL0.RESULT_SUCCESS + release_report_slot(s.ctx, s.dev, s.exceptions) + end + return +end + +# Check whether a kernel on `s` threw an exception, returning it after resetting the report. +# The caller must hold `s.lock`. +function check_exception!(s::oneStream) + # Kernels may still write the report while the stream is busy. That is only the case + # after waiting for it if another task submitted work in the meantime, in which case the + # next synchronization reports the exception. + oneL0.unchecked_zeCommandListHostSynchronize(s.immediate, 0) == oneL0.RESULT_SUCCESS || + return nothing + + slot = s.exceptions + unsafe_load(convert(Ptr{Int32}, slot.report)) == 0 && return nothing # `status` + report = unsafe_load(slot.report) + + # reset the claim first: if that fails, the exception is still reported later + reset_claim!(s.immediate, slot) + clear_report!(slot) + + nframes = clamp(Int(report.num_frames), 0, EXCEPTION_MAX_FRAMES) + frames = report.frames[1:nframes] + strings = read_device_strings(s.immediate, s.ctx, + [report.name; report.reason; + [frame.func for frame in frames]; + [frame.file for frame in frames]]) + str(address) = get(strings, address, "") + backtrace = Tuple{String, String, Int}[(str(frame.func), str(frame.file), + Int(frame.line)) for frame in frames] + return KernelException(s.dev, str(report.name), str(report.reason), + Int.(report.work_item[1:3]), Int.(report.work_group[1:3]), + backtrace) +end diff --git a/src/compiler/execution.jl b/src/compiler/execution.jl index 0851c7d1..69281739 100644 --- a/src/compiler/execution.jl +++ b/src/compiler/execution.jl @@ -36,14 +36,15 @@ launches the kernel on the GPU. - `sub_group_size::Union{Int,Nothing}=nothing`: The sub-group size the kernel has to be compiled for, one of the device's `oneL0.compute_properties(dev).subGroupSizes`. By default, the compiler chooses one. -- `debug_level::Int`: How much to report about exceptions thrown by the kernel. Defaults to - Julia's debug level (the `-g` option). +- `debug_level::Int`: How much to report about exceptions thrown by the kernel (see + [`KernelException`](@ref)). Defaults to Julia's debug level (the `-g` option). ## 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=global_stream(...)`: Submission target — the task's `oneStream` by default. An - explicit `ZeCommandQueue` is also accepted and submits each launch as a command list. + explicit `ZeCommandQueue` is also accepted and submits each launch as a command list; + exceptions thrown by such launches are not reported. # Examples @@ -221,7 +222,9 @@ kernel_convert(arg, refs=nothing) = adapt(KernelAdaptor(refs), arg) abstract type AbstractKernel{F,TT} end -@inline @generated function call(kernel::AbstractKernel{F,TT}, args::Tuple; call_kwargs...) where {F,TT} +@inline @generated function call(kernel::AbstractKernel{F,TT}, args::Tuple; + queue=global_stream(context(), device()), + call_kwargs...) where {F,TT} sig = Tuple{F, TT.parameters...} # Base.signature_type with a function type args = (:(kernel.f), (:( args[$i] ) for i in 1:fieldcount(args))...) @@ -241,13 +244,13 @@ abstract type AbstractKernel{F,TT} end # Match GPUCompiler's hidden state argument. Keep onecall usable for foreign kernels. pushfirst!(call_t, KernelState) - pushfirst!(call_args, :(KernelState())) + pushfirst!(call_args, :(KernelState(queue))) # finalize types call_tt = Base.to_tuple_type(call_t) quote - onecall(kernel.fun, $call_tt, ($(call_args...),); call_kwargs...) + onecall(kernel.fun, $call_tt, ($(call_args...),); queue, call_kwargs...) end end diff --git a/src/context.jl b/src/context.jl index abbf3a40..7b0c0808 100644 --- a/src/context.jl +++ b/src/context.jl @@ -253,6 +253,12 @@ function batching_supported(ctx::ZeContext, dev::ZeDevice) end end +# a stream's exception report, see `allocate_report_slot` +struct ReportSlot + report::Ptr{ExceptionReport} # in host memory + claim::ZePtr{UInt32} # in device memory +end + # The per-task submission target. `immediate` is where Julia-side work (kernels, copies, # fills) is submitted: an in-order asynchronous immediate command list. Kernel launches are # mostly recorded in batches first, see above. @@ -289,6 +295,9 @@ mutable struct oneStream # stream, maintained by the scratch hedge (see src/compiler/execution.jl) scratch_hwm::Int const priority::oneL0.ze_command_queue_priority_t + + # where kernels report exceptions, see `check_exception!` + const exceptions::ReportSlot end function create_stream(ctx::ZeContext, dev::ZeDevice, @@ -312,9 +321,11 @@ function create_stream(ctx::ZeContext, dev::ZeDevice, "with $(err.code)") end batching = BATCHING[] && batching_supported(ctx, dev) + slot = allocate_report_slot(ctx, dev, list) s = oneStream(ctx, dev, list, ReentrantLock(), batching, nothing, Batch[], Batch[], 0, nothing, true, nothing, DataRef[], - Batch[], nothing, nothing, false, 0, priority) + Batch[], nothing, nothing, false, 0, priority, slot) + finalizer(release_report_slot, s) return register_stream!(ctx, dev, s) end @@ -666,7 +677,7 @@ function oneL0.synchronize(s::oneStream; blocking::Bool=false) return end -# synchronize a stream, returning the error of a failed submission +# synchronize a stream, returning the error of a failed submission or of a kernel 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 @@ -686,7 +697,8 @@ function synchronize_stream!(s::oneStream; blocking::Bool=false) maybe_unregister_batching!(s) err = s.error s.error = nothing - err + # a kernel exception stays pending while there's a submission error to report + err === nothing ? check_exception!(s) : err end return err end diff --git a/src/device/quirks.jl b/src/device/quirks.jl index 942fd1a3..5f701e56 100644 --- a/src/device/quirks.jl +++ b/src/device/quirks.jl @@ -1,11 +1,3 @@ -# throw a device-side exception of type `name`, printing the type and `reason` -macro gputhrow(name::String, reason::String) - quote - @println "ERROR: " $name ": " $reason "." - throw(nothing) - end -end - # math.jl @device_override @noinline Base.Math.throw_complex_domainerror(f::Symbol, x) = @gputhrow "DomainError" "This operation requires a complex input to return a complex result" diff --git a/src/device/runtime.jl b/src/device/runtime.jl index c7551b9b..3ea9be8a 100644 --- a/src/device/runtime.jl +++ b/src/device/runtime.jl @@ -7,9 +7,17 @@ struct KernelState # Initialized on the device by add_heap!; the host passes a null pointer. heap::LLVMPtr{UInt8, AS.Function} + + # Addresses of the launching stream's exception report and claim word, or zero when + # exceptions are not reported. Integers, as SPIR-V validation rejects pointer fields in + # by-value kernel arguments. + exception_report::UInt64 + exception_claim::UInt64 end -KernelState() = KernelState(reinterpret(LLVMPtr{UInt8, AS.Function}, C_NULL)) +KernelState(exception_report::UInt64=UInt64(0), exception_claim::UInt64=UInt64(0)) = + KernelState(reinterpret(LLVMPtr{UInt8, AS.Function}, C_NULL), + exception_report, exception_claim) @inline @generated kernel_state() = GPUCompiler.kernel_state_value(KernelState) @@ -59,20 +67,154 @@ end ## exceptions -# SPIR-V has no way to abort a kernel, and the exception is not reported to the host: the -# work-item that threw simply exits (see `lower_unreachable_control_flow!` in GPUCompiler). -function signal_exception() - return +# A work-item that throws reports to its stream's `ExceptionReport`, which the host checks +# when synchronizing the stream (see `check_exception!`). SPIR-V cannot abort a kernel, so +# the work-item then exits (see `lower_unreachable_control_flow!` in GPUCompiler) while the +# others keep running. +# +# The report lives in host memory, which the host can read without migrating pages, but on +# which devices need not support atomics. So the first work-item to claim the stream's claim +# word, in device memory, becomes the report's only writer. The claim is held until the +# host has read the report, so a stream reports one exception per synchronization. +# +# No per-work-item state is kept, as reserving private memory slows down every kernel. +# +# What is reported depends on the debug level: at 0 only that an exception was thrown, at 1 +# also the exception's name and reason, and at 2 also the position of the work-item and a +# backtrace. +# +# Strings are reported by the address of the constant holding them, and read by the host. +# Copying them here would put a loop in every reporting function, which GPUCompiler calls +# for every frame of every throw site, and makes IGC generate many times more code. + +const EXCEPTION_MAX_FRAMES = 16 + +# String fields are addresses of NUL-terminated strings in device memory, or zero. +struct ExceptionFrame + func::UInt64 + file::UInt64 + line::Int32 end -function report_exception(ex) - return +struct ExceptionReport + status::Int32 + num_frames::Int32 + # Four components, as narrowing or merging adjacent three-component accesses can + # produce six-wide vectors, which SPIR-V does not support. + work_item::NTuple{4, UInt64} + work_group::NTuple{4, UInt64} + name::UInt64 + reason::UInt64 + frames::NTuple{EXCEPTION_MAX_FRAMES, ExceptionFrame} +end + +# Offsets are computed when generating the code, as inference does not always constant-fold +# `fieldoffset` in the code `@gputhrow` reaches, and are added as `UInt64`, as mixing in an +# `Int` would be a checked conversion. +@generated function report_field(::Val{field}) where {field} + T = fieldtype(ExceptionReport, field) + offset = fieldoffset(ExceptionReport, Base.fieldindex(ExceptionReport, field)) % UInt64 + :(reinterpret(LLVMPtr{$T, AS.CrossWorkgroup}, kernel_state().exception_report + $offset)) +end + +@generated function frame_field(idx, ::Val{field}) where {field} + T = fieldtype(ExceptionFrame, field) + offset = (fieldoffset(ExceptionReport, Base.fieldindex(ExceptionReport, :frames)) + + fieldoffset(ExceptionFrame, Base.fieldindex(ExceptionFrame, field))) % UInt64 + stride = sizeof(ExceptionFrame) % UInt64 + quote + frame = (idx - 1) % UInt64 * $stride + reinterpret(LLVMPtr{$T, AS.CrossWorkgroup}, + kernel_state().exception_report + frame + $offset) + end +end + +const CString = LLVMPtr{Cchar, AS.CrossWorkgroup} + +@llvmgenerated builder function string_pointer(::Val{str})::CString where {str} + globalstring_ptr!(builder, String(str); addrspace=AS.CrossWorkgroup) +end + +# throw a device-side exception of type `name`, which is reported together with `reason` +macro gputhrow(name::String, reason::String) + # without line numbers, so that backtraces start at the caller + Base.remove_linenums!(quote + if kernel_debug_level() >= 1 + reporting_call(record_exception!, string_pointer($(Val(Symbol(name)))), + string_pointer($(Val(Symbol(reason))))) + end + throw(nothing) + end) +end + +# The work-item's id in the claim word, where zero means unclaimed. Ids repeat beyond 2^32-1 +# work-items, in which case two throwing work-items could both write the report. +@inline function work_item_id() + id = get_global_linear_id() % UInt32 + return id == 0 ? 1 % UInt32 : id +end + +# Claim the report, returning whether this work-item owns it. Idempotent, so that every +# reporting function can check ownership with it. The owner sets the report's status before +# exiting, which keeps work-items with the same id in later kernels from claiming it too. +@inline function claim_report!() + claim = kernel_state().exception_claim + claim == 0 && return false + # A relaxed claim suffices: the host only reads the report once the kernel has completed. + claim_ptr = reinterpret(LLVMPtr{UInt32, AS.CrossWorkgroup}, claim) + id = work_item_id() + old = SPIRVIntrinsics.atomic_cmpxchg!(claim_ptr, 0 % UInt32, id) + old == 0 && return true + return old == id && unsafe_load(report_field(Val(:status))) == 0 end +# GPUCompiler inlines throwing functions into every call site, where reporting is kept out +# of line, as it slows down kernels otherwise. Except at -g2, where every frame of every +# throw site is reported too: IGC compiles that faster inlined than as calls. +@inline reporting_call(f, args...) = + kernel_debug_level() >= 2 ? f(args...) : out_of_line(f, args...) +@noinline out_of_line(f, args...) = f(args...) + +string_address(str::CString) = reinterpret(UInt64, str) + +# Record the exception's name and reason, unless one was recorded already: `@gputhrow` comes +# first, and knows a more specific name than GPUCompiler derives from the throw. +@inline function record_exception!(name::CString, reason::Union{CString, Nothing}=nothing) + reporting_call(claim_report!) || return false + name_ptr = report_field(Val(:name)) + if unsafe_load(name_ptr) == 0 + unsafe_store!(name_ptr, string_address(name)) + reason === nothing || + unsafe_store!(report_field(Val(:reason)), string_address(reason)) + end + return true +end + +report_exception(ex) = (reporting_call(record_exception!, ex); return) + function report_exception_name(ex) + record_exception!(ex) || return + unsafe_store!(report_field(Val(:work_item)), + (get_local_id(1) % UInt64, get_local_id(2) % UInt64, + get_local_id(3) % UInt64, 0 % UInt64)) + unsafe_store!(report_field(Val(:work_group)), + (get_group_id(1) % UInt64, get_group_id(2) % UInt64, + get_group_id(3) % UInt64, 0 % UInt64)) return end function report_exception_frame(idx, func, file, line) + claim_report!() || return + 1 <= idx <= EXCEPTION_MAX_FRAMES || return + unsafe_store!(frame_field(idx, Val(:func)), string_address(func)) + unsafe_store!(frame_field(idx, Val(:file)), string_address(file)) + unsafe_store!(frame_field(idx, Val(:line)), line % Int32) + unsafe_store!(report_field(Val(:num_frames)), idx % Int32) + return +end + +function signal_exception() + # At debug level 0 this is the only reporting function, and claims the report. + reporting_call(claim_report!) && unsafe_store!(report_field(Val(:status)), 1 % Int32) return end diff --git a/src/memory.jl b/src/memory.jl index 4afeabdf..10d2fe31 100644 --- a/src/memory.jl +++ b/src/memory.jl @@ -82,8 +82,9 @@ end Base.sizeof(managed::Managed) = sizeof(managed.mem) -# wait for all work on a stream, possibly from another task. unlike `synchronize`, this -# leaves `mkl_dirty` alone, which only the stream's own task may modify. +# wait for all work on a stream, possibly from another task, throwing exceptions its +# kernels threw: whoever waits for memory is about to use what they computed. unlike +# `synchronize`, this leaves `mkl_dirty` alone, which only the stream's own task may modify. function drain(s::oneStream) @lock s.lock begin check_error(s) @@ -92,10 +93,12 @@ function drain(s::oneStream) oneL0.nonblocking_synchronize(s.immediate) q = Base.@atomic s.queue q === nothing || oneL0.nonblocking_synchronize(q) - @lock s.lock begin + err = @lock s.lock begin retire!(s) s.idle = true + check_exception!(s) end + err === nothing || throw(err) return end drain(q::ZeCommandQueue) = oneL0.nonblocking_synchronize(q) diff --git a/src/oneAPI.jl b/src/oneAPI.jl index fb8ddbfd..e9d3a88a 100644 --- a/src/oneAPI.jl +++ b/src/oneAPI.jl @@ -55,6 +55,7 @@ include("array.jl") # compiler implementation include("compiler/compilation.jl") include("compiler/execution.jl") +include("compiler/exceptions.jl") include("compiler/reflection.jl") if Sys.islinux() diff --git a/test/execution.jl b/test/execution.jl index c9563c4e..2fd2e432 100644 --- a/test/execution.jl +++ b/test/execution.jl @@ -62,6 +62,7 @@ end @test_throws ArgumentError @oneapi sub_group_size=maximum(sizes)+1 dummy() end + @testset "debug level" begin T = Tuple{oneDeviceVector{Int, oneAPI.AS.CrossWorkgroup}, Int} ir(level) = sprint(io -> oneAPI.code_llvm(io, store_checked, T; kernel = true, @@ -1284,10 +1285,11 @@ end @oneapi items = 256 loop_kernel(a, fits) @test Array(a) == Float32.(1:256) .+ 2 * fits a = oneArray(Float32[1]) - _, out = @grab_output begin + err, out = @grab_output begin @oneapi loop_kernel(a, fits + 1) - synchronize() + try synchronize(); nothing catch err; err end end + @test err isa KernelException @test occursin("Out of dynamic GPU memory", out) @test Array(a) == [1] @@ -1333,6 +1335,186 @@ end @test !occursin(r"%heap\d* = alloca", sprint(io -> oneAPI.code_llvm(io, plain_kernel, T; kernel = true))) end +# the exception a kernel threw, as reported by synchronizing `stream` +function kernel_exception(stream=oneAPI.global_stream(context(), device())) + try + synchronize(stream) + return nothing + catch err + err isa KernelException || rethrow() + return err + end +end + +function div_kernel(a, d) + a[1] = div(a[1], d) + return +end + +@testset "exceptions" begin + a = oneArray([0]) + + # at debug level 0, only that an exception was thrown + @oneapi debug_level=0 store_checked(a, 2) + err = kernel_exception() + @test err isa KernelException + @test err.name == "" && err.reason == "" && isempty(err.backtrace) + @test err.work_item == err.work_group == (0, 0, 0) + @test kernel_exception() === nothing + + # at level 1, also the exception's type and reason, as passed to `@gputhrow` + @oneapi debug_level=1 store_checked(a, 2) + err = kernel_exception() + @test err.name == "BoundsError" + @test err.reason == "Out-of-bounds array access" + @test err.work_item == (0, 0, 0) && isempty(err.backtrace) + @test occursin("A BoundsError was thrown on", sprint(showerror, err)) + # or as derived by GPUCompiler + @oneapi debug_level=1 div_kernel(a, 0) + err = kernel_exception() + @test err.name == "exception" && err.reason == "" + + # at level 2, also the position of the work-item, and a backtrace + @oneapi debug_level=2 items=4 groups=3 store_checked(a, 2) + err = kernel_exception() + @test err.name == "BoundsError" + @test all(1 .<= err.work_item .<= (4, 1, 1)) + @test all(1 .<= err.work_group .<= (3, 1, 1)) + @test endswith(err.backtrace[1][1], "throw_boundserror") + @test err.backtrace[end][1] == "store_checked" + @test endswith(err.backtrace[end][2], "execution.jl") + @test occursin(r"thrown by work-item \d×1×1 in work-group \d×1×1", sprint(showerror, err)) + @oneapi debug_level=2 div_kernel(a, 0) + err = kernel_exception() + @test err.name == "exception" + @test err.backtrace[end][1] == "div_kernel" + + # strings are reported by address, which the host only reads if it is valid + stream = oneAPI.global_stream(context(), device()) + @oneapi debug_level=2 store_checked(a, 2) + oneAPI.flush!(stream) + oneL0.synchronize(stream.immediate) + report = convert(Ptr{UInt8}, stream.exceptions.report) + field(T, name) = report + fieldoffset(T, Base.fieldindex(T, name)) + unsafe_store!(convert(Ptr{UInt64}, field(oneAPI.ExceptionReport, :name)), + 0x00000deadbeef000) + unsafe_store!(convert(Ptr{UInt64}, field(oneAPI.ExceptionReport, :frames)), + typemax(UInt64)) + err = kernel_exception() + @test err.name == "" && err.reason == "Out-of-bounds array access" + @test err.backtrace[1][1] == "" && err.backtrace[end][1] == "store_checked" + @test kernel_exception() === nothing + + # the stream remains usable, and other work-items keep running + function partially_oob_kernel(a) + i = get_global_id() + a[i + 1] = i + return + end + a = oneArray(zeros(Int, 256)) + @oneapi items=256 partially_oob_kernel(a) + @test kernel_exception() isa KernelException + @test Array(a)[2:end] == 1:255 + @oneapi store_checked(a, 1) + @test kernel_exception() === nothing + @test Array(a)[1] == 1 + + # an exception is reported by the synchronization following it, also from a batch + a = oneArray([0]) + for i in 1:10 + @oneapi store_checked(a, i == 5 ? 2 : 1) + end + @test kernel_exception() isa KernelException + @test kernel_exception() === nothing + + # and by host accesses that synchronize + @oneapi store_checked(a, 2) + @test_throws KernelException Array(a) + b = oneArray{Int, 1, oneL0.SharedBuffer}([0]) + @oneapi store_checked(b, 2) + @test_throws KernelException b[1] + # also when the memory is another task's, which the host waits for + wait(@async @oneapi store_checked(a, 2)) + @test_throws KernelException Array(a) + @test kernel_exception() === nothing + + # kernels that use the device heap too + function heap_oob_kernel(a, i) + x = heap_consume(Ref(a[1])) + a[i] = x + return + end + a = oneArray(Float32[1]) + @oneapi debug_level=1 heap_oob_kernel(a, 2) + @test kernel_exception().name == "BoundsError" + @oneapi debug_level=1 heap_oob_kernel(a, 1) + @test kernel_exception() === nothing + @test Array(a) == [2] + + # every stream reports the exceptions of its own kernels + a = oneArray([0]) + launched = Channel{Nothing}(1) + checked = Channel{Nothing}(1) + owner = @async begin + @oneapi store_checked(a, 2) + put!(launched, nothing) + take!(checked) + kernel_exception() + end + take!(launched) + @test kernel_exception() === nothing + put!(checked, nothing) + @test fetch(owner) isa KernelException + # also when launching on another task's stream + stream = fetch(@async oneAPI.global_stream(context(), device())) + @oneapi queue=stream store_checked(a, 2) + @test kernel_exception(stream) isa KernelException + @test kernel_exception() === nothing + + # and of tasks that finished without synchronizing, by whoever adopts their stream + batching = oneAPI.global_stream(context(), device()).batching && + !oneL0.sync_each_submission() + wait(@async begin + @oneapi store_checked(a, 1) + @oneapi store_checked(a, 2) + end) + err = try synchronize(); nothing catch err; err end + @test batching ? err isa KernelException : err === nothing + + # explicitly created queues have no report + queue = oneL0.ZeCommandQueue(context(), device(); + flags = oneL0.ZE_COMMAND_QUEUE_FLAG_IN_ORDER) + k = @oneapi launch=false store_checked(a, 2) + k(a, 2; queue) + oneL0.synchronize(queue) + @test kernel_exception() === nothing + + # collected streams return their report, unless their work may still use it + ctx, dev = context(), device() + pool = oneL0.ZeEventPool(ctx, 1, dev; flags=oneL0.ZE_EVENT_POOL_FLAG_HOST_VISIBLE) + gate = pool[1] + busy, idle = oneAPI.create_stream(ctx, dev), oneAPI.create_stream(ctx, dev) + oneL0.sync_each_submission(false) do + oneL0.execute!(busy) do list + oneL0.append_wait!(list, gate) + end + end + free = oneAPI.free_report_slots[(ctx, dev)] + finalize(busy) + @test !(busy.exceptions in free) + oneL0.signal(gate) + oneL0.synchronize(busy) + oneL0.synchronize(idle) + finalize(idle) + @test idle.exceptions in free + + # reporting needs no per-work-item memory, which would slow down every kernel + for level in 0:2 + k = @oneapi launch=false debug_level=level store_checked(a, 2) + @test oneL0.properties(k.fun).privateMemSize == 0 + end +end + @testset "loop exiting on counter wrap-around" begin # The SPIR-V back-end turns `i += 1; i == 0` into OpIAddCarry, and IGC's optimizer sinks # the extracts of its {i64, i1} result through the loop PHI. Before libigc_jll 2.34.4+5, From 39eb4f939438b2b56c18be8974de6bf2c8cf704e Mon Sep 17 00:00:00 2001 From: Tim Besard Date: Thu, 22 Oct 2026 16:23:59 +0200 Subject: [PATCH 3/4] Report exceptions of finished tasks independently of batching `synchronize()` reported the kernel exceptions of tasks that finished without synchronizing only if their stream was registered for adoption, which happened for batched launches only. Without batching, or when a task's last launch was submitted directly, the exception instead surfaced at whatever next waited for the task's memory, e.g. an unrelated launch on an explicit queue. Register every stream that launches a kernel, and only forget it once it is idle with nothing left to report, through a helper shared by `synchronize`, memory waiters and adoption. Adoption checks again after waiting that the stream is still orphaned, so that it doesn't take the exception of a running task that launched on it in the meantime. Tasks are referenced weakly, so that finished ones can be collected. --- docs/src/api/context.md | 2 +- docs/src/kernels.md | 6 ++- src/compiler/exceptions.jl | 8 +--- src/compiler/execution.jl | 2 +- src/context.jl | 86 +++++++++++++++++++++----------------- src/memory.jl | 10 +++-- test/execution.jl | 75 ++++++++++++++++++++++++++++++--- 7 files changed, 129 insertions(+), 60 deletions(-) diff --git a/docs/src/api/context.md b/docs/src/api/context.md index f4676c06..5207a75a 100644 --- a/docs/src/api/context.md +++ b/docs/src/api/context.md @@ -81,7 +81,7 @@ Block the host thread until all operations on the global stream for the current and device have completed: both the immediate command list and, if one has been created, the companion command queue. `synchronize(stream::oneStream)` does the same for an explicit stream. `synchronize()` also submits and waits for the work of tasks that -finished without synchronizing. +finished without synchronizing, and reports exceptions their kernels threw. ### Command batching diff --git a/docs/src/kernels.md b/docs/src/kernels.md index afe7f9a5..7c187f53 100644 --- a/docs/src/kernels.md +++ b/docs/src/kernels.md @@ -109,6 +109,8 @@ For more details, run Julia with `-g2`, or launch the kernel with `@oneapi debug The amount of detail depends on the debug level the kernel is compiled with, which defaults to Julia's `-g` level. At level 0 only the fact that an exception was thrown is reported; at level 1 (the default) also its type and reason; at level 2 also the work-item that threw it -and a backtrace. Only one exception is reported per synchronization. Exceptions in kernels -launched on an explicitly created `ZeCommandQueue` are not reported. +and a backtrace. Only one exception is reported per synchronization. If the task that +launched the kernel finishes without synchronizing, the exception is reported by the next +`synchronize()` of another task, unless something waited for the kernel's results first. +Exceptions in kernels launched on an explicitly created `ZeCommandQueue` are not reported. diff --git a/src/compiler/exceptions.jl b/src/compiler/exceptions.jl index 94a3f14d..a6f0f012 100644 --- a/src/compiler/exceptions.jl +++ b/src/compiler/exceptions.jl @@ -182,14 +182,8 @@ function release_report_slot(s::oneStream) end # Check whether a kernel on `s` threw an exception, returning it after resetting the report. -# The caller must hold `s.lock`. +# The caller must hold `s.lock`, and have checked that the stream is idle. function check_exception!(s::oneStream) - # Kernels may still write the report while the stream is busy. That is only the case - # after waiting for it if another task submitted work in the meantime, in which case the - # next synchronization reports the exception. - oneL0.unchecked_zeCommandListHostSynchronize(s.immediate, 0) == oneL0.RESULT_SUCCESS || - return nothing - slot = s.exceptions unsafe_load(convert(Ptr{Int32}, slot.report)) == 0 && return nothing # `status` report = unsafe_load(slot.report) diff --git a/src/compiler/execution.jl b/src/compiler/execution.jl index 69281739..50bb77c0 100644 --- a/src/compiler/execution.jl +++ b/src/compiler/execution.jl @@ -422,13 +422,13 @@ end check_error(s) direct |= !s.batching || s.idle || oneL0.sync_each_submission() batch = direct ? nothing : open_batch!(s) + register_pending!(s) if batch === nothing flush_batch!(s) append_launch!(s.immediate, kernel, groups) s.idle = false else append_launch!(batch.list, kernel, groups) - s.recorder === current_task() || register_batching!(s) root!(batch, roots) batch.launches += 1 batch.launches == BATCH_SIZE && flush_batch!(s) diff --git a/src/context.jl b/src/context.jl index 7b0c0808..7bbbb30a 100644 --- a/src/context.jl +++ b/src/context.jl @@ -282,7 +282,7 @@ mutable struct oneStream nbatches::Int events::Union{Nothing, oneL0.ZeEventPool} idle::Bool # submit the next launch directly - recorder::Union{Nothing, Task} # the last task to record a launch, if registered + recorder::Union{Nothing, WeakRef} # the last task to launch a kernel, if registered const retired::Vector{DataRef} # roots of executed batches, still to be released const failed::Vector{Batch} # possibly submitted, so never released # a failed submission, reported by the next synchronization @@ -460,53 +460,65 @@ function release_retired!(s::oneStream) return end -# streams that recorded launches since they were last synchronized, so that `synchronize` -# can submit what tasks recorded before finishing. the stream is referenced strongly, so -# that it doesn't get destroyed with work recorded. a stream's lock is taken before this -# one, and the stream's `recorder` is the task last to record on it. -const batching_streams = Base.IdSet{oneStream}() -const batching_streams_lock = ReentrantLock() - -function register_batching!(s::oneStream) - @lock batching_streams_lock push!(batching_streams, s) - s.recorder = current_task() +# streams that launched kernels since they were last synchronized, so that `synchronize` +# can submit what tasks recorded before finishing, and report exceptions their kernels threw. +# the stream is referenced strongly, so that it doesn't get destroyed with work recorded. a +# stream's lock is taken before this one, and the stream's `recorder` is the task last to +# launch on it, referenced weakly so that finished tasks and their results can be collected. +const pending_streams = Base.IdSet{oneStream}() +const pending_streams_lock = ReentrantLock() + +function register_pending!(s::oneStream) + task = current_task() + r = s.recorder + r !== nothing && r.value === task && return + @lock pending_streams_lock push!(pending_streams, s) + s.recorder = WeakRef(task) return end -# forget a stream once nothing it recorded is still to execute. streams with batches that -# failed to submit, which may still execute, are kept around. -function maybe_unregister_batching!(s::oneStream) +# Take the error to report once a stream was waited for: a failed submission, or else an +# exception a kernel threw. The latter stays pending while there's a submission error. A +# stream with nothing left to execute or report is forgotten by `adopt_orphans!`; one with +# batches that failed to submit, which may still execute, is kept around. +# The caller must hold `s.lock`. +function take_error!(s::oneStream) + retire!(s) + s.idle = true + s.error === nothing || return s.error + # Kernels may still write the report while the stream is busy. That is only the case + # after waiting for it if another task submitted work in the meantime, in which case the + # next synchronization reports the exception. + oneL0.unchecked_zeCommandListHostSynchronize(s.immediate, 0) == oneL0.RESULT_SUCCESS || + return nothing + err = check_exception!(s) if s.recorder !== nothing && s.batch === nothing && isempty(s.submitted) && isempty(s.failed) - @lock batching_streams_lock delete!(batching_streams, s) + @lock pending_streams_lock delete!(pending_streams, s) s.recorder = nothing end - return + return err end -# wait for the streams on which tasks that have finished were the last to record. other -# tasks' streams are left alone: their tasks synchronize them. +# whether the task last to launch on a stream has finished. the caller must hold `s.lock`. +function orphaned(s::oneStream) + r = s.recorder + return r !== nothing && (r.value === nothing || istaskdone(r.value)) +end + +# wait for the streams on which tasks that have finished were the last to launch, reporting +# what they left behind. other tasks' streams are left alone: their tasks synchronize them. function adopt_orphans!(ctx::ZeContext, dev::ZeDevice) - streams = @lock batching_streams_lock collect(batching_streams) + streams = @lock pending_streams_lock collect(pending_streams) for s in streams (s.ctx == ctx && s.dev == dev) || continue - orphaned = @lock s.lock begin - task = s.recorder - task !== nothing && istaskdone(task) - end - orphaned || continue - err = nothing try - drain(s) - catch e - err = e - end - @lock s.lock begin + drain(s; orphan=true) + catch err # reported here, as the task that would have isn't around anymore - err === nothing || (s.error = nothing) - maybe_unregister_batching!(s) + @lock s.lock s.error === err && (s.error = nothing) + rethrow() end - err === nothing || throw(err) end return end @@ -692,13 +704,9 @@ function synchronize_stream!(s::oneStream; blocking::Bool=false) s.mkl_dirty = false end err = @lock s.lock begin - retire!(s) - s.idle = true - maybe_unregister_batching!(s) - err = s.error + err = take_error!(s) s.error = nothing - # a kernel exception stays pending while there's a submission error to report - err === nothing ? check_exception!(s) : err + err end return err end diff --git a/src/memory.jl b/src/memory.jl index 10d2fe31..9a5a660d 100644 --- a/src/memory.jl +++ b/src/memory.jl @@ -85,8 +85,10 @@ Base.sizeof(managed::Managed) = sizeof(managed.mem) # wait for all work on a stream, possibly from another task, throwing exceptions its # kernels threw: whoever waits for memory is about to use what they computed. unlike # `synchronize`, this leaves `mkl_dirty` alone, which only the stream's own task may modify. -function drain(s::oneStream) +# with `orphan`, only do so while the stream is orphaned (see `adopt_orphans!`). +function drain(s::oneStream; orphan::Bool=false) @lock s.lock begin + orphan && !orphaned(s) && return check_error(s) flush_batch!(s) end @@ -94,9 +96,9 @@ function drain(s::oneStream) q = Base.@atomic s.queue q === nothing || oneL0.nonblocking_synchronize(q) err = @lock s.lock begin - retire!(s) - s.idle = true - check_exception!(s) + # a task that launched in the meantime reports what's left itself + orphan && !orphaned(s) && return + take_error!(s) end err === nothing || throw(err) return diff --git a/test/execution.jl b/test/execution.jl index 2fd2e432..fdf0165d 100644 --- a/test/execution.jl +++ b/test/execution.jl @@ -976,7 +976,7 @@ end take!(done) end stream = take!(streams) - batching && @test stream.recorder === owner + @test stream.recorder.value === owner wait(@async begin for _ in 1:3 @oneapi queue=stream ptr_increment_kernel(ptr) @@ -1096,7 +1096,7 @@ end 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 === err @test try synchronize(); nothing catch e; e end === nothing end @@ -1472,14 +1472,77 @@ end @test kernel_exception() === nothing # and of tasks that finished without synchronizing, by whoever adopts their stream - batching = oneAPI.global_stream(context(), device()).batching && - !oneL0.sync_each_submission() + adopted() = try synchronize(); nothing catch err; err end + wait(@async @oneapi store_checked(a, 2)) + @test adopted() isa KernelException wait(@async begin @oneapi store_checked(a, 1) @oneapi store_checked(a, 2) end) - err = try synchronize(); nothing catch err; err end - @test batching ? err isa KernelException : err === nothing + @test adopted() isa KernelException + @test adopted() === nothing + # unless something waited for their results first + wait(@async @oneapi store_checked(a, 2)) + @test_throws KernelException Array(a) + @test adopted() === nothing + task = @async begin + @oneapi store_checked(a, 2) + try Array(a); nothing catch err; err end + end + @test fetch(task) isa KernelException + @test adopted() === nothing + # a failed submission is reported first, the exception by the next synchronization + err = ErrorException("submission failed") + wait(@async begin + @oneapi store_checked(a, 2) + oneAPI.global_stream(context(), device()).error = err + end) + @test adopted() === err + @test adopted() isa KernelException + @test adopted() === nothing + # but not once a task that is still running launched on their stream, while adopting + ctx, dev = context(), device() + pool = oneL0.ZeEventPool(ctx, 1, dev; flags=oneL0.ZE_EVENT_POOL_FLAG_HOST_VISIBLE) + gate = pool[1] + # nothing may wait for the GPU while the gate is closed, as freeing memory does + GC.gc(true) + gc_enabled = GC.enable(false) + try + # the adopter's own stream is created first: doing so while the gate is closed + # makes other streams look idle on some stacks + ready, go = Channel{Nothing}(1), Channel{Nothing}(1) + adopter = @async begin + synchronize() + put!(ready, nothing) + take!(go) + adopted() + end + take!(ready) + # (with a kernel after the wait, as waits alone don't keep a stream busy) + stream = fetch(@async begin + oneL0.sync_each_submission(false) do + oneL0.execute!(oneAPI.global_stream(ctx, dev)) do list + oneL0.append_wait!(list, gate) + end + @oneapi store_checked(a, 1) + end + oneAPI.global_stream(ctx, dev) + end) + put!(go, nothing) + for _ in 1:100 + yield() + end + oneL0.sync_each_submission(false) do + @oneapi queue=stream store_checked(a, 2) + oneAPI.flush!(stream) + end + oneL0.signal(gate) + @test fetch(adopter) === nothing + finally + oneL0.signal(gate) + GC.enable(gc_enabled) + end + @test kernel_exception(stream) isa KernelException # explicitly created queues have no report queue = oneL0.ZeCommandQueue(context(), device(); From 24ec33b250e3289095e2f00f2258c1040db7fc74 Mon Sep 17 00:00:00 2001 From: Tim Besard Date: Thu, 22 Oct 2026 16:23:59 +0200 Subject: [PATCH 4/4] CI: test the core functionality without command batching Drivers without `zeCommandListImmediateAppendCommandListsExp`, like Aurora's, submit every launch directly, which takes different paths through synchronization and exception reporting. Run the execution and array tests that way on branch builds, to keep the load on the shared runner low. --- .buildkite/pipeline.yml | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index 0608f3da..21e309e4 100644 --- a/.buildkite/pipeline.yml +++ b/.buildkite/pipeline.yml @@ -105,6 +105,26 @@ steps: build.message !~ /\[only / && !build.pull_request.draft) timeout_in_minutes: 120 + # drivers without `zeCommandListImmediateAppendCommandListsExp` don't batch, so check + # that the core functionality behaves the same. not tested on pull requests. + - label: "Without batching" + plugins: + - JuliaCI/julia#v1: + version: "1.13" + - JuliaCI/julia-test#v1: + test_args: "--quickfail execution array" + command: | + julia --project=deps deps/build_ci.jl + env: + ONEAPI_COMMAND_BATCHING: '0' + agents: + queue: "oneapi" + if: | + build.message !~ /\[skip [^\]]*(tests|special)/ && + (build.message =~ /\[only [^\]]*(tests|special)/ || + build.message !~ /\[only / && build.pull_request.id == null) + timeout_in_minutes: 60 + env: JULIA_PKG_SERVER_REGISTRY_PREFERENCE: "eager" # OK to downloading JLLs from GitHub SECRET_CODECOV_TOKEN: "OYpS8fj3vGhj7iZf9vLAeapyxQNSOEW6mApcSvGboL9AlS+0nfOSFjFrIBNnIU0prxQQy1gR9AwR/JO1m2OFWeRhjYtkQPPhk4xVtSKmv0LLTL0snA8IohUopqfu722i7zLrPcz/A0LFIFsb0ey+oReJs2xnGOshNIJu4FDowUV3wmZvfKWNsSK4cGN+HFQ3387Ow4SsmiUr7oqh0iMBQNqaY8oZ2BY1dFOgPaOegIp70YEFRdJ8DKaLd7WGxFLY9oQEhZZdmx/zx0xo56/NGtDwVYkDPa4qPhJczDBoIn5XvcRiIW0VJ/MaRARxnpenBX5H6gwdcZYUGtjXWIRXBw==;U2FsdGVkX1/bZy1Bp4/dBH5scPpWqLKusXGvSkRGUa+1F7hi4P4Cu5a6GcfNIEvQr+bBj2VlZvqhNW0FAqN3QQ=="