https://github.com/jellytabby updated https://github.com/llvm/llvm-project/pull/218049
>From 2725719aba762d55c1d0c109e51d4a56c9b36d2c Mon Sep 17 00:00:00 2001 From: Sophia Herrmann <[email protected]> Date: Thu, 13 Aug 2026 16:47:23 -0700 Subject: [PATCH 1/2] add blocking semantics to LaunchKernel and Memcpy --- clang/lib/CodeGen/CGCUDANV.cpp | 4 +- clang/lib/Driver/ToolChains/Clang.cpp | 1 + .../languages/kernel/include/LanguageUtils.h | 39 ++++++ offload/languages/kernel/include/State.h | 2 +- .../languages/kernel/src/LanguageLaunch.cpp | 20 ++- .../languages/kernel/src/LanguageRuntime.cpp | 15 +- .../CUDA/blocking_stream_semantics.cu | 131 ++++++++++++++++++ offload/test/offloading/CUDA/stream_api.cu | 6 +- .../HIP/blocking_stream_semantics.hip | 125 +++++++++++++++++ offload/test/offloading/HIP/stream_api.hip | 6 +- 10 files changed, 336 insertions(+), 13 deletions(-) create mode 100644 offload/test/offloading/CUDA/blocking_stream_semantics.cu create mode 100644 offload/test/offloading/HIP/blocking_stream_semantics.hip diff --git a/clang/lib/CodeGen/CGCUDANV.cpp b/clang/lib/CodeGen/CGCUDANV.cpp index ac2ed9a109c4f..0b0693de0d24d 100644 --- a/clang/lib/CodeGen/CGCUDANV.cpp +++ b/clang/lib/CodeGen/CGCUDANV.cpp @@ -443,7 +443,9 @@ void CGNVCUDARuntime::emitDeviceStubBodyNew(CodeGenFunction &CGF, std::string KernelLaunchAPI = "LaunchKernel"; if (CGF.getLangOpts().GPUDefaultStream == LangOptions::GPUDefaultStreamKind::PerThread) { - if (CGF.getLangOpts().HIP) + if (CGF.getLangOpts().OffloadViaLLVM) + KernelLaunchAPI = KernelLaunchAPI + ""; + else if (CGF.getLangOpts().HIP) KernelLaunchAPI = KernelLaunchAPI + "_spt"; else if (CGF.getLangOpts().CUDA) KernelLaunchAPI = KernelLaunchAPI + "_ptsz"; diff --git a/clang/lib/Driver/ToolChains/Clang.cpp b/clang/lib/Driver/ToolChains/Clang.cpp index bd546dbb1fed2..0e572bd99934c 100644 --- a/clang/lib/Driver/ToolChains/Clang.cpp +++ b/clang/lib/Driver/ToolChains/Clang.cpp @@ -8444,6 +8444,7 @@ void Clang::ConstructJob(Compilation &C, const JobAction &JA, if (IsHIP) { CmdArgs.push_back("-fcuda-allow-variadic-functions"); + /// TODO: Why is this not forwarded when IsCUDA? Args.AddLastArg(CmdArgs, options::OPT_fgpu_default_stream_EQ); } diff --git a/offload/languages/kernel/include/LanguageUtils.h b/offload/languages/kernel/include/LanguageUtils.h index 72d1a3be6fb4c..cd893dcc60cc8 100644 --- a/offload/languages/kernel/include/LanguageUtils.h +++ b/offload/languages/kernel/include/LanguageUtils.h @@ -61,6 +61,45 @@ static inline StreamTy *toInternalStream(Stream_t Stream) { return reinterpret_cast<StreamTy *>(Stream); } +/// Wait for blocking streams before executing if we are legacy default stream. +static inline ol_result_t waitOnBlockingStreams() { + ol_device_handle_t Device = ThreadState::getDefaultDevice(); + if (!RuntimeState::hasLegacyDefaultStream(Device) || + RuntimeState::getBlockingStreams(Device).empty()) + return OL_SUCCESS; + StreamTy *DefaultStream = ThreadState::getDefaultStream(); + llvm::SmallVector<ol_event_handle_t, 8> Events; + for (StreamTy *BlockingStream : RuntimeState::getBlockingStreams(Device)) { + ol_event_handle_t Event = nullptr; + ol_result_t Result = + olCreateEvent(BlockingStream->Queue, OL_EVENT_FLAGS_NONE, &Event); + if (Result != OL_SUCCESS) + return Result; + Events.push_back(Event); + } + + return olWaitEvents(DefaultStream->Queue, Events.data(), Events.size()); +} + +/// Wait for the legacy default stream to complete before launching a kernel on +/// a blocking stream. +static inline ol_result_t waitOnLegacyDefaultStream(StreamTy *SourceStream, + ol_device_handle_t Device) { + if (!RuntimeState::hasLegacyDefaultStream(Device)) + return OL_SUCCESS; + + StreamTy *DefaultStream = ThreadState::getDefaultStream(); + assert(DefaultStream->Kind == llvm::offload::QueueKind::LegacyDefault && + "Default stream is not a legacy default stream"); + + ol_event_handle_t Event = nullptr; + ol_result_t Result = + olCreateEvent(DefaultStream->Queue, OL_EVENT_FLAGS_NONE, &Event); + if (Result != OL_SUCCESS) + return Result; + return olWaitEvents(SourceStream->Queue, &Event, 1); +} + /// Convert a Stream_t to an ol_queue_handle_t. static inline Error_t getQueueFromStream(Stream_t Stream, ol_queue_handle_t *Queue) { diff --git a/offload/languages/kernel/include/State.h b/offload/languages/kernel/include/State.h index a3c8712723a7a..1b42f4fbfcd19 100644 --- a/offload/languages/kernel/include/State.h +++ b/offload/languages/kernel/include/State.h @@ -156,7 +156,7 @@ struct StateTy { /// Return all explicitly created blocking streams for \p Device. SmallPtrSet<StreamTy *, 8> getBlockingStreams(ol_device_handle_t Device); - /// Return true if \p Device has an existing legacy default stream. + /// Return true if \p Device has an initialized (i.e. previously used) legacy default stream. bool hasLegacyDefaultStream(ol_device_handle_t Device); /// Create a stream for \p Device and register it with the process state. diff --git a/offload/languages/kernel/src/LanguageLaunch.cpp b/offload/languages/kernel/src/LanguageLaunch.cpp index edc1209516ad6..0ab4c6d665742 100644 --- a/offload/languages/kernel/src/LanguageLaunch.cpp +++ b/offload/languages/kernel/src/LanguageLaunch.cpp @@ -8,6 +8,7 @@ #include "LanguageLaunch.h" #include "LanguageUtils.h" +#include "OffloadAPI.h" #include "OffloadErrors.h" #include "State.h" #include "Stream.h" @@ -45,8 +46,21 @@ ol_result_t __llvmLaunchKernelImpl(const char *KernelID, dim3 GridDim, LaunchSizeArgs.GroupSize.z = BlockDim.z; LaunchSizeArgs.DynSharedMemory = DynamicSharedMem; - ol_queue_handle_t Queue = Stream ? reinterpret_cast<StreamTy *>(Stream)->Queue - : ThreadState.getDefaultQueue(); + StreamTy *LaunchStream = Stream ? reinterpret_cast<StreamTy *>(Stream) + : ThreadState::getDefaultStream(); + if (!LaunchStream || !RuntimeState::isStreamRegistered(LaunchStream) || + LaunchStream->Device != Device) + return &InvalidConfigurationError; + + if (LaunchStream->Kind == llvm::offload::QueueKind::LegacyDefault) { + ol_result_t Result = waitOnBlockingStreams(); + if (Result != OL_SUCCESS) + return Result; + } else if (LaunchStream->Kind == llvm::offload::QueueKind::ExplicitBlocking) { + ol_result_t Result = waitOnLegacyDefaultStream(LaunchStream, Device); + if (Result != OL_SUCCESS) + return Result; + } struct OffloadKernelArgs { void **Args; @@ -62,7 +76,7 @@ ol_result_t __llvmLaunchKernelImpl(const char *KernelID, dim3 GridDim, if (!OKA->Args[I] || OKA->ArgSizes[I] == 0) return &InvalidArgumentError; - return olLaunchKernel(Queue, Device, Kernel, &LaunchSizeArgs, + return olLaunchKernel(LaunchStream->Queue, Device, Kernel, &LaunchSizeArgs, /*Properties=*/nullptr, OKA->NumArgs, OKA->Args, OKA->ArgSizes); } diff --git a/offload/languages/kernel/src/LanguageRuntime.cpp b/offload/languages/kernel/src/LanguageRuntime.cpp index b4eb18c59de30..978c575cfcc9d 100644 --- a/offload/languages/kernel/src/LanguageRuntime.cpp +++ b/offload/languages/kernel/src/LanguageRuntime.cpp @@ -23,6 +23,7 @@ #include "Types.h" #include "OffloadAPI.h" +#include "llvm/ADT/SmallVector.h" #include <cassert> #include <cstdio> @@ -47,10 +48,13 @@ Error_t Free(void *DevPtr) { } Error_t Memcpy(void *Dst, const void *Src, size_t Size, MemcpyKind Kind) { - StateTy &State = StateTy::get(); - ThreadStateTy &ThreadState = ThreadStateTy::get(); - ol_queue_handle_t Queue = ThreadState.getDefaultQueue(); - + if (Kind != MemcpyHostToHost) { + ol_result_t Result = waitOnBlockingStreams(); + if (Result != OL_SUCCESS) + return convertAndSetLastError(Result); + } + ol_device_handle_t Device = ThreadState::getDefaultDevice(); + ol_queue_handle_t Queue = ThreadState::getDefaultQueue(); ol_result_t Result; switch (Kind) { case MemcpyHostToHost: { @@ -85,6 +89,9 @@ Error_t Memcpy(void *Dst, const void *Src, size_t Size, MemcpyKind Kind) { if (Result != OL_SUCCESS) return convertAndSetLastError(Result); + if (!Queue) + return convertAndSetLastError(Result); + Result = olSyncQueue(Queue); return convertAndSetLastError(Result); } diff --git a/offload/test/offloading/CUDA/blocking_stream_semantics.cu b/offload/test/offloading/CUDA/blocking_stream_semantics.cu new file mode 100644 index 0000000000000..83aa3383571f6 --- /dev/null +++ b/offload/test/offloading/CUDA/blocking_stream_semantics.cu @@ -0,0 +1,131 @@ +// clang-format off +// RUN: %clang++ %flags -foffload-via-llvm --offload-arch=native %s -o %t +// RUN: %t | %fcheck-generic --check-prefix=LEGACY +// RUN: %clang++ %flags -foffload-via-llvm --offload-arch=native %s -o %t -fopenmp +// RUN: %t | %fcheck-generic --check-prefix=LEGACY +// RUN: %clang++ %flags -foffload-via-llvm --offload-arch=native %s -o %t -fgpu-default-stream=per-thread +// RUN: %t | %fcheck-generic --check-prefix=PERTHREAD +// clang-format on + +// UNSUPPORTED: aarch64-unknown-linux-gnu +// UNSUPPORTED: x86_64-unknown-linux-gnu +// UNSUPPORTED: nvptx64-nvidia-cuda-LTO +// UNSUPPORTED: amdgcn-amd-amdhsa-LTO +// UNSUPPORTED: amdgpu-amd-amdhsa-LTO +// UNSUPPORTED: intelgpu + +#include <stdio.h> + +__global__ void delayedSetValue(int *Out, int Value) { + volatile unsigned long long Delay = 0; + for (unsigned I = 0; I < 1000000; ++I) + Delay += I; + if (Delay) + *Out = Value; +} + +__global__ void copyValue(int *In, int *Out) { *Out = *In; } + +__global__ void waitThenSetValue(int *Gate, int *Out, int Value) { + volatile int *VolatileGate = Gate; + for (unsigned I = 0; I < 100000000 && *VolatileGate == 0; ++I) + ; + *Out = Value; +} + +__global__ void copyValueAndRelease(int *In, int *Out, int *Gate) { + *Out = *In; + volatile int *VolatileGate = Gate; + *VolatileGate = 1; +} + +int main(int argc, char **argv) { + cudaStream_t BlockingStream = nullptr; + if (cudaStreamCreateWithFlags(&BlockingStream, cudaStreamDefault) != + cudaSuccess) + return 1; + cudaStream_t NonBlockingStream = nullptr; + if (cudaStreamCreateWithFlags(&NonBlockingStream, cudaStreamNonBlocking) != + cudaSuccess) + return 1; + + int *In = nullptr; + int *Out = nullptr; + int *Gate = nullptr; + if (cudaMalloc(&In, sizeof(int)) != cudaSuccess) + return 1; + if (cudaMalloc(&Out, sizeof(int)) != cudaSuccess) + return 1; + if (cudaMalloc(&Gate, sizeof(int)) != cudaSuccess) + return 1; + + int Initial = 0; + int Result = 0; + if (cudaMemcpy(In, &Initial, sizeof(int), cudaMemcpyHostToDevice) != + cudaSuccess) + return 1; + if (cudaMemcpy(Out, &Initial, sizeof(int), cudaMemcpyHostToDevice) != + cudaSuccess) + return 1; + + delayedSetValue<<<1, 1, 0, BlockingStream>>>(In, 99); + copyValue<<<1, 1>>>(In, Out); + if (cudaMemcpy(&Result, Out, sizeof(int), cudaMemcpyDeviceToHost) != + cudaSuccess) + return 1; + + printf("legacy default waited on blocking stream: %d\n", Result); + // LEGACY: legacy default waited on blocking stream: 99 + // PERTHREAD: legacy default waited on blocking stream: 0 + + Result = 0; + if (cudaMemcpy(Out, &Initial, sizeof(int), cudaMemcpyHostToDevice) != + cudaSuccess) + return 1; + + delayedSetValue<<<1, 1>>>(In, 123); + copyValue<<<1, 1, 0, BlockingStream>>>(In, Out); + if (cudaStreamSynchronize(BlockingStream) != cudaSuccess) + return 1; + if (cudaMemcpy(&Result, Out, sizeof(int), cudaMemcpyDeviceToHost) != + cudaSuccess) + return 1; + + printf("blocking stream waited on legacy default: %d\n", Result); + // LEGACY: blocking stream waited on legacy default: 123 + // PERTHREAD: blocking stream waited on legacy default: 99 + + Result = 0; + if (cudaMemcpy(In, &Initial, sizeof(int), cudaMemcpyHostToDevice) != + cudaSuccess) + return 1; + if (cudaMemcpy(Out, &Initial, sizeof(int), cudaMemcpyHostToDevice) != + cudaSuccess) + return 1; + if (cudaMemcpy(Gate, &Initial, sizeof(int), cudaMemcpyHostToDevice) != + cudaSuccess) + return 1; + + waitThenSetValue<<<1, 1>>>(Gate, In, 321); + copyValueAndRelease<<<1, 1, 0, NonBlockingStream>>>(In, Out, Gate); + if (cudaStreamSynchronize(NonBlockingStream) != cudaSuccess) + return 1; + if (cudaMemcpy(&Result, Out, sizeof(int), cudaMemcpyDeviceToHost) != + cudaSuccess) + return 1; + + printf("nonblocking stream did not wait on legacy default: %d\n", Result); + // LEGACY: nonblocking stream did not wait on legacy default: 0 + // PERTHREAD: nonblocking stream did not wait on legacy default: 0 + + if (cudaStreamDestroy(BlockingStream) != cudaSuccess) + return 1; + if (cudaStreamDestroy(NonBlockingStream) != cudaSuccess) + return 1; + if (cudaFree(In) != cudaSuccess) + return 1; + if (cudaFree(Out) != cudaSuccess) + return 1; + if (cudaFree(Gate) != cudaSuccess) + return 1; +} diff --git a/offload/test/offloading/CUDA/stream_api.cu b/offload/test/offloading/CUDA/stream_api.cu index 688c2f5a54c70..f11514c16aeaa 100644 --- a/offload/test/offloading/CUDA/stream_api.cu +++ b/offload/test/offloading/CUDA/stream_api.cu @@ -70,8 +70,6 @@ int main(int argc, char **argv) { if (cudaStreamSynchronize(Stream) != cudaSuccess) return 1; - if (cudaDeviceSynchronize() != cudaSuccess) - return 1; if (cudaMemcpy(&StreamResult, StreamPtr, sizeof(int), cudaMemcpyDeviceToHost) != cudaSuccess) return 1; @@ -86,6 +84,10 @@ int main(int argc, char **argv) { if (cudaStreamDestroy(Stream) != cudaSuccess) return 1; + if (cudaStreamDestroy(BlockingStream) != cudaSuccess) + return 1; + if (cudaStreamDestroy(NonBlockingStream) != cudaSuccess) + return 1; print_error("destroyed stream destroy", cudaStreamDestroy(Stream)); // CHECK: destroyed stream destroy value: 4 // CHECK: destroyed stream destroy name: cudaErrorInvalidResourceHandle diff --git a/offload/test/offloading/HIP/blocking_stream_semantics.hip b/offload/test/offloading/HIP/blocking_stream_semantics.hip new file mode 100644 index 0000000000000..8c28c7b1b2ee8 --- /dev/null +++ b/offload/test/offloading/HIP/blocking_stream_semantics.hip @@ -0,0 +1,125 @@ +// clang-format off +// RUN: %clang++ %flags -foffload-via-llvm --offload-arch=native %s -o %t +// RUN: %t | %fcheck-generic --check-prefix=LEGACY +// RUN: %clang++ %flags -foffload-via-llvm --offload-arch=native %s -o %t -fopenmp +// RUN: %t | %fcheck-generic --check-prefix=LEGACY +// RUN: %clang++ %flags -foffload-via-llvm --offload-arch=native %s -o %t -fgpu-default-stream=per-thread +// RUN: %t | %fcheck-generic --check-prefix=PERTHREAD +// clang-format on + +// UNSUPPORTED: aarch64-unknown-linux-gnu +// UNSUPPORTED: x86_64-unknown-linux-gnu +// UNSUPPORTED: nvptx64-nvidia-cuda-LTO +// UNSUPPORTED: amdgcn-amd-amdhsa-LTO +// UNSUPPORTED: amdgpu-amd-amdhsa-LTO +// UNSUPPORTED: intelgpu + +#include <stdio.h> + +__global__ void delayedSetValue(int *Out, int Value) { + volatile unsigned long long Delay = 0; + for (unsigned I = 0; I < 1000000; ++I) + Delay += I; + if (Delay) + *Out = Value; +} + +__global__ void copyValue(int *In, int *Out) { *Out = *In; } + +__global__ void waitThenSetValue(int *Gate, int *Out, int Value) { + volatile int *VolatileGate = Gate; + for (unsigned I = 0; I < 100000000 && *VolatileGate == 0; ++I) + ; + *Out = Value; +} + +__global__ void copyValueAndRelease(int *In, int *Out, int *Gate) { + *Out = *In; + volatile int *VolatileGate = Gate; + *VolatileGate = 1; +} + +int main(int argc, char **argv) { + hipStream_t BlockingStream = nullptr; + if (hipStreamCreateWithFlags(&BlockingStream, hipStreamDefault) != hipSuccess) + return 1; + hipStream_t NonBlockingStream = nullptr; + if (hipStreamCreateWithFlags(&NonBlockingStream, hipStreamNonBlocking) != + hipSuccess) + return 1; + + int *In = nullptr; + int *Out = nullptr; + int *Gate = nullptr; + if (hipMalloc(&In, sizeof(int)) != hipSuccess) + return 1; + if (hipMalloc(&Out, sizeof(int)) != hipSuccess) + return 1; + if (hipMalloc(&Gate, sizeof(int)) != hipSuccess) + return 1; + + int Initial = 0; + int Result = 0; + if (hipMemcpy(In, &Initial, sizeof(int), hipMemcpyHostToDevice) != hipSuccess) + return 1; + if (hipMemcpy(Out, &Initial, sizeof(int), hipMemcpyHostToDevice) != + hipSuccess) + return 1; + + delayedSetValue<<<1, 1, 0, BlockingStream>>>(In, 99); + copyValue<<<1, 1>>>(In, Out); + if (hipMemcpy(&Result, Out, sizeof(int), hipMemcpyDeviceToHost) != hipSuccess) + return 1; + + printf("legacy default waited on blocking stream: %d\n", Result); + // LEGACY: legacy default waited on blocking stream: 99 + // PERTHREAD: legacy default waited on blocking stream: 0 + + Result = 0; + if (hipMemcpy(Out, &Initial, sizeof(int), hipMemcpyHostToDevice) != + hipSuccess) + return 1; + + delayedSetValue<<<1, 1>>>(In, 123); + copyValue<<<1, 1, 0, BlockingStream>>>(In, Out); + if (hipStreamSynchronize(BlockingStream) != hipSuccess) + return 1; + if (hipMemcpy(&Result, Out, sizeof(int), hipMemcpyDeviceToHost) != hipSuccess) + return 1; + + printf("blocking stream waited on legacy default: %d\n", Result); + // LEGACY: blocking stream waited on legacy default: 123 + // PERTHREAD: blocking stream waited on legacy default: 99 + + Result = 0; + if (hipMemcpy(In, &Initial, sizeof(int), hipMemcpyHostToDevice) != hipSuccess) + return 1; + if (hipMemcpy(Out, &Initial, sizeof(int), hipMemcpyHostToDevice) != + hipSuccess) + return 1; + if (hipMemcpy(Gate, &Initial, sizeof(int), hipMemcpyHostToDevice) != + hipSuccess) + return 1; + + waitThenSetValue<<<1, 1>>>(Gate, In, 321); + copyValueAndRelease<<<1, 1, 0, NonBlockingStream>>>(In, Out, Gate); + if (hipStreamSynchronize(NonBlockingStream) != hipSuccess) + return 1; + if (hipMemcpy(&Result, Out, sizeof(int), hipMemcpyDeviceToHost) != hipSuccess) + return 1; + + printf("nonblocking stream did not wait on legacy default: %d\n", Result); + // LEGACY: nonblocking stream did not wait on legacy default: 0 + // PERTHREAD: nonblocking stream did not wait on legacy default: 0 + + if (hipStreamDestroy(BlockingStream) != hipSuccess) + return 1; + if (hipStreamDestroy(NonBlockingStream) != hipSuccess) + return 1; + if (hipFree(In) != hipSuccess) + return 1; + if (hipFree(Out) != hipSuccess) + return 1; + if (hipFree(Gate) != hipSuccess) + return 1; +} diff --git a/offload/test/offloading/HIP/stream_api.hip b/offload/test/offloading/HIP/stream_api.hip index 954344b6f7ce2..d83163c6df8f0 100644 --- a/offload/test/offloading/HIP/stream_api.hip +++ b/offload/test/offloading/HIP/stream_api.hip @@ -69,8 +69,6 @@ int main(int argc, char **argv) { if (hipStreamSynchronize(Stream) != hipSuccess) return 1; - if (hipDeviceSynchronize() != hipSuccess) - return 1; if (hipMemcpy(&StreamResult, StreamPtr, sizeof(int), hipMemcpyDeviceToHost) != hipSuccess) return 1; @@ -85,6 +83,10 @@ int main(int argc, char **argv) { if (hipStreamDestroy(Stream) != hipSuccess) return 1; + if (hipStreamDestroy(BlockingStream) != hipSuccess) + return 1; + if (hipStreamDestroy(NonBlockingStream) != hipSuccess) + return 1; print_error("destroyed stream destroy", hipStreamDestroy(Stream)); // CHECK: destroyed stream destroy value: 4 // CHECK: destroyed stream destroy name: hipErrorInvalidResourceHandle >From 58082a6cd5cfb352b6850a59c9005c4840f045fd Mon Sep 17 00:00:00 2001 From: Sophia Herrmann <[email protected]> Date: Wed, 19 Aug 2026 17:42:04 -0700 Subject: [PATCH 2/2] add event cleanup --- offload/languages/kernel/CMakeLists.txt | 1 + .../languages/kernel/include/LanguageUtils.h | 68 +++++++++------ offload/languages/kernel/include/State.h | 3 +- offload/languages/kernel/include/Stream.h | 19 +++++ .../languages/kernel/src/LanguageLaunch.cpp | 15 ++-- .../languages/kernel/src/LanguageRuntime.cpp | 42 +++++----- offload/languages/kernel/src/State.cpp | 8 +- offload/languages/kernel/src/Stream.cpp | 83 +++++++++++++++++++ 8 files changed, 179 insertions(+), 60 deletions(-) create mode 100644 offload/languages/kernel/src/Stream.cpp diff --git a/offload/languages/kernel/CMakeLists.txt b/offload/languages/kernel/CMakeLists.txt index a103a2eaa5d6c..e35ccd4f79f2a 100644 --- a/offload/languages/kernel/CMakeLists.txt +++ b/offload/languages/kernel/CMakeLists.txt @@ -64,6 +64,7 @@ add_library( src/LanguageLaunch.cpp src/LanguageRegistration.cpp src/State.cpp + src/Stream.cpp ) if(LLVM_LINK_LLVM_DYLIB) diff --git a/offload/languages/kernel/include/LanguageUtils.h b/offload/languages/kernel/include/LanguageUtils.h index cd893dcc60cc8..545afc5bf87a4 100644 --- a/offload/languages/kernel/include/LanguageUtils.h +++ b/offload/languages/kernel/include/LanguageUtils.h @@ -13,6 +13,7 @@ #include "OffloadAPI.h" #include "State.h" #include "Stream.h" +#include "llvm/ADT/SmallVector.h" namespace llvm { namespace offload { @@ -61,55 +62,70 @@ static inline StreamTy *toInternalStream(Stream_t Stream) { return reinterpret_cast<StreamTy *>(Stream); } +static inline ol_result_t +syncAndDestroyEvents(llvm::SmallVectorImpl<ol_event_handle_t> &Events) { + ol_result_t FirstError = OL_SUCCESS; + for (ol_event_handle_t Event : Events) { + if (!Event) + continue; + + ol_result_t SyncResult = olSyncEvent(Event); + if (FirstError == OL_SUCCESS && SyncResult != OL_SUCCESS) + FirstError = SyncResult; + + ol_result_t DestroyResult = olDestroyEvent(Event); + if (FirstError == OL_SUCCESS && DestroyResult != OL_SUCCESS) + FirstError = DestroyResult; + } + Events.clear(); + return FirstError; +} + /// Wait for blocking streams before executing if we are legacy default stream. -static inline ol_result_t waitOnBlockingStreams() { - ol_device_handle_t Device = ThreadState::getDefaultDevice(); - if (!RuntimeState::hasLegacyDefaultStream(Device) || - RuntimeState::getBlockingStreams(Device).empty()) +static inline ol_result_t waitOnBlockingStreams(StateTy &State, + ThreadStateTy &ThreadState) { + ol_device_handle_t Device = ThreadState.getDefaultDevice(); + SmallPtrSet<StreamTy *, 8> BlockingStreams = State.getBlockingStreams(Device); + if (!State.hasLegacyDefaultStream(Device) || BlockingStreams.empty()) return OL_SUCCESS; - StreamTy *DefaultStream = ThreadState::getDefaultStream(); - llvm::SmallVector<ol_event_handle_t, 8> Events; - for (StreamTy *BlockingStream : RuntimeState::getBlockingStreams(Device)) { + + StreamTy *DefaultStream = ThreadState.getDefaultStream(); + SmallVector<ol_event_handle_t, 8> Events; + for (StreamTy *BlockingStream : BlockingStreams) { ol_event_handle_t Event = nullptr; ol_result_t Result = olCreateEvent(BlockingStream->Queue, OL_EVENT_FLAGS_NONE, &Event); - if (Result != OL_SUCCESS) + if (Result != OL_SUCCESS) { + syncAndDestroyEvents(Events); return Result; + } Events.push_back(Event); } - return olWaitEvents(DefaultStream->Queue, Events.data(), Events.size()); + return DefaultStream->waitOnAndTrackDependencyEvents(Events); } /// Wait for the legacy default stream to complete before launching a kernel on /// a blocking stream. -static inline ol_result_t waitOnLegacyDefaultStream(StreamTy *SourceStream, +static inline ol_result_t waitOnLegacyDefaultStream(StateTy &State, + ThreadStateTy &ThreadState, + StreamTy *SourceStream, ol_device_handle_t Device) { - if (!RuntimeState::hasLegacyDefaultStream(Device)) + if (!State.hasLegacyDefaultStream(Device)) return OL_SUCCESS; - StreamTy *DefaultStream = ThreadState::getDefaultStream(); - assert(DefaultStream->Kind == llvm::offload::QueueKind::LegacyDefault && + StreamTy *DefaultStream = ThreadState.getDefaultStream(); + assert(DefaultStream->Kind == QueueKind::LegacyDefault && "Default stream is not a legacy default stream"); ol_event_handle_t Event = nullptr; + SmallVector<ol_event_handle_t, 8> Events; ol_result_t Result = olCreateEvent(DefaultStream->Queue, OL_EVENT_FLAGS_NONE, &Event); if (Result != OL_SUCCESS) return Result; - return olWaitEvents(SourceStream->Queue, &Event, 1); -} - -/// Convert a Stream_t to an ol_queue_handle_t. -static inline Error_t getQueueFromStream(Stream_t Stream, - ol_queue_handle_t *Queue) { - if (!Stream) - return ErrorInvalidValue; - - // TODO: add proper DEBUG/assert guarded checks - StreamTy *InternalStream = toInternalStream(Stream); - *Queue = InternalStream->Queue; - return Success; + Events.push_back(Event); + return SourceStream->waitOnAndTrackDependencyEvents(Events); } } // namespace offload diff --git a/offload/languages/kernel/include/State.h b/offload/languages/kernel/include/State.h index 1b42f4fbfcd19..8d5d75e64a8bf 100644 --- a/offload/languages/kernel/include/State.h +++ b/offload/languages/kernel/include/State.h @@ -156,7 +156,8 @@ struct StateTy { /// Return all explicitly created blocking streams for \p Device. SmallPtrSet<StreamTy *, 8> getBlockingStreams(ol_device_handle_t Device); - /// Return true if \p Device has an initialized (i.e. previously used) legacy default stream. + /// Return true if \p Device has an initialized (i.e. previously used) legacy + /// default stream. bool hasLegacyDefaultStream(ol_device_handle_t Device); /// Create a stream for \p Device and register it with the process state. diff --git a/offload/languages/kernel/include/Stream.h b/offload/languages/kernel/include/Stream.h index e66fc7000281d..903a69b9e77f4 100644 --- a/offload/languages/kernel/include/Stream.h +++ b/offload/languages/kernel/include/Stream.h @@ -10,6 +10,10 @@ #define LLVM_OFFLOAD_LANGUAGES_KERNEL_INCLUDE_STREAM_H #include "OffloadAPI.h" +#include "llvm/ADT/ArrayRef.h" +#include "llvm/ADT/SmallVector.h" +#include <cstddef> +#include <mutex> namespace llvm { namespace offload { @@ -22,9 +26,24 @@ enum class QueueKind { }; struct StreamTy { + StreamTy(ol_queue_handle_t Queue, ol_device_handle_t Device, QueueKind Kind) + : Queue(Queue), Device(Device), Kind(Kind) {} + + ol_result_t + waitOnAndTrackDependencyEvents(SmallVectorImpl<ol_event_handle_t> &Events); + ol_result_t sync(); + ol_queue_handle_t Queue = nullptr; ol_device_handle_t Device = nullptr; QueueKind Kind = QueueKind::ExplicitBlocking; + +private: + ol_result_t reclaimDependencyEventsLocked(); + + static constexpr size_t MaxPendingDependencyEvents = 64; + + std::mutex DependencyEventsLock; + SmallVector<ol_event_handle_t, 8> DependencyEvents; }; } // namespace offload diff --git a/offload/languages/kernel/src/LanguageLaunch.cpp b/offload/languages/kernel/src/LanguageLaunch.cpp index 0ab4c6d665742..14614badd33c2 100644 --- a/offload/languages/kernel/src/LanguageLaunch.cpp +++ b/offload/languages/kernel/src/LanguageLaunch.cpp @@ -47,17 +47,16 @@ ol_result_t __llvmLaunchKernelImpl(const char *KernelID, dim3 GridDim, LaunchSizeArgs.DynSharedMemory = DynamicSharedMem; StreamTy *LaunchStream = Stream ? reinterpret_cast<StreamTy *>(Stream) - : ThreadState::getDefaultStream(); - if (!LaunchStream || !RuntimeState::isStreamRegistered(LaunchStream) || - LaunchStream->Device != Device) - return &InvalidConfigurationError; + : ThreadState.getDefaultStream(); - if (LaunchStream->Kind == llvm::offload::QueueKind::LegacyDefault) { - ol_result_t Result = waitOnBlockingStreams(); + // TODO:Add proper DEBUG/assert check for Stream validity + if (LaunchStream->Kind == QueueKind::LegacyDefault) { + ol_result_t Result = waitOnBlockingStreams(State, ThreadState); if (Result != OL_SUCCESS) return Result; - } else if (LaunchStream->Kind == llvm::offload::QueueKind::ExplicitBlocking) { - ol_result_t Result = waitOnLegacyDefaultStream(LaunchStream, Device); + } else if (LaunchStream->Kind == QueueKind::ExplicitBlocking) { + ol_result_t Result = + waitOnLegacyDefaultStream(State, ThreadState, LaunchStream, Device); if (Result != OL_SUCCESS) return Result; } diff --git a/offload/languages/kernel/src/LanguageRuntime.cpp b/offload/languages/kernel/src/LanguageRuntime.cpp index 978c575cfcc9d..cf6bf084f9f4f 100644 --- a/offload/languages/kernel/src/LanguageRuntime.cpp +++ b/offload/languages/kernel/src/LanguageRuntime.cpp @@ -48,36 +48,32 @@ Error_t Free(void *DevPtr) { } Error_t Memcpy(void *Dst, const void *Src, size_t Size, MemcpyKind Kind) { + StateTy &State = StateTy::get(); + ThreadStateTy &ThreadState = ThreadStateTy::get(); if (Kind != MemcpyHostToHost) { - ol_result_t Result = waitOnBlockingStreams(); + ol_result_t Result = waitOnBlockingStreams(State, ThreadState); if (Result != OL_SUCCESS) return convertAndSetLastError(Result); } - ol_device_handle_t Device = ThreadState::getDefaultDevice(); - ol_queue_handle_t Queue = ThreadState::getDefaultQueue(); + ol_device_handle_t Device = ThreadState.getDefaultDevice(); + ol_device_handle_t Host = State.getHostDevice(); + StreamTy *DefaultStream = ThreadState.getDefaultStream(); + ol_queue_handle_t Queue = DefaultStream->Queue; ol_result_t Result; switch (Kind) { case MemcpyHostToHost: { - ol_device_handle_t Host = State.getHostDevice(); Result = olMemcpy(nullptr, Dst, Host, const_cast<void *>(Src), Host, Size); break; } case MemcpyHostToDevice: { - ol_device_handle_t Device = ThreadState.getDefaultDevice(); - ol_device_handle_t Host = State.getHostDevice(); Result = olMemcpy(Queue, Dst, Device, const_cast<void *>(Src), Host, Size); break; } case MemcpyDeviceToHost: { - ol_device_handle_t Device = ThreadState.getDefaultDevice(); - ol_device_handle_t Host = State.getHostDevice(); - Result = olMemcpy(Queue, Dst, Host, const_cast<void *>(Src), Device, Size); break; } case MemcpyDeviceToDevice: { - ol_device_handle_t Device = ThreadState.getDefaultDevice(); - Result = olMemcpy(Queue, Dst, Device, const_cast<void *>(Src), Device, Size); break; @@ -89,19 +85,16 @@ Error_t Memcpy(void *Dst, const void *Src, size_t Size, MemcpyKind Kind) { if (Result != OL_SUCCESS) return convertAndSetLastError(Result); - if (!Queue) - return convertAndSetLastError(Result); - - Result = olSyncQueue(Queue); + Result = DefaultStream->sync(); return convertAndSetLastError(Result); } Error_t DeviceSynchronize() { // TODO: This is not correct. We likely want to pipe this through to the // plugins. - ThreadStateTy &ThreadState = ThreadStateTy::get(); - ol_queue_handle_t Queue = ThreadState.getDefaultQueue(); - ol_result_t Result = olSyncQueue(Queue); + StreamTy *DefaultStream = ThreadStateTy::get().getDefaultStream(); + ol_result_t Result = + DefaultStream ? DefaultStream->sync() : olSyncQueue(nullptr); return convertAndSetLastError(Result); } @@ -205,11 +198,14 @@ Error_t StreamDestroy(Stream_t Stream) { } Error_t StreamSynchronize(Stream_t Stream) { - ol_queue_handle_t Queue; - Error_t Err = getQueueFromStream(Stream, &Queue); - if (Err != Success) - return setLastError(Err); - ol_result_t Result = olSyncQueue(Queue); + if (!Stream) + return setLastError(ErrorInvalidValue); + + StreamTy *InternalStream = toInternalStream(Stream); + if (!StateTy::get().isStreamRegistered(InternalStream)) + return setLastError(ErrorInvalidResourceHandle); + + ol_result_t Result = InternalStream->sync(); return convertAndSetLastError(Result); } diff --git a/offload/languages/kernel/src/State.cpp b/offload/languages/kernel/src/State.cpp index b5a2e9f4fdf9a..819b23be44fae 100644 --- a/offload/languages/kernel/src/State.cpp +++ b/offload/languages/kernel/src/State.cpp @@ -78,7 +78,7 @@ static void destroyStreamHandle(StreamTy *&Stream) { if (!Stream) return; - olSyncQueue(Stream->Queue); + (void)Stream->sync(); olDestroyQueue(Stream->Queue); delete Stream; Stream = nullptr; @@ -320,8 +320,12 @@ ol_result_t StateTy::destroyStream(StreamTy *Stream) { if (!isStreamRegistered(Stream)) return &InvalidStreamError; + ol_result_t Result = Stream->sync(); + if (Result != OL_SUCCESS) + return Result; + removeStream(Stream); - ol_result_t Result = olDestroyQueue(Stream->Queue); + Result = olDestroyQueue(Stream->Queue); delete Stream; return Result; } diff --git a/offload/languages/kernel/src/Stream.cpp b/offload/languages/kernel/src/Stream.cpp new file mode 100644 index 0000000000000..185a599667a83 --- /dev/null +++ b/offload/languages/kernel/src/Stream.cpp @@ -0,0 +1,83 @@ +//===-- Stream.cpp - Kernel language stream state -------------------------===// +// +// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions. +// See https://llvm.org/LICENSE.txt for license information. +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception +// +//===----------------------------------------------------------------------===// + +#include "Stream.h" +#include "OffloadAPI.h" +#include "llvm/ADT/ArrayRef.h" +#include "llvm/ADT/SmallVector.h" + +#include <mutex> +#include <utility> + +using namespace llvm; +using namespace offload; + +static ol_result_t syncAndDestroyEvents(ArrayRef<ol_event_handle_t> Events) { + ol_result_t FirstError = OL_SUCCESS; + for (ol_event_handle_t Event : Events) { + if (!Event) + continue; + + ol_result_t SyncResult = olSyncEvent(Event); + if (FirstError == OL_SUCCESS && SyncResult != OL_SUCCESS) + FirstError = SyncResult; + + ol_result_t DestroyResult = olDestroyEvent(Event); + if (FirstError == OL_SUCCESS && DestroyResult != OL_SUCCESS) + FirstError = DestroyResult; + } + return FirstError; +} + +ol_result_t StreamTy::waitOnAndTrackDependencyEvents( + SmallVectorImpl<ol_event_handle_t> &Events) { + if (Events.empty()) + return OL_SUCCESS; + + ol_result_t WaitResult = olWaitEvents(Queue, Events.data(), Events.size()); + if (WaitResult != OL_SUCCESS) { + syncAndDestroyEvents(Events); + return WaitResult; + } + + std::lock_guard<std::mutex> LG(DependencyEventsLock); + DependencyEvents.append(Events.begin(), Events.end()); + ol_result_t ReclaimResult = OL_SUCCESS; + if (DependencyEvents.size() >= MaxPendingDependencyEvents) { + ReclaimResult = olSyncQueue(Queue); + if (ReclaimResult == OL_SUCCESS) + ReclaimResult = reclaimDependencyEventsLocked(); + } + return ReclaimResult; +} + +ol_result_t StreamTy::sync() { + std::lock_guard<std::mutex> LG(DependencyEventsLock); + ol_result_t Result = olSyncQueue(Queue); + if (Result != OL_SUCCESS) + return Result; + return reclaimDependencyEventsLocked(); +} + +ol_result_t StreamTy::reclaimDependencyEventsLocked() { + if (DependencyEvents.empty()) + return OL_SUCCESS; + + ol_result_t FirstError = OL_SUCCESS; + SmallVector<ol_event_handle_t, 8> RemainingEvents; + for (ol_event_handle_t Event : DependencyEvents) { + ol_result_t Result = olDestroyEvent(Event); + if (Result != OL_SUCCESS) { + if (FirstError == OL_SUCCESS) + FirstError = Result; + RemainingEvents.push_back(Event); + } + } + DependencyEvents = std::move(RemainingEvents); + return FirstError; +} _______________________________________________ cfe-commits mailing list [email protected] https://lists.llvm.org/cgi-bin/mailman/listinfo/cfe-commits
