[hipfile] Add thread pool for batch operations.

projects/hipfile/cmake/AISAddExecutable.cmake

@@ -33,9 +33,6 @@ function(ais_add_executable)
     target_compile_features(${arg_NAME} PRIVATE cxx_std_${AIS_CXX_STANDARD})
     set_target_properties(${arg_NAME} PROPERTIES CXX_EXTENSIONS OFF)
 
-    # Turn sanitizers off for executables
-    target_compile_options(${arg_NAME} PRIVATE -fno-sanitize=all)
-
     get_target_property(linker_language ${arg_NAME} LINKER_LANGUAGE)
     if(linker_language STREQUAL "HIP")
         target_link_libraries(${arg_NAME} PRIVATE hip::device)

projects/hipfile/cmake/AISGNUCompilerOptions.cmake

@@ -65,7 +65,7 @@ function(get_ais_gnu_warning_flags outvar compiler_version)
         -Wshadow
         -Wshadow-local
         -Wshift-overflow=2
-        -Wstrict-overflow=4
+        -Wno-strict-overflow
         -Wswitch-default
         -Wswitch-enum
         -Wtrampolines

projects/hipfile/cmake/AISSanitizers.cmake

@@ -27,16 +27,18 @@ function(ais_add_sanitizers target)
 
     if(AIS_USE_SANITIZERS)
         set(SANITIZER_LIST address,float-divide-by-zero,integer,leak,local-bounds,nullability,undefined,vptr)
-        # unsigned-integer-overflow is well-defined wrap-around in C++ (not UB).
-        # Keep -fsanitize=integer enabled, but disable the unsigned overflow check.
+        # unsigned-integer-overflow and unsigned-shift-base are not UB.
+        # Keep -fsanitize=integer enabled, but disable both checks.
         target_compile_options(${target} PRIVATE
             $<$<COMPILE_LANGUAGE:CXX>:-fsanitize=${SANITIZER_LIST}>
             $<$<COMPILE_LANGUAGE:CXX>:-fno-sanitize=unsigned-integer-overflow>
+            $<$<COMPILE_LANGUAGE:CXX>:-fno-sanitize=unsigned-shift-base>
             $<$<COMPILE_LANGUAGE:CXX>:-fno-sanitize-recover=${SANITIZER_LIST}>
         )
         target_link_options(${target} PRIVATE
             -Xarch_host -fsanitize=${SANITIZER_LIST}
             -Xarch_host -fno-sanitize=unsigned-integer-overflow
+            -Xarch_host -fno-sanitize=unsigned-shift-base
             -Xarch_host -fno-sanitize-recover=${SANITIZER_LIST}
         )
     endif()

projects/hipfile/src/amd_detail/CMakeLists.txt

@@ -30,6 +30,7 @@ set(HIPFILE_SOURCES
     stats.cpp
     stream.cpp
     sys.cpp
+    thread-pool.cpp
 )
 
 # Create hipfile target
@@ -39,9 +40,21 @@ ais_add_libraries(
     LIBS     mount
     SRCS     ${HIPFILE_SOURCES}
     LIBINCL  ${HIPFILE_INCLUDE_PATH}
-    SYSINCLS ${HIPFILE_AMD_SOURCE_PATH}
+    SYSINCLS ${HIPFILE_AMD_SOURCE_PATH} ${HIPFILE_THIRD_PARTY_PATH}
 )
 
+# There are some warnings that appear while compiling thread-pool.cpp due
+# to inlining from Taskflow, so we see these warnings even with third-party
+# set as a system include directory. Disable these warnings.
+if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
+    set_source_files_properties(
+        thread-pool.cpp
+        PROPERTIES
+            COMPILE_OPTIONS
+                "-Wno-null-dereference;-Wno-array-bounds"
+    )
+endif()
+
 if(HIPFILE_ROCPROFILER_REGISTER)
     target_compile_definitions(
         hipfile

projects/hipfile/src/amd_detail/thread-pool.cpp

@@ -0,0 +1,146 @@
+/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include "taskflow/taskflow.hpp"
+#include "thread-pool.h"
+
+#include <algorithm>
+#include <condition_variable>
+#include <cstdint>
+#include <exception>
+#include <mutex>
+#include <utility>
+
+namespace hipFile {
+
+namespace {
+
+    class TaskflowTaskGroup final : public ITaskGroup {
+    public:
+        explicit TaskflowTaskGroup(std::shared_ptr<tf::Executor> _executor)
+            : executor{std::move(_executor)}, state{std::make_shared<State>()}
+        {
+        }
+
+        ~TaskflowTaskGroup() override
+        {
+            cancel();
+            wait();
+        }
+
+        void run(std::function<void()> task) override
+        {
+            uint64_t task_generation = 0;
+            auto     task_state      = state;
+
+            {
+                std::lock_guard<std::mutex> lock{task_state->mutex};
+
+                task_generation = task_state->generation;
+                task_state->outstanding++;
+            }
+
+            try {
+                executor->silent_async([task_state, task_generation, work = std::move(task)]() mutable {
+                    Completion            completion{task_state};
+                    std::function<void()> local_work;
+                    local_work.swap(work);
+
+                    {
+                        std::lock_guard<std::mutex> lock{task_state->mutex};
+                        if (task_generation != task_state->generation) {
+                            return;
+                        }
+                    }
+
+                    local_work();
+                });
+            }
+            catch (...) {
+                finish(task_state);
+                throw;
+            }
+        }
+
+        void cancel() override
+        {
+            std::lock_guard<std::mutex> lock{state->mutex};
+
+            state->generation++;
+        }
+
+        void wait() override
+        {
+            std::unique_lock<std::mutex> lock{state->mutex};
+
+            state->cv.wait(lock, [this]() { return state->outstanding == 0; });
+        }
+
+    private:
+        struct State {
+            std::mutex              mutex;
+            std::condition_variable cv;
+            uint64_t                generation  = 0;
+            size_t                  outstanding = 0;
+        };
+
+        class Completion {
+        public:
+            explicit Completion(std::shared_ptr<State> _state) : state{std::move(_state)}
+            {
+            }
+
+            ~Completion()
+            {
+                finish(state);
+            }
+
+            Completion(const Completion &)            = delete;
+            Completion &operator=(const Completion &) = delete;
+
+            Completion(Completion &&)            = delete;
+            Completion &operator=(Completion &&) = delete;
+
+        private:
+            std::shared_ptr<State> state;
+        };
+
+        static void finish(const std::shared_ptr<State> &state)
+        {
+            {
+                std::lock_guard<std::mutex> lock{state->mutex};
+
+                state->outstanding--;
+            }
+            state->cv.notify_all();
+        }
+
+        std::shared_ptr<tf::Executor> executor;
+        std::shared_ptr<State>        state;
+    };
+
+}
+
+struct ThreadPool::Impl {
+    explicit Impl(size_t workers) : executor{std::make_shared<tf::Executor>(std::max<size_t>(workers, 1))}
+    {
+    }
+
+    std::shared_ptr<tf::Executor> executor;
+};
+
+ThreadPool::ThreadPool(size_t workers) : impl{std::make_unique<Impl>(workers)}
+{
+}
+
+ThreadPool::~ThreadPool() = default;
+
+std::unique_ptr<ITaskGroup>
+ThreadPool::makeTaskGroup()
+{
+    return std::make_unique<TaskflowTaskGroup>(impl->executor);
+}
+
+}

projects/hipfile/src/amd_detail/thread-pool.h

@@ -0,0 +1,67 @@
+/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#pragma once
+
+#include "context.h"
+
+#include <cstddef>
+#include <functional>
+#include <memory>
+#include <thread>
+
+namespace hipFile {
+
+/// @brief Manage a set of tasks on a thread pool
+/// Task groups manage a set of tasks. Multiple task groups can share the
+/// same thread pool.
+class ITaskGroup {
+public:
+    virtual ~ITaskGroup() = default;
+
+    /// @brief Run a task function on the thread pool
+    ///
+    /// @param task Task function to run
+    virtual void run(std::function<void()> task) = 0;
+    /// @brief Cancel all tasks in task group
+    virtual void cancel() = 0;
+    /// @brief Wait until no tasks are outstanding in task group
+    virtual void wait() = 0;
+};
+
+class IThreadPool {
+public:
+    virtual ~IThreadPool() = default;
+
+    /// @brief Construct a task group on the thread pool
+    virtual std::unique_ptr<ITaskGroup> makeTaskGroup() = 0;
+};
+
+class ThreadPool : public IThreadPool {
+public:
+    /// @brief Construct a threadpool
+    ///
+    /// @param workers Number of worker threads
+    explicit ThreadPool(size_t workers = std::thread::hardware_concurrency());
+    ~ThreadPool() override;
+
+    ThreadPool(const ThreadPool &)            = delete;
+    ThreadPool &operator=(const ThreadPool &) = delete;
+
+    ThreadPool(ThreadPool &&)            = delete;
+    ThreadPool &operator=(ThreadPool &&) = delete;
+
+    /// @brief Construct a task group on the threadpool
+    std::unique_ptr<ITaskGroup> makeTaskGroup() override;
+
+private:
+    struct Impl;
+
+    std::unique_ptr<Impl> impl;
+};
+
+HIPFILE_CONTEXT_DEFAULT_IMPL(IThreadPool, ThreadPool);
+
+}

projects/hipfile/test/amd_detail/CMakeLists.txt

@@ -30,6 +30,7 @@ set(TEST_SOURCE_FILES
     mountinfo.cpp
     stats.cpp
     stream.cpp
+    thread-pool.cpp
 )
 
 set(TEST_SYSINCLS

projects/hipfile/test/amd_detail/fallback.cpp

@@ -93,7 +93,7 @@ contains_expected_data(std::vector<uint8_t> &buffer, hoff_t buffer_offset, std::
         }
     }
 
-    if (memcmp(buffer.data() + buffer_offset, expected.data() + expected_offset, count)) {
+    if (count > 0 && memcmp(buffer.data() + buffer_offset, expected.data() + expected_offset, count)) {
         return false;
     }
 
@@ -349,7 +349,8 @@ struct FallbackWrite : public FallbackIo {
             file_data.resize(uoffset + count);
         }
 
-        memcpy(file_data.data() + uoffset, buf, count);
+        if (count > 0)
+            memcpy(file_data.data() + uoffset, buf, count);
 
         return static_cast<ssize_t>(count);
     }
@@ -544,7 +545,8 @@ struct FallbackRead : public FallbackIo {
             count = file_data.size() - uoffset;
         }
 
-        memcpy(buf, file_data.data() + uoffset, count);
+        if (count > 0)
+            memcpy(buf, file_data.data() + uoffset, count);
 
         return static_cast<ssize_t>(count);
     }

projects/hipfile/test/amd_detail/mthread-pool.h

@@ -0,0 +1,36 @@
+/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#pragma once
+
+#include "context.h"
+#include "thread-pool.h"
+
+#include <functional>
+#include <memory>
+
+#include <gmock/gmock.h>
+
+namespace hipFile {
+
+class MTaskGroup : public ITaskGroup {
+public:
+    MOCK_METHOD(void, run, (std::function<void()> work), (override));
+    MOCK_METHOD(void, cancel, (), (override));
+    MOCK_METHOD(void, wait, (), (override));
+};
+
+class MThreadPool : public IThreadPool {
+public:
+    ContextOverride<IThreadPool> o_co;
+
+    MThreadPool() : o_co{this}
+    {
+    }
+
+    MOCK_METHOD(std::unique_ptr<ITaskGroup>, makeTaskGroup, (), (override));
+};
+
+}