From 56ffd291fb5d5e4c257ac901626a164ec0570d31 Mon Sep 17 00:00:00 2001 From: PietroGhg Date: Fri, 12 Apr 2024 14:13:11 +0100 Subject: [PATCH 1/3] Refactor threadpool to fix coverity warnings --- source/adapters/native_cpu/threadpool.hpp | 92 +++++++++-------------- 1 file changed, 35 insertions(+), 57 deletions(-) diff --git a/source/adapters/native_cpu/threadpool.hpp b/source/adapters/native_cpu/threadpool.hpp index 03763224e8..55c32eb84b 100644 --- a/source/adapters/native_cpu/threadpool.hpp +++ b/source/adapters/native_cpu/threadpool.hpp @@ -11,9 +11,10 @@ #include #include #include +#include #include #include -#include +#include #include #include #include @@ -30,18 +31,14 @@ namespace detail { class worker_thread { public: // Initializes state, but does not start the worker thread - worker_thread() noexcept : m_isRunning(false), m_numTasks(0) {} - - // Creates and launches the worker thread - inline void start(size_t threadId) { + worker_thread(size_t threadId) noexcept + : m_threadId(threadId), m_isRunning(false), m_numTasks(0) { std::lock_guard lock(m_workMutex); if (this->is_running()) { return; } - m_threadId = threadId; m_worker = std::thread([this]() { while (true) { - // pin the thread to the cpu std::unique_lock lock(m_workMutex); // Wait until there's work available m_startWorkCondition.wait( @@ -51,7 +48,7 @@ class worker_thread { break; } // Retrieve a task from the queue - auto task = m_tasks.front(); + worker_task_t task = std::move(m_tasks.front()); m_tasks.pop(); // Not modifying internal state anymore, can release the mutex @@ -63,7 +60,7 @@ class worker_thread { } }); - m_isRunning = true; + m_isRunning.store(true, std::memory_order_release); } inline void schedule(const worker_task_t &task) { @@ -79,16 +76,12 @@ class worker_thread { size_t num_pending_tasks() const noexcept { // m_numTasks is an atomic counter because we don't want to lock the mutex // here, num_pending_tasks is only used for heuristics - return m_numTasks.load(); + return m_numTasks.load(std::memory_order_acquire); } // Waits for all tasks to finish and destroys the worker thread inline void stop() { - { - // Notify the worker thread to stop executing - std::lock_guard lock(m_workMutex); - m_isRunning = false; - } + m_isRunning.store(false, std::memory_order_release); m_startWorkCondition.notify_all(); if (m_worker.joinable()) { // Wait for the worker thread to finish handling the task queue @@ -97,18 +90,21 @@ class worker_thread { } // Checks whether the thread pool is currently running threads - inline bool is_running() const noexcept { return m_isRunning; } + inline bool is_running() const noexcept { + return m_isRunning.load(std::memory_order_acquire); + } private: // Unique ID identifying the thread in the threadpool - size_t m_threadId; + const size_t m_threadId; + std::thread m_worker; std::mutex m_workMutex; std::condition_variable m_startWorkCondition; - bool m_isRunning; + std::atomic m_isRunning; std::queue m_tasks; @@ -121,47 +117,21 @@ class worker_thread { // parameters and futures. class simple_thread_pool { public: - simple_thread_pool(size_t numThreads = 0) noexcept : m_isRunning(false) { - this->resize(numThreads); - this->start(); - } - - ~simple_thread_pool() { this->stop(); } - - // Creates and launches the worker threads - inline void start() { - if (this->is_running()) { - return; - } - size_t threadId = 0; - for (auto &t : m_workers) { - t.start(threadId); - threadId++; + simple_thread_pool() noexcept + : m_isRunning(false), m_numThreads(get_num_threads()) { + for (size_t i = 0; i < m_numThreads; i++) { + m_workers.emplace_front(i); } m_isRunning.store(true, std::memory_order_release); } - // Waits for all tasks to finish and destroys the worker threads - inline void stop() { + ~simple_thread_pool() { for (auto &t : m_workers) { t.stop(); } m_isRunning.store(false, std::memory_order_release); } - inline void resize(size_t numThreads) { - char *envVar = std::getenv("SYCL_NATIVE_CPU_HOST_THREADS"); - if (envVar) { - numThreads = std::stoul(envVar); - } - if (numThreads == 0) { - numThreads = std::thread::hardware_concurrency(); - } - if (!this->is_running() && (numThreads != this->num_threads())) { - m_workers = decltype(m_workers)(numThreads); - } - } - inline void schedule(const worker_task_t &task) { // Schedule the task on the best available worker thread this->best_worker().schedule(task); @@ -171,7 +141,7 @@ class simple_thread_pool { return m_isRunning.load(std::memory_order_acquire); } - inline size_t num_threads() const noexcept { return m_workers.size(); } + inline size_t num_threads() const noexcept { return m_numThreads; } inline size_t num_pending_tasks() const noexcept { return std::accumulate(std::begin(m_workers), std::end(m_workers), @@ -201,9 +171,22 @@ class simple_thread_pool { } private: - std::vector m_workers; + static size_t get_num_threads() { + size_t numThreads; + char *envVar = std::getenv("SYCL_NATIVE_CPU_HOST_THREADS"); + if (envVar) { + numThreads = std::stoul(envVar); + } else { + numThreads = std::thread::hardware_concurrency(); + } + return numThreads; + } + + std::forward_list m_workers; std::atomic m_isRunning; + + const size_t m_numThreads; }; } // namespace detail @@ -211,14 +194,9 @@ template class threadpool_interface { ThreadPoolT threadpool; public: - void start() { threadpool.start(); } - - void stop() { threadpool.stop(); } - size_t num_threads() const noexcept { return threadpool.num_threads(); } - threadpool_interface(size_t numThreads) : threadpool(numThreads) {} - threadpool_interface() : threadpool(0) {} + threadpool_interface() : threadpool() {} auto schedule_task(worker_task_t &&task) { auto workerTask = std::make_shared>( From 49046ecb2d1f7e51e53e0bd2510124a9c6cb4b0a Mon Sep 17 00:00:00 2001 From: "Neil R. Spruit" Date: Wed, 17 Apr 2024 17:34:12 -0700 Subject: [PATCH 2/3] [L0] reset and clean the command list on error unknown Signed-off-by: Neil R. Spruit --- source/adapters/level_zero/queue.cpp | 4 ---- 1 file changed, 4 deletions(-) diff --git a/source/adapters/level_zero/queue.cpp b/source/adapters/level_zero/queue.cpp index c57892c80b..c00b62b328 100644 --- a/source/adapters/level_zero/queue.cpp +++ b/source/adapters/level_zero/queue.cpp @@ -1207,10 +1207,6 @@ ur_queue_handle_t_::executeCommandList(ur_command_list_ptr_t CommandList, (ZeCommandQueue, 1, &ZeCommandList, CommandList->second.ZeFence)); if (ZeResult != ZE_RESULT_SUCCESS) { this->Healthy = false; - if (ZeResult == ZE_RESULT_ERROR_UNKNOWN) { - // Turn into a more informative end-user error. - return UR_RESULT_ERROR_UNKNOWN; - } // Reset Command List and erase the Fence forcing the user to resubmit // their commands. std::vector EventListToCleanup; From 275c753ad3eb1fc12311cd6ab5505715c590a23e Mon Sep 17 00:00:00 2001 From: "Kenneth Benzie (Benie)" Date: Thu, 25 Apr 2024 12:06:54 +0100 Subject: [PATCH 3/3] [L0] Use relative includes for adapter internals Fix build issues when using decoupled adapter fetch in intel/llvm by using relative include paths instead of longer explicit paths. This was causing redefinitions of objects. This patch also adds a missing `#pragma once` from the `adapter.hpp` header. --- source/adapters/level_zero/adapter.hpp | 1 + source/adapters/level_zero/context.cpp | 2 +- source/adapters/level_zero/queue.cpp | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/source/adapters/level_zero/adapter.hpp b/source/adapters/level_zero/adapter.hpp index 5de83b02e4..36eedb0f9d 100644 --- a/source/adapters/level_zero/adapter.hpp +++ b/source/adapters/level_zero/adapter.hpp @@ -7,6 +7,7 @@ // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception // //===----------------------------------------------------------------------===// +#pragma once #include "logger/ur_logger.hpp" #include diff --git a/source/adapters/level_zero/context.cpp b/source/adapters/level_zero/context.cpp index 782fa1c840..f9cf5009fb 100644 --- a/source/adapters/level_zero/context.cpp +++ b/source/adapters/level_zero/context.cpp @@ -13,9 +13,9 @@ #include #include -#include "adapters/level_zero/queue.hpp" #include "context.hpp" #include "logger/ur_logger.hpp" +#include "queue.hpp" #include "ur_level_zero.hpp" UR_APIEXPORT ur_result_t UR_APICALL urContextCreate( diff --git a/source/adapters/level_zero/queue.cpp b/source/adapters/level_zero/queue.cpp index 0701ad4318..c6aaf4b034 100644 --- a/source/adapters/level_zero/queue.cpp +++ b/source/adapters/level_zero/queue.cpp @@ -16,8 +16,8 @@ #include #include "adapter.hpp" -#include "adapters/level_zero/event.hpp" #include "common.hpp" +#include "event.hpp" #include "queue.hpp" #include "ur_api.h" #include "ur_level_zero.hpp"