[hipFile] Implement execution of submitted batch operations.

projects/hipfile/include/hipfile.h

@@ -720,7 +720,9 @@ typedef struct hipFileIOParams {
 typedef struct hipFileIOEvents {
     void *cookie; //!< Optionally used to track IO operations (e.g. pointer to corresponding hipFileIOParams)
     hipFileStatus_t status; //!< Status of the batch IO operation
-    size_t          ret;    //!< Number of bytes transferred or POSIX error code if negative
+    size_t          ret;    //!< Number of bytes transferred or error code. Cast to ssize_t to inspect.
+                            //!< If non-negative, number of bytes transferred
+                            //!< If negative, POSIX or hipFileOpError_t error
 } hipFileIOEvents_t;
 
 /*!

projects/hipfile/src/amd_detail/batch/batch.cpp

@@ -8,14 +8,17 @@
 #include "context.h"
 #include "file.h"
 #include "hipfile.h"
+#include "hipfile-private.h"
 #include "state.h"
+#include "thread-pool.h"
 
 #include <bit>
 #include <cstddef>
 #include <memory>
 #include <mutex>
 #include <shared_mutex>
 #include <sstream>
+#include <stdexcept>
 #include <string>
 #include <type_traits>
 #include <unordered_map>
@@ -24,6 +27,100 @@
 
 namespace hipFile {
 
+using batchOperationState::Canceled;
+using batchOperationState::Complete;
+using batchOperationState::Failed;
+using batchOperationState::Invalid;
+using batchOperationState::OperationState;
+using batchOperationState::Pending;
+using batchOperationState::Running;
+using batchOperationState::Timeout;
+using batchOperationState::Waiting;
+
+InvalidStateTransition::InvalidStateTransition(const char *from, const char *to)
+    : std::logic_error{std::string{"Invalid batch operation state transition: "} + from + " -> " + to}
+{
+}
+
+namespace batchOperationState {
+
+    Pending Waiting::transitionTo(const Pending &next) const
+    {
+        return next;
+    }
+
+    Invalid Waiting::transitionTo(const Invalid &next) const
+    {
+        return next;
+    }
+
+    Failed Waiting::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Running Pending::transitionTo(const Running &next) const
+    {
+        return next;
+    }
+
+    Canceled Pending::transitionTo(const Canceled &next) const
+    {
+        return next;
+    }
+
+    Failed Pending::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Complete Running::transitionTo(const Complete &next) const
+    {
+        return next;
+    }
+
+    Failed Running::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Timeout Running::transitionTo(const Timeout &next) const
+    {
+        return next;
+    }
+
+    Failed Complete::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Canceled Canceled::transitionTo(const Canceled &) const
+    {
+        return *this;
+    }
+
+    Failed Canceled::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Failed Invalid::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Failed Timeout::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+    Failed Failed::transitionTo(const Failed &next) const
+    {
+        return next;
+    }
+
+}
+
 BatchOperation::BatchOperation(std::unique_ptr<const hipFileIOParams_t> params,
                                std::shared_ptr<IBuffer> _buffer, std::shared_ptr<IFile> _file)
     : io_params{std::move(params)}, buffer{std::move(_buffer)}, file{std::move(_file)}
@@ -89,6 +186,104 @@
     }
 }
 
+template <class Next>
+void
+BatchOperation::transitionTo(Next next)
+{
+    state = std::visit([&next](const auto &current) -> OperationState { return current.transitionTo(next); },
+                       state);
+    // Drop references when they're no longer needed
+    if constexpr (Next::isTerminal()) {
+        buffer.reset();
+        file.reset();
+    }
+}
+
+void
+BatchOperation::markPending()
+{
+    std::lock_guard<std::mutex> lock{state_mutex};
+
+    transitionTo(Pending{});
+}
+
+bool
+BatchOperation::tryCancel()
+{
+    std::lock_guard<std::mutex> lock{state_mutex};
+
+    try {
+        transitionTo(Canceled{});
+        return true;
+    }
+    catch (const InvalidStateTransition &) {
+        return false;
+    }
+}
+
+hipFileIOEvents_t
+BatchOperation::event() const
+{
+    std::lock_guard<std::mutex> lock{state_mutex};
+    return std::visit(
+        [this](const auto &current) -> hipFileIOEvents_t {
+            return {io_params->cookie, current.toPublic(), static_cast<size_t>(current.ret())};
+        },
+        state);
+}
+
+void
+BatchOperation::run() noexcept
+try {
+    {
+        std::lock_guard<std::mutex> lock{state_mutex};
+        if (std::holds_alternative<Canceled>(state)) {
+            return;
+        }
+        transitionTo(Running{});
+    }
+
+    ssize_t result   = 0;
+    auto    backends = Context<DriverState>::get()->getBackends();
+    try {
+        if (io_params->opcode == hipFileBatchRead) {
+            result = hipFileIo(IoType::Read, file, buffer, io_params->u.batch.size,
+                               io_params->u.batch.file_offset, io_params->u.batch.devPtr_offset, backends);
+        }
+        else {
+            result = hipFileIo(IoType::Write, file, buffer, io_params->u.batch.size,
+                               io_params->u.batch.file_offset, io_params->u.batch.devPtr_offset, backends);
+        }
+        if (result == -1) {
+            result = -errno;
+        }
+    }
+    catch (const Hip::RuntimeError &) {
+        // Return hipFileHipDriverError for HIP runtime errors to avoid
+        // collision with POSIX errors.
+        result = -hipFileHipDriverError;
+    }
+
+    std::lock_guard<std::mutex> lock{state_mutex};
+    if (result >= 0) {
+        transitionTo(Complete{result});
+    }
+    else {
+        transitionTo(Failed{result});
+    }
+}
+catch (...) {
+    recordInternalError();
+}
+
+void
+BatchOperation::recordInternalError()
+{
+    std::lock_guard<std::mutex> lock{state_mutex};
+
+    transitionTo(Failed{-hipFileInternalError});
+}
+
 BatchContext::BatchContext(unsigned _capacity) : capacity{_capacity}
 {
     if (_capacity == 0) {
@@ -97,46 +292,40 @@
     if (_capacity > MAX_SIZE) {
         throw std::invalid_argument("Batch capacity is limited to " + std::to_string(MAX_SIZE));
     }
+
+    task_group = Context<IThreadPool>::get()->makeTaskGroup();
 }
 
+BatchContext::~BatchContext() = default;
+
 unsigned
-BatchContext::get_capacity() const noexcept
+BatchContext::getCapacity() const noexcept
 {
     return capacity;
 }
 
 void
-BatchContext::submit_operations(const hipFileIOParams_t *params, unsigned num_params)
+BatchContext::submitOperations(BatchOperations pending_ops)
 {
-    std::unique_lock<std::shared_mutex> _ulock{context_mutex};
-
-    // Check num_params first before doing anything else
-    if (num_params > capacity - outstanding_ops.size()) {
-        std::stringstream msg;
-        msg << "Submission exceeds the capacity of this context. Number of ops submitted: ";
-        msg << num_params << ". Context capacity: " << capacity << ". Current outstanding ops: ";
-        msg << outstanding_ops.size();
-        throw std::invalid_argument(msg.str());
+    if (pending_ops.empty()) {
+        throw std::invalid_argument("ops must not be empty");
     }
+    {
+        std::unique_lock<std::shared_mutex> context_lock{context_mutex};
 
-    std::vector<std::shared_ptr<BatchOperation>> pending_ops{};
+        if (pending_ops.size() > capacity - outstanding_ops.size()) {
+            throw BatchFull();
+        }
 
-    // It would be more performant to be able to perform multiple lookups
-    // rather than waiting to lock the DriverState lock for each lookup.
-    for (unsigned i = 0; i < num_params; i++) {
-        // Make a copy of the params so another thread cannot modify the operation.
-        auto param_copy = std::make_unique<const hipFileIOParams_t>(params[i]);
-        // flags currently unused. Ambiguous if flags in hipFileBatchIOSubmit is for buffer or
-        // file flags.
-        auto [_file, _buffer] =
-            Context<DriverState>::get()->getFileAndBuffer(param_copy->fh, param_copy->u.batch.devPtr_base);
-        auto op = std::make_shared<BatchOperation>(std::move(param_copy), _buffer, _file);
+        for (const auto &op : pending_ops) {
+            op->markPending();
+        }
+        outstanding_ops.insert(pending_ops.begin(), pending_ops.end());
 
-        pending_ops.push_back(std::move(op));
+        for (const auto &op : pending_ops) {
+            task_group->run([op]() { op->run(); });
+        }
     }
-
-    // All submitted operations look valid at this point. Accept them.
-    outstanding_ops.insert(pending_ops.begin(), pending_ops.end());
 }
 
 void