Backend: Add BackendWithFallback to retry IO

CHANGELOG.md

@@ -12,6 +12,7 @@
 * The CMake namespace was changed from `roc::` to `hip::`
 * `AIS_BUILD_EXAMPLES` has been renamed to `AIS_INSTALL_EXAMPLES`
 * `AIS_USE_SANITIZERS` now also enables the following sanitizers: integer, float-divide-by-zero, local-bounds, vptr, nullability (in addition to address, leak, and undefined). Sanitizers should also now emit usable stack trace info.
+* The AIS optimized IO path will automatically fallback to the POSIX IO path if a failure occurs and the compatability mode has not been disabled.
 
 ### Removed
 * The rocFile library has been completely removed and the code is now a part of hipFile.

src/amd_detail/CMakeLists.txt

@@ -7,6 +7,7 @@
 set(HIPFILE_SOURCES
     "${HIPFILE_SRC_COMMON_PATH}/hipfile-common.cpp"
     async.cpp
+    backend.cpp
     backend/asyncop-fallback.cpp
     backend/memcpy-kernel.hip
     backend/fallback.cpp

src/amd_detail/backend.cpp

@@ -0,0 +1,89 @@
+/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include "backend.h"
+#include "buffer.h"
+#include "file.h"
+#include "io.h"
+
+#include <cstddef>
+#include <exception>
+#include <memory>
+#include <sys/types.h>
+#include <system_error>
+#include <unistd.h>
+
+using namespace hipFile;
+
+ssize_t
+Backend::io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
+            hoff_t file_offset, hoff_t buffer_offset)
+{
+    ssize_t nbytes = _io_impl(type, file, buffer, size, file_offset, buffer_offset);
+    switch (type) {
+        case (IoType::Read):
+            update_read_stats(nbytes);
+            break;
+        case (IoType::Write):
+            update_write_stats(nbytes);
+            break;
+        default:
+            break;
+    }
+    return nbytes;
+}
+
+ssize_t
+BackendWithFallback::io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer,
+                        size_t size, hoff_t file_offset, hoff_t buffer_offset)
+{
+    ssize_t nbytes{0};
+    try {
+        nbytes = _io_impl(type, file, buffer, size, file_offset, buffer_offset);
+        if (nbytes < 0) {
+            // Typically we should not reach this point. But in case we do, throw
+            // an exception to use the fallback backend.
+            throw std::system_error(-static_cast<int>(nbytes), std::generic_category());
+        }
+    }
+    catch (...) {
+        std::exception_ptr e_ptr = std::current_exception();
+        if (is_fallback_eligible(e_ptr, nbytes, file, buffer, size, file_offset, buffer_offset)) {
+            nbytes = fallback_backend->io(type, file, buffer, size, file_offset, buffer_offset);
+        }
+        else {
+            throw;
+        }
+        return nbytes;
+    }
+    switch (type) {
+        case (IoType::Read):
+            update_read_stats(nbytes);
+            break;
+        case (IoType::Write):
+            update_write_stats(nbytes);
+            break;
+        default:
+            break;
+    }
+    return nbytes;
+}
+
+bool
+BackendWithFallback::is_fallback_eligible(std::exception_ptr e_ptr, ssize_t nbytes,
+                                          std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer,
+                                          size_t size, hoff_t file_offset, hoff_t buffer_offset) const
+{
+    (void)e_ptr;
+    (void)nbytes;
+    return static_cast<bool>(fallback_backend) &&
+           fallback_backend->score(file, buffer, size, file_offset, buffer_offset) >= 0;
+}
+
+void
+BackendWithFallback::register_fallback_backend(std::shared_ptr<Backend> backend) noexcept
+{
+    fallback_backend = backend;
+}

src/amd_detail/backend.h

@@ -12,6 +12,7 @@
 #include "sys.h"
 
 #include <cstddef>
+#include <exception>
 #include <memory>
 #include <sys/types.h>
 #include <unistd.h>
@@ -51,11 +52,74 @@ struct Backend {
     /// @param file_offset   Offset from the start of the file
     /// @param buffer_offset Offset from the start of the buffer
     ///
-    /// @return Number of bytes transferred, negative on error
+    /// @return Number of bytes transferred
     ///
     /// @throws Hip::RuntimeError Sys::RuntimeError
     virtual ssize_t io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
-                       hoff_t file_offset, hoff_t buffer_offset) = 0;
+                       hoff_t file_offset, hoff_t buffer_offset);
+
+    /// @brief Update the read stats for this Backend
+    ///
+    /// @param nbytes Number of bytes read
+    virtual void update_read_stats(ssize_t nbytes) = 0;
+
+    /// @brief Update the write stats for this Backend
+    ///
+    /// @param nbytes Number of bytes written
+    virtual void update_write_stats(ssize_t nbytes) = 0;
+
+protected:
+    /// @brief Perform a read or write operation
+    ///
+    /// @note Provides a common target across all Backends that provides the
+    ///       implementation for running IO.
+    /// @param type          IO type (read/write)
+    /// @param file          File to read from or write to
+    /// @param buffer        Buffer to write to or read from
+    /// @param size          Number of bytes to transfer
+    /// @param file_offset   Offset from the start of the file
+    /// @param buffer_offset Offset from the start of the buffer
+    ///
+    /// @return Number of bytes transferred
+    ///
+    /// @throws Hip::RuntimeError Sys::RuntimeError
+    virtual ssize_t _io_impl(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer,
+                             size_t size, hoff_t file_offset, hoff_t buffer_offset) = 0;
+};
+
+// BackendWithFallback allows for an IO to be retried automatically with a
+// different Backend in the event of an error.
+struct BackendWithFallback : public Backend {
+    ssize_t io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
+               hoff_t file_offset, hoff_t buffer_offset) override final;
+
+    /// @brief Check if a failed IO operation can be re-issued to the fallback Backend.
+    ///
+    /// @param e_ptr         exception_ptr to the thrown exception from the failed IO
+    /// @param nbytes        Return value from `_io_impl`, or 0 if an exception was thrown.
+    /// @param file          File to read from or write to
+    /// @param buffer        Buffer to write to or read from
+    /// @param size          Number of bytes to transfer
+    /// @param file_offset   Offset from the start of the file
+    /// @param buffer_offset Offset from the start of the buffer
+    ///
+    /// @note By default, BackendWithFallback checks if a Backend has been
+    ///       registered for retrying an IO, and that fallback backend supports
+    ///       the request.
+    /// @note The parameters from the original IO request are passed to this function.
+    ///
+    /// @return True if this BackendWithFallback can retry the IO, else False.
+    virtual bool is_fallback_eligible(std::exception_ptr e_ptr, ssize_t nbytes, std::shared_ptr<IFile> file,
+                                      std::shared_ptr<IBuffer> buffer, size_t size, hoff_t file_offset,
+                                      hoff_t buffer_offset) const;
+
+    /// @brief Register a Backend to retry a failed IO operation.
+    ///
+    /// @param backend Backend to retry a failed IO operation.
+    void register_fallback_backend(std::shared_ptr<Backend> backend) noexcept;
+
+protected:
+    std::shared_ptr<Backend> fallback_backend;
 };
 
 }

src/amd_detail/backend/fallback.cpp

@@ -58,8 +58,33 @@ Fallback::io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer>
 }
 
 ssize_t
-Fallback::io(IoType io_type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer, size_t size,
+Fallback::io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
              hoff_t file_offset, hoff_t buffer_offset, size_t chunk_size)
+{
+    ssize_t nbytes = _io_impl(type, file, buffer, size, file_offset, buffer_offset, chunk_size);
+    switch (type) {
+        case (IoType::Read):
+            update_read_stats(nbytes);
+            break;
+        case (IoType::Write):
+            update_write_stats(nbytes);
+            break;
+        default:
+            break;
+    }
+    return nbytes;
+}
+
+ssize_t
+Fallback::_io_impl(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
+                   hoff_t file_offset, hoff_t buffer_offset)
+{
+    return _io_impl(type, file, buffer, size, file_offset, buffer_offset, DefaultChunkSize);
+}
+
+ssize_t
+Fallback::_io_impl(IoType io_type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer, size_t size,
+                   hoff_t file_offset, hoff_t buffer_offset, size_t chunk_size)
 {
     size = min(size, hipFile::MAX_RW_COUNT);
 
@@ -115,19 +140,21 @@ Fallback::io(IoType io_type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer,
         }
     } while (static_cast<size_t>(total_io_bytes) < size);
 
-    switch (io_type) {
-        case IoType::Read:
-            statsAddFallbackPathRead(static_cast<size_t>(total_io_bytes));
-            break;
-        case IoType::Write:
-            statsAddFallbackPathWrite(static_cast<size_t>(total_io_bytes));
-            break;
-        default:
-            break;
-    }
     return total_io_bytes;
 }
 
+void
+Fallback::update_read_stats(ssize_t nbytes)
+{
+    statsAddFallbackPathRead(static_cast<size_t>(nbytes));
+}
+
+void
+Fallback::update_write_stats(ssize_t nbytes)
+{
+    statsAddFallbackPathWrite(static_cast<size_t>(nbytes));
+}
+
 void
 Fallback::async_io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t *size_p,
                    hoff_t *file_offset_p, hoff_t *buffer_offset_p, ssize_t *bytes_transferred_p,

src/amd_detail/backend/fallback.h

@@ -34,16 +34,24 @@ struct Fallback : public Backend {
     ssize_t io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
                hoff_t file_offset, hoff_t buffer_offset) override;
 
+    void update_read_stats(ssize_t nbytes) override;
+
+    void update_write_stats(ssize_t nbytes) override;
+
     void async_io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t *size_p,
                   hoff_t *file_offset_p, hoff_t *buffer_offset_p, ssize_t *bytes_transferred_p,
                   std::shared_ptr<IStream> stream);
 
     // Once we can import gtest.h and make test suites or test friends everything
     // below here should be made protected.
-    // protected:
-
     ssize_t io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
                hoff_t file_offset, hoff_t buffer_offset, size_t chunk_size);
+
+protected:
+    ssize_t _io_impl(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
+                     hoff_t file_offset, hoff_t buffer_offset) override;
+    ssize_t _io_impl(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
+                     hoff_t file_offset, hoff_t buffer_offset, size_t chunk_size);
 };
 
 }

src/amd_detail/backend/fastpath.cpp

@@ -154,9 +154,21 @@ Fastpath::score(shared_ptr<IFile> file, shared_ptr<IBuffer> buffer, size_t size,
     return accept_io ? 100 : -1;
 }
 
+void
+Fastpath::update_read_stats(ssize_t nbytes)
+{
+    statsAddFastPathRead(static_cast<uint64_t>(nbytes));
+}
+
+void
+Fastpath::update_write_stats(ssize_t nbytes)
+{
+    statsAddFastPathWrite(static_cast<uint64_t>(nbytes));
+}
+
 ssize_t
-Fastpath::io(IoType type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer, size_t size, hoff_t file_offset,
-             hoff_t buffer_offset)
+Fastpath::_io_impl(IoType type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer, size_t size,
+                   hoff_t file_offset, hoff_t buffer_offset)
 {
     void *devptr{reinterpret_cast<void *>(reinterpret_cast<intptr_t>(buffer->getBuffer()) + buffer_offset)};
     hipAmdFileHandle_t handle{};
@@ -184,15 +196,5 @@ Fastpath::io(IoType type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer, si
         default:
             throw std::runtime_error("Invalid IoType");
     }
-    switch (type) {
-        case IoType::Read:
-            statsAddFastPathRead(nbytes);
-            break;
-        case IoType::Write:
-            statsAddFastPathWrite(nbytes);
-            break;
-        default:
-            break;
-    }
     return static_cast<ssize_t>(nbytes);
 }

src/amd_detail/backend/fastpath.h

@@ -9,6 +9,7 @@
 #include "hipfile.h"
 
 #include <memory>
+#include <stdexcept>
 #include <sys/types.h>
 
 namespace hipFile {
@@ -23,14 +24,17 @@ enum class IoType;
 
 namespace hipFile {
 
-struct Fastpath : public Backend {
+struct Fastpath : public BackendWithFallback {
     virtual ~Fastpath() override = default;
 
-    int score(std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size, hoff_t file_offset,
-              hoff_t buffer_offset) const override;
+    int  score(std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size, hoff_t file_offset,
+               hoff_t buffer_offset) const override;
+    void update_read_stats(ssize_t nbytes) override;
+    void update_write_stats(ssize_t nbytes) override;
 
-    ssize_t io(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
-               hoff_t file_offset, hoff_t buffer_offset) override;
+protected:
+    ssize_t _io_impl(IoType type, std::shared_ptr<IFile> file, std::shared_ptr<IBuffer> buffer, size_t size,
+                     hoff_t file_offset, hoff_t buffer_offset) override;
 };
 
 }

src/amd_detail/state.cpp

@@ -275,11 +275,18 @@ std::vector<std::shared_ptr<Backend>>
 DriverState::getBackends() const
 {
     static bool once = [&]() {
-        if (Context<Configuration>::get()->fastpath()) {
-            backends.emplace_back(new Fastpath{});
-        }
+        std::shared_ptr<Backend> fallback_backend;
         if (Context<Configuration>::get()->fallback()) {
-            backends.emplace_back(new Fallback{});
+            fallback_backend = std::make_shared<Fallback>();
+            backends.push_back(fallback_backend);
+        }
+
+        if (Context<Configuration>::get()->fastpath()) {
+            auto new_backend = std::make_shared<Fastpath>();
+            if (fallback_backend) {
+                new_backend->register_fallback_backend(fallback_backend);
+            }
+            backends.push_back(new_backend);
         }
         return true;
     }();

test/amd_detail/CMakeLists.txt

@@ -12,6 +12,7 @@ set(SHARED_SOURCE_FILES
 
 set(TEST_SOURCE_FILES
     async.cpp
+    backend.cpp
     batch/batch.cpp
     configuration.cpp
     context.cpp