[llvm-branch-commits] [clang] [llvm] [Offload][Lang] Add blocking to LaunchKernel and Memcpy (PR #218049)
Sophia Herrmann via llvm-branch-commits
llvm-branch-commits at lists.llvm.org
Tue Aug 25 15:56:27 PDT 2026
https://github.com/jellytabby updated https://github.com/llvm/llvm-project/pull/218049
>From f783fe34b0c824569fe33653c26d4b5f84eafae5 Mon Sep 17 00:00:00 2001
From: Sophia Herrmann <herrmann15 at llnl.gov>
Date: Thu, 13 Aug 2026 16:38:44 -0700
Subject: [PATCH 1/3] add StreamCreateWithFlags
---
.../include/kernel/DefineLanguageNames.inc | 3 +++
.../include/kernel/LanguageRuntime.h | 11 +++++++-
.../include/kernel/UndefineLanguageNames.inc | 4 ++-
.../languages/kernel/src/LanguageRuntime.cpp | 27 ++++++++++++++++++-
offload/test/offloading/CUDA/stream_api.cu | 27 +++++++++++++++++++
offload/test/offloading/HIP/stream_api.hip | 26 ++++++++++++++++++
6 files changed, 95 insertions(+), 3 deletions(-)
diff --git a/offload/languages/include/kernel/DefineLanguageNames.inc b/offload/languages/include/kernel/DefineLanguageNames.inc
index 790fd3ad1a359..f4732002c1b52 100644
--- a/offload/languages/include/kernel/DefineLanguageNames.inc
+++ b/offload/languages/include/kernel/DefineLanguageNames.inc
@@ -43,6 +43,9 @@
#define FreeHost COMBINE(LANGUAGE, FreeHost)
#define GetDeviceProperties COMBINE(LANGUAGE, GetDeviceProperties)
#define Stream_t COMBINE(LANGUAGE, Stream_t)
+#define StreamDefault COMBINE(LANGUAGE, StreamDefault)
+#define StreamNonBlocking COMBINE(LANGUAGE, StreamNonBlocking)
#define StreamCreate COMBINE(LANGUAGE, StreamCreate)
+#define StreamCreateWithFlags COMBINE(LANGUAGE, StreamCreateWithFlags)
#define StreamDestroy COMBINE(LANGUAGE, StreamDestroy)
#define StreamSynchronize COMBINE(LANGUAGE, StreamSynchronize)
diff --git a/offload/languages/include/kernel/LanguageRuntime.h b/offload/languages/include/kernel/LanguageRuntime.h
index 3fbc4c0af3e2c..b25e735004ee9 100644
--- a/offload/languages/include/kernel/LanguageRuntime.h
+++ b/offload/languages/include/kernel/LanguageRuntime.h
@@ -40,13 +40,20 @@ enum MemcpyKind {
MemcpyDefault = 4
};
-enum HostAllocFlags : unsigned int {
+/// Flags passed to HostAlloc
+enum : unsigned int {
HostAllocDefault = 0x00,
HostAllocPortable = 0x01,
HostAllocMapped = 0x02,
HostAllocWriteCombined = 0x04,
};
+/// Flags passed to StreamCreateWithFlags
+enum : unsigned int {
+ StreamDefault = 0x00,
+ StreamNonBlocking = 0x01,
+};
+
typedef struct Stream_st *Stream_t;
/// Malloc, with type template overlay.
@@ -99,6 +106,8 @@ Error_t GetDeviceProperties(DeviceProp_t *DeviceProp, int DeviceNo);
Error_t StreamCreate(Stream_t *stream);
+Error_t StreamCreateWithFlags(Stream_t *stream, unsigned int flags);
+
Error_t StreamDestroy(Stream_t stream);
Error_t StreamSynchronize(Stream_t stream);
diff --git a/offload/languages/include/kernel/UndefineLanguageNames.inc b/offload/languages/include/kernel/UndefineLanguageNames.inc
index cd0e6b33aafc6..0ec241ead0434 100644
--- a/offload/languages/include/kernel/UndefineLanguageNames.inc
+++ b/offload/languages/include/kernel/UndefineLanguageNames.inc
@@ -32,7 +32,6 @@
#undef GetDeviceCount
#undef SetDevice
#undef HostAlloc
-#undef HostAllocFlags
#undef HostAllocDefault
#undef HostAllocPortable
#undef HostAllocMapped
@@ -41,6 +40,9 @@
#undef FreeHost
#undef GetDeviceProperties
#undef Stream_t
+#undef StreamDefault
+#undef StreamNonBlocking
#undef StreamCreate
+#undef StreamCreateWithFlags
#undef StreamDestroy
#undef StreamSynchronize
diff --git a/offload/languages/kernel/src/LanguageRuntime.cpp b/offload/languages/kernel/src/LanguageRuntime.cpp
index bf1b135aa7d7c..6dab19fc2fd8c 100644
--- a/offload/languages/kernel/src/LanguageRuntime.cpp
+++ b/offload/languages/kernel/src/LanguageRuntime.cpp
@@ -152,9 +152,12 @@ Error_t GetDeviceProperties(DeviceProp_t *DeviceProp, int DeviceNo) {
}
Error_t StreamCreate(Stream_t *Stream) {
- StreamTy *StreamObj = nullptr;
+ if (!Stream)
+ return setLastError(ErrorInvalidValue);
+
StateTy &State = StateTy::get();
ThreadStateTy &ThreadState = ThreadStateTy::get();
+ StreamTy *StreamObj = nullptr;
ol_result_t Result = State.createStream(
ThreadState.getDefaultDevice(), QueueKind::ExplicitBlocking, &StreamObj);
if (Result == OL_SUCCESS)
@@ -162,7 +165,29 @@ Error_t StreamCreate(Stream_t *Stream) {
return convertAndSetLastError(Result);
}
+Error_t StreamCreateWithFlags(Stream_t *Stream, unsigned int Flags) {
+ if (!Stream)
+ return setLastError(ErrorInvalidValue);
+
+ if (Flags == StreamDefault)
+ return StreamCreate(Stream);
+ if (Flags != StreamNonBlocking)
+ return setLastError(ErrorInvalidValue);
+
+ StateTy &State = StateTy::get();
+ ThreadStateTy &ThreadState = ThreadStateTy::get();
+ StreamTy *StreamObj = nullptr;
+ ol_result_t Result =
+ State.createStream(ThreadState.getDefaultDevice(),
+ QueueKind::ExplicitNonBlocking, &StreamObj);
+ if (Result == OL_SUCCESS)
+ *Stream = toLanguageStream(StreamObj);
+ return convertAndSetLastError(Result);
+}
+
Error_t StreamDestroy(Stream_t Stream) {
+ if (!Stream)
+ return setLastError(ErrorInvalidValue);
ol_result_t Result = StateTy::get().destroyStream(toInternalStream(Stream));
return convertAndSetLastError(Result);
}
diff --git a/offload/test/offloading/CUDA/stream_api.cu b/offload/test/offloading/CUDA/stream_api.cu
index 885ef3b300069..c5b1caaa5315c 100644
--- a/offload/test/offloading/CUDA/stream_api.cu
+++ b/offload/test/offloading/CUDA/stream_api.cu
@@ -22,12 +22,39 @@ static void print_error(const char *Label, cudaError_t Error) {
__global__ void setValue(int *Out, int Value) { *Out = Value; }
int main(int argc, char **argv) {
+ print_error("null stream create", cudaStreamCreate(nullptr));
+ // CHECK: null stream create value: 1
+ // CHECK: null stream create name: cudaErrorInvalidValue
+ print_error("null flags stream create",
+ cudaStreamCreateWithFlags(nullptr, cudaStreamDefault));
+ // CHECK: null flags stream create value: 1
+ // CHECK: null flags stream create name: cudaErrorInvalidValue
+
+ cudaStream_t InvalidFlagsStream = nullptr;
+ print_error("invalid stream flags",
+ cudaStreamCreateWithFlags(&InvalidFlagsStream, ~0u));
+ // CHECK: invalid stream flags value: 1
+ // CHECK: invalid stream flags name: cudaErrorInvalidValue
+ printf("invalid flags stream: %d\n", InvalidFlagsStream == nullptr);
+ // CHECK: invalid flags stream: 1
+
cudaStream_t Stream = nullptr;
if (cudaStreamCreate(&Stream) != cudaSuccess)
return 1;
+ cudaStream_t BlockingStream = nullptr;
+ if (cudaStreamCreateWithFlags(&BlockingStream, cudaStreamDefault) !=
+ cudaSuccess)
+ return 1;
+ cudaStream_t NonBlockingStream = nullptr;
+ if (cudaStreamCreateWithFlags(&NonBlockingStream, cudaStreamNonBlocking) !=
+ cudaSuccess)
+ return 1;
printf("stream created: %d\n", Stream != nullptr);
// CHECK: stream created: 1
+ printf("stream flags created: %d %d\n", BlockingStream != nullptr,
+ NonBlockingStream != nullptr);
+ // CHECK: stream flags created: 1 1
int *StreamPtr = nullptr;
int *DefaultPtr = nullptr;
diff --git a/offload/test/offloading/HIP/stream_api.hip b/offload/test/offloading/HIP/stream_api.hip
index 6d16c2546245f..fbfca230ee697 100644
--- a/offload/test/offloading/HIP/stream_api.hip
+++ b/offload/test/offloading/HIP/stream_api.hip
@@ -22,12 +22,38 @@ static void print_error(const char *Label, hipError_t Error) {
__global__ void setValue(int *Out, int Value) { *Out = Value; }
int main(int argc, char **argv) {
+ print_error("null stream create", hipStreamCreate(nullptr));
+ // CHECK: null stream create value: 1
+ // CHECK: null stream create name: hipErrorInvalidValue
+ print_error("null flags stream create",
+ hipStreamCreateWithFlags(nullptr, hipStreamDefault));
+ // CHECK: null flags stream create value: 1
+ // CHECK: null flags stream create name: hipErrorInvalidValue
+
+ hipStream_t InvalidFlagsStream = nullptr;
+ print_error("invalid stream flags",
+ hipStreamCreateWithFlags(&InvalidFlagsStream, ~0u));
+ // CHECK: invalid stream flags value: 1
+ // CHECK: invalid stream flags name: hipErrorInvalidValue
+ printf("invalid flags stream: %d\n", InvalidFlagsStream == nullptr);
+ // CHECK: invalid flags stream: 1
+
hipStream_t Stream = nullptr;
if (hipStreamCreate(&Stream) != hipSuccess)
return 1;
+ hipStream_t BlockingStream = nullptr;
+ if (hipStreamCreateWithFlags(&BlockingStream, hipStreamDefault) != hipSuccess)
+ return 1;
+ hipStream_t NonBlockingStream = nullptr;
+ if (hipStreamCreateWithFlags(&NonBlockingStream, hipStreamNonBlocking) !=
+ hipSuccess)
+ return 1;
printf("stream created: %d\n", Stream != nullptr);
// CHECK: stream created: 1
+ printf("stream flags created: %d %d\n", BlockingStream != nullptr,
+ NonBlockingStream != nullptr);
+ // CHECK: stream flags created: 1 1
int *StreamPtr = nullptr;
int *DefaultPtr = nullptr;
>From d084a26319d9efd2511b2ba97bd1cc0e95ddc60b Mon Sep 17 00:00:00 2001
From: Sophia Herrmann <herrmann15 at llnl.gov>
Date: Thu, 13 Aug 2026 16:47:23 -0700
Subject: [PATCH 2/3] 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 | 43 ++++++
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, 340 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 e03b7e754ab3f..14904c52b98ff 100644
--- a/clang/lib/CodeGen/CGCUDANV.cpp
+++ b/clang/lib/CodeGen/CGCUDANV.cpp
@@ -442,7 +442,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 54583fe3abbd8..8b21400ab959f 100644
--- a/clang/lib/Driver/ToolChains/Clang.cpp
+++ b/clang/lib/Driver/ToolChains/Clang.cpp
@@ -8396,6 +8396,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 7dd19fa6e44c8..f90832f7e61e3 100644
--- a/offload/languages/kernel/include/LanguageUtils.h
+++ b/offload/languages/kernel/include/LanguageUtils.h
@@ -14,6 +14,10 @@
#include "State.h"
#include "Stream.h"
+using RuntimeState = llvm::offload::StateTy;
+using ThreadState = llvm::offload::ThreadStateTy;
+using StreamTy = llvm::offload::StreamTy;
+
namespace llvm {
namespace offload {
@@ -61,6 +65,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 02feb43012d67..2753c3e01cc53 100644
--- a/offload/languages/kernel/include/State.h
+++ b/offload/languages/kernel/include/State.h
@@ -155,7 +155,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 6dab19fc2fd8c..7b31987050101 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>
@@ -44,10 +45,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: {
@@ -82,6 +86,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 c5b1caaa5315c..0e1c328cc5f1b 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 fbfca230ee697..3460a6a0c351c 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 2cd53889863a35e76ad77850917c30b35901780f Mon Sep 17 00:00:00 2001
From: Sophia Herrmann <herrmann15 at llnl.gov>
Date: Wed, 19 Aug 2026 17:42:04 -0700
Subject: [PATCH 3/3] add event cleanup
---
offload/languages/kernel/CMakeLists.txt | 1 +
.../languages/kernel/include/LanguageUtils.h | 74 ++++++++++------
offload/languages/kernel/include/State.h | 3 +-
offload/languages/kernel/include/Stream.h | 19 +++++
.../languages/kernel/src/LanguageLaunch.cpp | 8 +-
.../languages/kernel/src/LanguageRuntime.cpp | 40 ++++-----
offload/languages/kernel/src/State.cpp | 10 ++-
offload/languages/kernel/src/Stream.cpp | 85 +++++++++++++++++++
8 files changed, 183 insertions(+), 57 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 bb03092f6e9d0..dfd1b9541e197 100644
--- a/offload/languages/kernel/CMakeLists.txt
+++ b/offload/languages/kernel/CMakeLists.txt
@@ -64,6 +64,7 @@ add_llvm_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 f90832f7e61e3..60d3e1a9782dd 100644
--- a/offload/languages/kernel/include/LanguageUtils.h
+++ b/offload/languages/kernel/include/LanguageUtils.h
@@ -65,57 +65,77 @@ 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())
+ StateTy &State = StateTy::get();
+ ThreadStateTy &ThreadState = ThreadStateTy::get();
+ 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) {
+ if (Event)
+ Events.push_back(Event);
+ 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,
ol_device_handle_t Device) {
- if (!RuntimeState::hasLegacyDefaultStream(Device))
+ StateTy &State = StateTy::get();
+ ThreadStateTy &ThreadState = ThreadStateTy::get();
+ 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;
ol_result_t Result =
olCreateEvent(DefaultStream->Queue, OL_EVENT_FLAGS_NONE, &Event);
- if (Result != OL_SUCCESS)
+ if (Result != OL_SUCCESS) {
+ if (Event) {
+ SmallVector<ol_event_handle_t, 1> Events = {Event};
+ syncAndDestroyEvents(Events);
+ }
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;
-
- StreamTy *InternalStream = toInternalStream(Stream);
- if (!StateTy::get().isStreamRegistered(InternalStream))
- return ErrorInvalidResourceHandle;
-
- *Queue = InternalStream->Queue;
- return Success;
+ }
+ return SourceStream->waitOnAndTrackDependencyEvents(
+ llvm::ArrayRef<ol_event_handle_t>(&Event, 1));
}
} // namespace offload
diff --git a/offload/languages/kernel/include/State.h b/offload/languages/kernel/include/State.h
index 2753c3e01cc53..7b4ae77a1f51f 100644
--- a/offload/languages/kernel/include/State.h
+++ b/offload/languages/kernel/include/State.h
@@ -155,7 +155,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..5cb2d0756ba69 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(ArrayRef<ol_event_handle_t> Events);
+ ol_result_t syncStream();
+
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..d7640b780f483 100644
--- a/offload/languages/kernel/src/LanguageLaunch.cpp
+++ b/offload/languages/kernel/src/LanguageLaunch.cpp
@@ -47,16 +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) ||
+ : ThreadState.getDefaultStream();
+ if (!LaunchStream || !State.isStreamRegistered(LaunchStream) ||
LaunchStream->Device != Device)
return &InvalidConfigurationError;
- if (LaunchStream->Kind == llvm::offload::QueueKind::LegacyDefault) {
+ if (LaunchStream->Kind == QueueKind::LegacyDefault) {
ol_result_t Result = waitOnBlockingStreams();
if (Result != OL_SUCCESS)
return Result;
- } else if (LaunchStream->Kind == llvm::offload::QueueKind::ExplicitBlocking) {
+ } else if (LaunchStream->Kind == QueueKind::ExplicitBlocking) {
ol_result_t Result = waitOnLegacyDefaultStream(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 7b31987050101..375f6120e2dd5 100644
--- a/offload/languages/kernel/src/LanguageRuntime.cpp
+++ b/offload/languages/kernel/src/LanguageRuntime.cpp
@@ -50,31 +50,27 @@ Error_t Memcpy(void *Dst, const void *Src, size_t Size, MemcpyKind Kind) {
if (Result != OL_SUCCESS)
return convertAndSetLastError(Result);
}
- ol_device_handle_t Device = ThreadState::getDefaultDevice();
- ol_queue_handle_t Queue = ThreadState::getDefaultQueue();
+ StateTy &State = StateTy::get();
+ ThreadStateTy &ThreadState = ThreadState::get();
+ 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;
@@ -86,19 +82,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->syncStream();
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->syncStream() : olSyncQueue(nullptr);
return convertAndSetLastError(Result);
}
@@ -200,11 +193,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->syncStream();
return convertAndSetLastError(Result);
}
diff --git a/offload/languages/kernel/src/State.cpp b/offload/languages/kernel/src/State.cpp
index 82292e7813c97..5cc4452f846b9 100644
--- a/offload/languages/kernel/src/State.cpp
+++ b/offload/languages/kernel/src/State.cpp
@@ -91,7 +91,7 @@ static void destroyStreamHandle(StreamTy *&Stream) {
if (!Stream)
return;
- olSyncQueue(Stream->Queue);
+ (void)Stream->syncStream();
olDestroyQueue(Stream->Queue);
delete Stream;
Stream = nullptr;
@@ -332,8 +332,12 @@ ol_result_t StateTy::destroyStream(StreamTy *Stream) {
if (!isStreamRegistered(Stream))
return &InvalidStreamError;
- removeStream(Stream);
- ol_result_t Result = olDestroyQueue(Stream->Queue);
+ ol_result_t Result = Stream->syncStream();
+ if (Result != OL_SUCCESS)
+ return Result;
+
+ get().removeStream(Stream);
+ 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..a6c012259146e
--- /dev/null
+++ b/offload/languages/kernel/src/Stream.cpp
@@ -0,0 +1,85 @@
+//===-- 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(ArrayRef<ol_event_handle_t> Events) {
+ if (Events.empty())
+ return OL_SUCCESS;
+
+ SmallVector<ol_event_handle_t, 8> MutableEvents(Events.begin(), Events.end());
+ std::lock_guard<std::mutex> LG(DependencyEventsLock);
+ ol_result_t WaitResult =
+ olWaitEvents(Queue, MutableEvents.data(), MutableEvents.size());
+ if (WaitResult != OL_SUCCESS) {
+ syncAndDestroyEvents(MutableEvents);
+ return WaitResult;
+ }
+
+ DependencyEvents.append(MutableEvents.begin(), MutableEvents.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::syncStream() {
+ 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;
+}
More information about the llvm-branch-commits
mailing list