feature: add host functions workers

* add common host function worker interface
* add worker as a single thread per csr with 3 modes
* add logic for waiting on internal tag, check gpu hang
* if tag is in pending state, read callback data, run callback
and signal completion
* threads will exit the work loop once stop request
is called in finish
* add multi thread unit tests

Related-To: NEO-14577
Signed-off-by: Kamil Kopryk <[email protected]>

Author: KamilKoprykIntel

level_zero/core/source/cmdlist/cmdlist_hw.inl

@@ -1864,6 +1864,7 @@ void CommandListCoreFamily<gfxCoreFamily>::dispatchHostFunction(
         auto csr = getCsr(false);
         csr->ensureHostFunctionDataInitialization();
         this->commandContainer.addToResidencyContainer(csr->getHostFunctionDataAllocation());
+        csr->signalHostFunctionWorker();
         NEO::HostFunctionHelper::programHostFunction<GfxFamily>(*this->commandContainer.getCommandStream(), csr->getHostFunctionData(), userHostFunctionAddress, userDataAddress);
     } else {
         addHostFunctionToPatchCommands(userHostFunctionAddress, userDataAddress);

level_zero/core/source/cmdqueue/cmdqueue_hw_gen12lp.inl

@@ -199,6 +199,7 @@ void CommandQueueHw<gfxCoreFamily>::patchCommands(CommandList &commandList, uint
         case CommandToPatch::HostFunctionEntry:
             csr->ensureHostFunctionDataInitialization();
             csr->makeResidentHostFunctionAllocation();
+            csr->signalHostFunctionWorker();
             NEO::HostFunctionHelper::programHostFunctionAddress<GfxFamily>(nullptr, commandToPatch.pCommand, csr->getHostFunctionData(), commandToPatch.baseAddress);
             break;
 

level_zero/core/source/cmdqueue/cmdqueue_xe_hp_core_and_later.inl

@@ -287,6 +287,7 @@ void CommandQueueHw<gfxCoreFamily>::patchCommands(CommandList &commandList, uint
         case CommandToPatch::HostFunctionEntry:
             csr->ensureHostFunctionDataInitialization();
             csr->makeResidentHostFunctionAllocation();
+            csr->signalHostFunctionWorker();
             NEO::HostFunctionHelper::programHostFunctionAddress<GfxFamily>(nullptr, commandToPatch.pCommand, csr->getHostFunctionData(), commandToPatch.baseAddress);
             break;
 

level_zero/core/test/unit_tests/sources/cmdqueue/test_cmdqueue_2.cpp

@@ -1156,6 +1156,10 @@ HWTEST_F(HostFunctionsCmdPatchTests, givenHostFunctionPatchCommandsWhenPatchComm
     NEO::CommandStreamReceiver *csr = nullptr;
     device->getCsrForOrdinalAndIndex(&csr, 0u, 0u, ZE_COMMAND_QUEUE_PRIORITY_NORMAL, 0, false);
     auto commandQueue = std::make_unique<MockCommandQueueHw<FamilyType::gfxCoreFamily>>(device, csr, &desc);
+    MockCommandStreamReceiver mockCsr(*neoDevice->executionEnvironment, neoDevice->getRootDeviceIndex(), neoDevice->getDeviceBitfield());
+    const auto oldCsr = commandQueue->csr;
+    commandQueue->csr = &mockCsr;
+
     auto commandList = std::make_unique<WhiteBox<::L0::CommandListCoreFamily<FamilyType::gfxCoreFamily>>>();
     commandList->commandsToPatch.clear();
 
@@ -1197,6 +1201,8 @@ HWTEST_F(HostFunctionsCmdPatchTests, givenHostFunctionPatchCommandsWhenPatchComm
     commandQueue->patchCommands(*commandList, 0, false, nullptr);
 
     EXPECT_NE(nullptr, commandQueue->csr->getHostFunctionDataAllocation());
+    EXPECT_EQ(1u, mockCsr.createHostFunctionWorkerCounter);
+    EXPECT_EQ(1u, mockCsr.signalHostFunctionWorkerCounter);
 
     auto &hostFunctionDataFromCsr = commandQueue->csr->getHostFunctionData();
 
@@ -1220,6 +1226,8 @@ HWTEST_F(HostFunctionsCmdPatchTests, givenHostFunctionPatchCommandsWhenPatchComm
     // internal tag wait - semaphore wait
     EXPECT_EQ(static_cast<uint32_t>(HostFunctionTagStatus::completed), internalTagMiWait.getSemaphoreDataDword());
     EXPECT_EQ(reinterpret_cast<uint64_t>(hostFunctionDataFromCsr.internalTag), internalTagMiWait.getSemaphoreGraphicsAddress());
+
+    commandQueue->csr = oldCsr;
 }
 
 } // namespace ult

shared/source/command_stream/CMakeLists.txt

@@ -30,8 +30,17 @@ set(NEO_CORE_COMMAND_STREAM
     ${CMAKE_CURRENT_SOURCE_DIR}/definitions${BRANCH_DIR_SUFFIX}stream_properties.inl
     ${CMAKE_CURRENT_SOURCE_DIR}/device_command_stream.h
     ${CMAKE_CURRENT_SOURCE_DIR}/host_function.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/host_function.inl
     ${CMAKE_CURRENT_SOURCE_DIR}/host_function_enablers.inl
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_cv.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_cv.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_interface.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_interface.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_counting_semaphore.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_counting_semaphore.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_atomic.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/host_function_worker_atomic.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/linear_stream.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/linear_stream.h
     ${CMAKE_CURRENT_SOURCE_DIR}/preemption.cpp

shared/source/command_stream/command_stream_receiver.cpp

@@ -9,6 +9,7 @@
 
 #include "shared/source/command_container/implicit_scaling.h"
 #include "shared/source/command_stream/aub_subcapture_status.h"
+#include "shared/source/command_stream/host_function_worker_interface.h"
 #include "shared/source/command_stream/scratch_space_controller.h"
 #include "shared/source/command_stream/submission_status.h"
 #include "shared/source/command_stream/submissions_aggregator.h"
@@ -97,6 +98,7 @@ CommandStreamReceiver::CommandStreamReceiver(ExecutionEnvironment &executionEnvi
     auto &compilerProductHelper = rootDeviceEnvironment.getHelper<CompilerProductHelper>();
     this->heaplessModeEnabled = compilerProductHelper.isHeaplessModeEnabled(hwInfo);
     this->heaplessStateInitEnabled = compilerProductHelper.isHeaplessStateInitEnabled(heaplessModeEnabled);
+    this->hostFunctionWorkerMode = debugManager.flags.HostFunctionWorkMode.get();
 }
 
 CommandStreamReceiver::~CommandStreamReceiver() {
@@ -235,6 +237,25 @@ WaitStatus CommandStreamReceiver::waitForTaskCountAndCleanTemporaryAllocationLis
     return waitForTaskCountAndCleanAllocationList(requiredTaskCount, TEMPORARY_ALLOCATION);
 }
 
+void CommandStreamReceiver::createHostFunctionWorker() {
+
+    if (this->hostFunctionWorker != nullptr) {
+        return;
+    }
+
+    this->hostFunctionWorker = HostFunctionFactory::createHostFunctionWorker(this->hostFunctionWorkerMode,
+                                                                             this->isAubMode(),
+                                                                             this->downloadAllocationImpl,
+                                                                             this->getHostFunctionDataAllocation(),
+                                                                             &this->getHostFunctionData());
+
+    this->hostFunctionWorker->start();
+}
+
+IHostFunctionWorker *CommandStreamReceiver::getHostFunctionWorker() {
+    return this->hostFunctionWorker;
+}
+
 void CommandStreamReceiver::ensureCommandBufferAllocation(LinearStream &commandStream, size_t minimumRequiredSize, size_t additionalAllocationSize) {
     if (commandStream.getAvailableSpace() >= minimumRequiredSize) {
         return;
@@ -419,6 +440,10 @@ void CommandStreamReceiver::cleanupResources() {
         tagsMultiAllocation = nullptr;
     }
 
+    if (hostFunctionWorker) {
+        cleanupHostFunctionWorker();
+    }
+
     if (hostFunctionDataMultiAllocation) {
         hostFunctionDataAllocation = nullptr;
 
@@ -464,6 +489,12 @@ void CommandStreamReceiver::cleanupResources() {
     ownedPrivateAllocations.clear();
 }
 
+void CommandStreamReceiver ::cleanupHostFunctionWorker() {
+    hostFunctionWorker->finish();
+    delete hostFunctionWorker;
+    hostFunctionWorker = nullptr;
+}
+
 WaitStatus CommandStreamReceiver::waitForCompletionWithTimeout(const WaitParams &params, TaskCountType taskCountToWait) {
     bool printWaitForCompletion = debugManager.flags.LogWaitingForCompletion.get();
     if (printWaitForCompletion) {
@@ -697,6 +728,10 @@ void *CommandStreamReceiver::getIndirectHeapCurrentPtr(IndirectHeapType heapType
     return nullptr;
 }
 
+void CommandStreamReceiver::signalHostFunctionWorker() {
+    hostFunctionWorker->submit();
+}
+
 void CommandStreamReceiver::ensureHostFunctionDataInitialization() {
     if (!this->hostFunctionInitialized.load(std::memory_order_acquire)) {
         initializeHostFunctionData();
@@ -717,6 +752,9 @@ void CommandStreamReceiver::initializeHostFunctionData() {
     this->hostFunctionData.entry = reinterpret_cast<decltype(HostFunctionData::entry)>(ptrOffset(hostFunctionBuffer, HostFunctionHelper::entryOffset));
     this->hostFunctionData.userData = reinterpret_cast<decltype(HostFunctionData::userData)>(ptrOffset(hostFunctionBuffer, HostFunctionHelper::userDataOffset));
     this->hostFunctionData.internalTag = reinterpret_cast<decltype(HostFunctionData::internalTag)>(ptrOffset(hostFunctionBuffer, HostFunctionHelper::internalTagOffset));
+
+    createHostFunctionWorker();
+
     this->hostFunctionInitialized.store(true, std::memory_order_release);
 }
 
@@ -968,12 +1006,14 @@ bool CommandStreamReceiver::createPreemptionAllocation() {
 std::unique_lock<CommandStreamReceiver::MutexType> CommandStreamReceiver::obtainUniqueOwnership() {
     return std::unique_lock<CommandStreamReceiver::MutexType>(this->ownershipMutex);
 }
+
 std::unique_lock<CommandStreamReceiver::MutexType> CommandStreamReceiver::tryObtainUniqueOwnership() {
     return std::unique_lock<CommandStreamReceiver::MutexType>(this->ownershipMutex, std::try_to_lock);
 }
 std::unique_lock<CommandStreamReceiver::MutexType> CommandStreamReceiver::obtainHostPtrSurfaceCreationLock() {
     return std::unique_lock<CommandStreamReceiver::MutexType>(this->hostPtrSurfaceCreationMutex);
 }
+
 AllocationsList &CommandStreamReceiver::getTemporaryAllocations() { return internalAllocationStorage->getTemporaryAllocations(); }
 AllocationsList &CommandStreamReceiver::getAllocationsForReuse() { return internalAllocationStorage->getAllocationsForReuse(); }
 AllocationsList &CommandStreamReceiver::getDeferredAllocations() { return internalAllocationStorage->getDeferredAllocations(); }

shared/source/command_stream/command_stream_receiver.h

@@ -67,6 +67,7 @@ class KmdNotifyHelper;
 class GfxCoreHelper;
 class ProductHelper;
 class ReleaseHelper;
+class IHostFunctionWorker;
 enum class WaitStatus;
 struct AubSubCaptureStatus;
 class SharedPoolAllocation;
@@ -148,6 +149,8 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
     MOCKABLE_VIRTUAL WaitStatus waitForTaskCount(TaskCountType requiredTaskCount);
     WaitStatus waitForTaskCountAndCleanAllocationList(TaskCountType requiredTaskCount, uint32_t allocationUsage);
     MOCKABLE_VIRTUAL WaitStatus waitForTaskCountAndCleanTemporaryAllocationList(TaskCountType requiredTaskCount);
+    MOCKABLE_VIRTUAL void createHostFunctionWorker();
+    IHostFunctionWorker *getHostFunctionWorker();
 
     LinearStream &getCS(size_t minRequiredSize = 1024u);
     OSInterface *getOSInterface() const;
@@ -270,7 +273,6 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
     MOCKABLE_VIRTUAL bool createPerDssBackedBuffer(Device &device);
     [[nodiscard]] MOCKABLE_VIRTUAL std::unique_lock<MutexType> obtainUniqueOwnership();
     [[nodiscard]] MOCKABLE_VIRTUAL std::unique_lock<MutexType> tryObtainUniqueOwnership();
-
     bool peekTimestampPacketWriteEnabled() const { return timestampPacketWriteEnabled; }
 
     bool isLatestTaskCountFlushed() {
@@ -567,6 +569,7 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
     bool isLatestFlushIsTaskCountUpdateOnly() const { return latestFlushIsTaskCountUpdateOnly; }
 
     MOCKABLE_VIRTUAL uint32_t getContextGroupId() const;
+    MOCKABLE_VIRTUAL void signalHostFunctionWorker();
 
     void ensureHostFunctionDataInitialization();
     HostFunctionData &getHostFunctionData();
@@ -584,6 +587,7 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
                                              TaskCountType taskLevel, DispatchFlags &dispatchFlags, Device &device) = 0;
 
     void cleanupResources();
+    void cleanupHostFunctionWorker();
     void printDeviceIndex();
     void checkForNewResources(TaskCountType submittedTaskCount, TaskCountType allocationTaskCount, GraphicsAllocation &gfxAllocation);
     bool checkImplicitFlushForGpuIdle();
@@ -610,6 +614,7 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
     std::unique_ptr<TagAllocatorBase> timestampPacketAllocator;
     std::unique_ptr<Thread> userPauseConfirmation;
     std::unique_ptr<IndirectHeap> globalStatelessHeap;
+    IHostFunctionWorker *hostFunctionWorker = nullptr;
 
     ResidencyContainer residencyAllocations;
     PrivateAllocsToReuseContainer ownedPrivateAllocations;
@@ -664,7 +669,6 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
     std::atomic<TaskCountType> taskCount{0};
 
     std::atomic<uint32_t> numClients = 0u;
-
     DispatchMode dispatchMode = DispatchMode::immediateDispatch;
     SamplerCacheFlushState samplerCacheFlushRequired = SamplerCacheFlushState::samplerCacheFlushNotRequired;
     PreemptionMode lastPreemptionMode = PreemptionMode::Initial;
@@ -673,7 +677,7 @@ class CommandStreamReceiver : NEO::NonCopyableAndNonMovableClass {
     uint32_t lastSentL3Config = 0;
     uint32_t latestSentStatelessMocsConfig;
     uint64_t lastSentSliceCount;
-
+    int32_t hostFunctionWorkerMode = -1;
     uint32_t requiredScratchSlot0Size = 0;
     uint32_t requiredScratchSlot1Size = 0;
     uint32_t lastAdditionalKernelExecInfo;

shared/source/command_stream/host_function.cpp

@@ -0,0 +1,37 @@
+/*
+ * Copyright (C) 2025 Intel Corporation
+ *
+ * SPDX-License-Identifier: MIT
+ *
+ */
+
+#include "shared/source/command_stream/host_function.h"
+
+#include "shared/source/command_stream/command_stream_receiver.h"
+#include "shared/source/command_stream/host_function_worker_atomic.h"
+#include "shared/source/command_stream/host_function_worker_counting_semaphore.h"
+#include "shared/source/command_stream/host_function_worker_cv.h"
+#include "shared/source/command_stream/host_function_worker_interface.h"
+
+namespace NEO::HostFunctionFactory {
+
+IHostFunctionWorker *createHostFunctionWorker(int32_t hostFunctionWorkerMode,
+                                              bool isAubMode,
+                                              const std::function<void(GraphicsAllocation &)> &downloadAllocationImpl,
+                                              GraphicsAllocation *allocation,
+                                              HostFunctionData *data) {
+
+    bool skipHostFunctionExecution = isAubMode;
+
+    switch (hostFunctionWorkerMode) {
+    default:
+    case 0:
+        return new HostFunctionWorkerCountingSemaphore(skipHostFunctionExecution, downloadAllocationImpl, allocation, data);
+    case 1:
+        return new HostFunctionWorkerCV(skipHostFunctionExecution, downloadAllocationImpl, allocation, data);
+    case 2:
+        return new HostFunctionWorkerAtomic(skipHostFunctionExecution, downloadAllocationImpl, allocation, data);
+    }
+}
+
+} // namespace NEO::HostFunctionFactory

shared/source/command_stream/host_function.h

@@ -9,10 +9,14 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 
 namespace NEO {
 
 class LinearStream;
+class CommandStreamReceiver;
+class IHostFunctionWorker;
+class GraphicsAllocation;
 
 struct HostFunctionData {
     volatile uint64_t *entry = nullptr;
@@ -47,4 +51,13 @@ struct HostFunctionHelper {
     static void programWaitForHostFunctionCompletion(LinearStream *commandStream, void *cmdBuffer, const HostFunctionData &hostFunctionData);
 };
 
+namespace HostFunctionFactory {
+IHostFunctionWorker *createHostFunctionWorker(int32_t hostFunctionWorkerMode,
+                                              bool isAubMode,
+                                              const std::function<void(GraphicsAllocation &)> &downloadAllocationImpl,
+                                              GraphicsAllocation *allocation,
+                                              HostFunctionData *data);
+
+}
+
 } // namespace NEO

shared/source/command_stream/host_function_worker_atomic.cpp

@@ -0,0 +1,68 @@
+/*
+ * Copyright (C) 2025 Intel Corporation
+ *
+ * SPDX-License-Identifier: MIT
+ *
+ */
+
+#include "shared/source/command_stream/host_function_worker_atomic.h"
+
+#include "shared/source/command_stream/host_function.h"
+
+namespace NEO {
+HostFunctionWorkerAtomic::HostFunctionWorkerAtomic(bool skipHostFunctionExecution,
+                                                   const std::function<void(GraphicsAllocation &)> &downloadAllocationImpl,
+                                                   GraphicsAllocation *allocation,
+                                                   HostFunctionData *data)
+    : IHostFunctionWorker(skipHostFunctionExecution, downloadAllocationImpl, allocation, data) {
+}
+
+HostFunctionWorkerAtomic::~HostFunctionWorkerAtomic() = default;
+
+void HostFunctionWorkerAtomic::start() {
+
+    std::lock_guard<std::mutex> lg{workerMutex};
+    if (!worker) {
+        worker = std::make_unique<std::jthread>([this](std::stop_token st) {
+            this->workerLoop(st);
+        });
+    }
+}
+
+void HostFunctionWorkerAtomic::finish() {
+    std::lock_guard<std::mutex> lg{workerMutex};
+    if (worker) {
+        worker->request_stop();
+        pending.fetch_add(1u);
+        pending.notify_one();
+        worker.reset(nullptr);
+    }
+}
+
+void HostFunctionWorkerAtomic::submit() noexcept {
+    pending.fetch_add(1, std::memory_order_release);
+    pending.notify_one();
+}
+
+void HostFunctionWorkerAtomic::workerLoop(std::stop_token st) noexcept {
+
+    while (true) {
+
+        while (pending.load(std::memory_order_acquire) == 0) {
+            pending.wait(0, std::memory_order_acquire);
+        }
+
+        if (st.stop_requested()) {
+            return;
+        }
+
+        pending.fetch_sub(1, std::memory_order_acq_rel);
+
+        bool sucess = this->runHostFunction(st);
+        if (!sucess) [[unlikely]] {
+            return;
+        }
+    }
+}
+
+} // namespace NEO