Add fallback async IO support.

.github/workflows/build-ais.yml

@@ -19,6 +19,7 @@ env:
   AIS_INPUT_ROCM_VERSION: ${{ inputs.rocm_version }}
   # Code coverage report only vetted to work for amdclang++ on Ubuntu
   AIS_USE_CODE_COVERAGE: ${{ inputs.cxx_compiler == 'amdclang++' && inputs.platform == 'ubuntu' }}
+  AIS_HIP_ARCHITECTURES: gfx950;gfx1201;gfx1200;gfx1101;gfx1100;gfx1030;gfx942;gfx90a;gfx908
 on:
   workflow_call:
     inputs:
@@ -141,6 +142,7 @@ jobs:
               cmake \
                 -DCMAKE_CXX_COMPILER="${_AIS_INPUT_CXX_COMPILER}" \
                 -DCMAKE_CXX_FLAGS="-Werror" \
+                -DCMAKE_HIP_ARCHITECTURES="${{ env.AIS_HIP_ARCHITECTURES }}" \
                 -DCMAKE_HIP_PLATFORM=amd \
                 -DAIS_BUILD_DOCS=ON \
                 -DAIS_USE_CODE_COVERAGE=${{ env.AIS_USE_CODE_COVERAGE == 'true' && 'ON' || 'OFF' }} \

cmake/AISGNUCompilerOptions.cmake

@@ -52,7 +52,8 @@ function(get_ais_gnu_warning_flags outvar compiler_version)
         -Wformat-truncation=2
         -Wformat-y2k
         -Winvalid-pch
-        -Wlong-long
+        # This is a warning for when using <C++11
+        #-Wlong-long
         -Wlogical-op
         -Wmissing-declarations
         -Wnormalized

shared/hipfile-literals.h

@@ -0,0 +1,38 @@
+/* Copyright (c) Advanced Micro Devices, Inc. All rights reserved.
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#pragma once
+
+#include <cstddef>
+
+constexpr size_t
+operator""_KiB(unsigned long long int x)
+{
+    return 1024ULL * x;
+}
+
+constexpr size_t
+operator""_MiB(unsigned long long int x)
+{
+    return 1024_KiB * x;
+}
+
+constexpr size_t
+operator""_GiB(unsigned long long int x)
+{
+    return 1024_MiB * x;
+}
+
+constexpr size_t
+operator""_TiB(unsigned long long int x)
+{
+    return 1024_GiB * x;
+}
+
+constexpr size_t
+operator""_PiB(unsigned long long int x)
+{
+    return 1024_TiB * x;
+}

src/amd_detail/CMakeLists.txt

@@ -8,6 +8,7 @@ set(HIPFILE_SOURCES
     "${HIPFILE_SRC_COMMON_PATH}/hipfile-common.cpp"
     async.cpp
     backend/asyncop-fallback.cpp
+    backend/memcpy-kernel.hip
     backend/fallback.cpp
     backend/fastpath.cpp
     batch/batch.cpp
@@ -19,6 +20,7 @@ set(HIPFILE_SOURCES
     file-descriptor.cpp
     hip.cpp
     hipfile.cpp
+    io.cpp
     mountinfo.cpp
     state.cpp
     stream.cpp

src/amd_detail/async.h

@@ -57,8 +57,8 @@ class AsyncMonitor {
     virtual ~AsyncMonitor();
     AsyncMonitor();
 
-    void addOp(std::shared_ptr<AsyncOp> op);
-    void completeOp(AsyncOp *op);
+    virtual void addOp(std::shared_ptr<AsyncOp> op);
+    virtual void completeOp(AsyncOp *op);
 
 private:
     void                                                 completion_thread();

src/amd_detail/backend/asyncop-fallback.cpp

@@ -5,6 +5,7 @@
 
 #include "async.h"
 #include "asyncop-fallback.h"
+#include "backend.h"
 #include "buffer.h"
 #include "context.h"
 #include "hip.h"
@@ -42,10 +43,11 @@ AsyncOpFallback::AsyncOpFallback(IoType _io_type, std::shared_ptr<IFile> _file,
                                  size_t *_size, hoff_t *_file_offset, hoff_t *_buffer_offset,
                                  ssize_t *_bytes_transferred)
     : AsyncOp{_io_type, _file, _buffer, _stream, _size, _file_offset, _buffer_offset, _bytes_transferred},
-      bytes_transferred_internal{0}, gpu_buffer{buffer->getBuffer()}, bounce_buffer_dev_ptr{nullptr},
+      submitted_size{std::min(*_size, hipFile::MAX_RW_COUNT)}, bytes_transferred_internal{0},
+      gpu_buffer{buffer->getBuffer()}, bounce_buffer_dev_ptr{nullptr},
       bounce_buffer{nullptr, [](void *addr) { (void)addr; }}
 {
-    void                                            *host_ptr = Context<Hip>::get()->hipHostMalloc(*_size, 0);
+    void *host_ptr = Context<Hip>::get()->hipHostMalloc(submitted_size, 0);
     std::unique_ptr<void, decltype(&hipHostDeleter)> _bounce_buffer{host_ptr, hipHostDeleter};
     std::swap(bounce_buffer, _bounce_buffer);
     void *dev_ptr         = Context<Hip>::get()->hipHostGetDevicePointer(bounce_buffer.get(), 0);

src/amd_detail/backend/asyncop-fallback.h

@@ -26,6 +26,7 @@ enum class IoType;
 namespace hipFile {
 
 struct AsyncOpFallback : AsyncOp {
+    size_t      submitted_size;
     ssize_t     bytes_transferred_internal;
     void *const gpu_buffer;
     void       *bounce_buffer_dev_ptr;

src/amd_detail/backend/fallback.cpp

@@ -3,23 +3,32 @@
  * SPDX-License-Identifier: MIT
  */
 
+#include "async.h"
 #include "backend.h"
 #include "buffer.h"
+#include "backend/asyncop-fallback.h"
+#include "backend/memcpy-kernel.h"
 #include "context.h"
 #include "fallback.h"
 #include "file.h"
 #include "hip.h"
+#include "hipfile.h"
 #include "io.h"
 #include "sys.h"
+#include "stream.h"
+#include "util.h"
 
 #include <algorithm>
 #include <cerrno>
 #include <cstdint>
 #include <hip/hip_runtime_api.h>
 #include <hip/driver_types.h>
+#include <memory>
 #include <stdexcept>
 #include <sys/mman.h>
+#include <syslog.h>
 #include <system_error>
+#include <variant>
 
 using namespace hipFile;
 
@@ -51,24 +60,10 @@ ssize_t
 Fallback::io(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_t buflen{buffer->getLength()};
-
     size = min(size, hipFile::MAX_RW_COUNT);
 
-    if (file_offset < 0) {
-        throw std::invalid_argument("Negative file offset");
-    }
-
-    if (buffer_offset < 0) {
-        throw std::invalid_argument("Negative buffer offset");
-    }
-
-    if (buflen <= static_cast<size_t>(buffer_offset)) {
-        throw std::invalid_argument("Buffer offset larger than buffer length");
-    }
-
-    if (buflen - static_cast<size_t>(buffer_offset) < size) {
-        throw std::invalid_argument("IO could overflow buffer");
+    if (!paramsValid(buffer, size, file_offset, buffer_offset)) {
+        throw std::invalid_argument("The selected file or buffer region is invalid");
     }
 
     auto ptr     = Context<Sys>::get()->mmap(nullptr, chunk_size, PROT_READ | PROT_WRITE,
@@ -121,3 +116,162 @@ Fallback::io(IoType io_type, shared_ptr<IFile> file, shared_ptr<IBuffer> buffer,
 
     return total_io_bytes;
 }
+
+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,
+                   std::shared_ptr<IStream> stream)
+{
+    size_t limited_size = min(*size_p, hipFile::MAX_RW_COUNT);
+
+    if (!paramsValid(buffer, limited_size, *file_offset_p, *buffer_offset_p)) {
+        throw std::invalid_argument("The selected file or buffer region is invalid");
+    }
+
+    if (buffer->getGpuId() != stream->getHipDevice()) {
+        throw std::invalid_argument("Buffer GPU ID does not match Stream GPU ID");
+    }
+
+    *bytes_transferred_p = 0;
+
+    if (*size_p == 0) {
+        return;
+    }
+
+    auto op = std::shared_ptr<AsyncOpFallback>(new AsyncOpFallback(
+        type, file, buffer, stream, size_p, file_offset_p, buffer_offset_p, bytes_transferred_p));
+    Context<AsyncMonitor>::get()->addOp(op);
+    auto  op_dev_ptr     = op->devPtr();
+    void *kernel_args[1] = {&op_dev_ptr};
+
+    try {
+        int max_threads_per_block = Context<Hip>::get()->hipDeviceGetAttribute(
+            hipDeviceAttributeMaxThreadsPerBlock, buffer->getGpuId());
+        auto stream_lock = stream->getLock();
+
+        // Launch a host function to bind parameters if anything not fixed
+        if (!op->stream->fixedBufferOffset() || !op->stream->fixedFileOffset() ||
+            !op->stream->fixedIOSize()) {
+            Context<Hip>::get()->hipLaunchHostFunc(op->stream->getHipStream(), async_io_bind_params,
+                                                   op.get());
+        }
+
+        switch (op->io_type) {
+            case IoType::Read:
+                Context<Hip>::get()->hipLaunchHostFunc(op->stream->getHipStream(), async_io_cpu_copy,
+                                                       op.get());
+                Context<Hip>::get()->hipLaunchKernel(reinterpret_cast<void *>(hipFileMemcpyKernel), dim3(1),
+                                                     dim3(static_cast<uint32_t>(max_threads_per_block)),
+                                                     kernel_args, 0, op->stream->getHipStream());
+                break;
+            case IoType::Write:
+                Context<Hip>::get()->hipLaunchKernel(reinterpret_cast<void *>(hipFileMemcpyKernel), dim3(1),
+                                                     dim3(static_cast<uint32_t>(max_threads_per_block)),
+                                                     kernel_args, 0, op->stream->getHipStream());
+                Context<Hip>::get()->hipLaunchHostFunc(op->stream->getHipStream(), async_io_cpu_copy,
+                                                       op.get());
+                break;
+            default:
+                throw std::runtime_error("Invalid IO type");
+        }
+
+        Context<Hip>::get()->hipLaunchHostFunc(op->stream->getHipStream(), async_io_cleanup, op.get());
+    }
+    catch (...) {
+        // If something threw, still try to enqueue cleanup function
+        try {
+            Context<Hip>::get()->hipLaunchHostFunc(op->stream->getHipStream(), async_io_cleanup, op.get());
+        }
+        catch (...) {
+            Context<Sys>::get()->syslog(LOG_CRIT,
+                                        "Unable to enqueue async cleanup function. This will leak memory.");
+        }
+        throw;
+    }
+}
+
+extern "C" {
+void
+async_io_bind_params(void *userargs)
+{
+    auto op = static_cast<AsyncOpFallback *>(userargs);
+    // Bind params. Will maintain same value if already bound.
+    const hoff_t *buffer_offset = get_variant_ptr(op->buffer_offset);
+    op->buffer_offset.emplace<const hoff_t>(*buffer_offset);
+    const hoff_t *file_offset = get_variant_ptr(op->file_offset);
+    op->file_offset.emplace<const hoff_t>(*file_offset);
+    const size_t *size = get_variant_ptr(op->size);
+    op->size           = std::min(*size, hipFile::MAX_RW_COUNT);
+
+    if (std::get<size_t>(op->size) > op->submitted_size) {
+        op->bytes_transferred_internal = -hipFileInvalidValue;
+        return;
+    }
+    if (!paramsValid(op->buffer, std::get<size_t>(op->size), std::get<const hoff_t>(op->file_offset),
+                     std::get<const hoff_t>(op->buffer_offset))) {
+        op->bytes_transferred_internal = -hipFileInvalidValue;
+    }
+}
+
+void
+async_io_cleanup(void *userargs)
+{
+    auto     op                = static_cast<AsyncOpFallback *>(userargs);
+    ssize_t *bytes_transferred = op->bytes_transferred;
+    try {
+        Context<AsyncMonitor>::get()->completeOp(op);
+    }
+    catch (const std::invalid_argument &) {
+        *bytes_transferred = -hipFileInternalError;
+        return;
+    }
+    *bytes_transferred = op->bytes_transferred_internal;
+}
+
+void
+async_io_cpu_copy(void *userargs)
+{
+    auto         op                = static_cast<AsyncOpFallback *>(userargs);
+    size_t       bytes_transferred = 0;
+    const size_t size              = std::get<size_t>(op->size);
+    const hoff_t file_offset       = std::get<const hoff_t>(op->file_offset);
+    ssize_t      ret               = 0;
+
+    if (op->bytes_transferred_internal < 0) {
+        return;
+    }
+
+    while (bytes_transferred < size) {
+        void *cur_buf_position = reinterpret_cast<void *>(
+            reinterpret_cast<uintptr_t>(op->bounceBufferHostPtr()) + bytes_transferred);
+        hoff_t cur_file_offset = file_offset + static_cast<hoff_t>(bytes_transferred);
+        size_t remaining_bytes = size - bytes_transferred;
+        try {
+            switch (op->io_type) {
+                case IoType::Read:
+                    ret = Context<Sys>::get()->pread(op->file->getBufferedFd(), cur_buf_position,
+                                                     remaining_bytes, cur_file_offset);
+                    break;
+                case IoType::Write:
+                    ret = Context<Sys>::get()->pwrite(op->file->getBufferedFd(), cur_buf_position,
+                                                      remaining_bytes, cur_file_offset);
+                    break;
+                default:
+                    throw std::runtime_error("Invalid IO type");
+            }
+        }
+        catch (const std::system_error &e) {
+            if (e.code().value() == EINTR) {
+                continue;
+            }
+            op->bytes_transferred_internal = -1;
+            return;
+        }
+        if (ret == 0) {
+            break;
+        }
+        bytes_transferred += static_cast<size_t>(ret);
+    }
+    op->bytes_transferred_internal = static_cast<ssize_t>(bytes_transferred);
+}
+}

src/amd_detail/backend/fallback.h

@@ -5,6 +5,8 @@
 
 #pragma once
 
+#include "async.h"
+#include "backend/asyncop-fallback.h"
 #include "backend.h"
 #include "hipfile.h"
 
@@ -32,6 +34,10 @@ 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 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:
@@ -41,3 +47,9 @@ struct Fallback : public Backend {
 };
 
 }
+
+extern "C" {
+void async_io_bind_params(void *userargs);
+void async_io_cleanup(void *userargs);
+void async_io_cpu_copy(void *userargs);
+}