diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml index 0608f3daf..21e309e45 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==" diff --git a/docs/src/api/context.md b/docs/src/api/context.md index f4676c06e..5207a75a3 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/api/kernels.md b/docs/src/api/kernels.md index 8a525fe73..2c3d64784 100644 --- a/docs/src/api/kernels.md +++ b/docs/src/api/kernels.md @@ -20,13 +20,22 @@ 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; 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 a9eeb46f9..7c187f530 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,35 @@ 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. 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/compilation.jl b/src/compiler/compilation.jl index a78b01e3f..ef456c87e 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/exceptions.jl b/src/compiler/exceptions.jl new file mode 100644 index 000000000..a6f0f0127 --- /dev/null +++ b/src/compiler/exceptions.jl @@ -0,0 +1,207 @@ +# 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`, and have checked that the stream is idle. +function check_exception!(s::oneStream) + 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 f9660c68f..50bb77c08 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,12 +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 (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 @@ -219,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))...) @@ -239,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 @@ -417,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 abbf3a408..7bbbb30a3 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. @@ -276,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 @@ -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 @@ -449,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 + +# 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 record. other -# tasks' streams are left alone: their tasks synchronize them. +# 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 @@ -666,7 +689,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 @@ -681,10 +704,7 @@ 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 err end diff --git a/src/device/quirks.jl b/src/device/quirks.jl index 942fd1a31..5f701e562 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 c7551b9b1..3ea9be8ac 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 4afeabdf2..9a5a660d2 100644 --- a/src/memory.jl +++ b/src/memory.jl @@ -82,20 +82,25 @@ 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. -function drain(s::oneStream) +# 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. +# 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 oneL0.nonblocking_synchronize(s.immediate) q = Base.@atomic s.queue q === nothing || oneL0.nonblocking_synchronize(q) - @lock s.lock begin - retire!(s) - s.idle = true + err = @lock s.lock begin + # 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 end drain(q::ZeCommandQueue) = oneL0.nonblocking_synchronize(q) diff --git a/src/oneAPI.jl b/src/oneAPI.jl index fb8ddbfd4..e9d3a88af 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 42a0fef57..fdf0165d2 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() @@ -58,6 +63,16 @@ end 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() @inferred foo() @@ -961,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) @@ -1081,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 @@ -1270,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] @@ -1319,6 +1335,249 @@ 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 + 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) + @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(); + 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,