enhance: add async reads to InputStream

include/filemanager/InputStream.h

@@ -11,6 +11,10 @@
 
 #pragma once
 
+#include <folly/futures/Future.h>
+
+#include <cstddef>
+#include <cstdint>
 #include <string>
 
 namespace milvus {
@@ -74,6 +78,27 @@ class InputStream {
     virtual size_t
     ReadAt(void* ptr, size_t offset, size_t size) = 0;
 
+    /**
+     * @brief asynchronously reads bytes into ptr at the given offset
+     *
+     * The default implementation defers the synchronous ReadAt call until the
+     * returned future is driven. Callers can attach an executor with via() to
+     * keep the synchronous fallback off the current thread. Implementations
+     * backed by native asynchronous IO should override this method.
+     *
+     * The stream and ptr must remain valid until the future completes.
+     *
+     * @param ptr
+     * @param offset
+     * @param size
+     * @return a future containing the number of bytes read
+     */
+    virtual folly::SemiFuture<size_t>
+    ReadAtAsync(void* ptr, size_t offset, size_t size) {
+        return folly::makeSemiFuture().deferValue(
+            [this, ptr, offset, size](folly::Unit) { return ReadAt(ptr, offset, size); });
+    }
+
     /**
      * @brief read data from the stream to a object with given type
      *

test/StreamTest.cpp

@@ -1,3 +1,4 @@
+#include <folly/executors/InlineExecutor.h>
 #include <gtest/gtest.h>
 
 #include <algorithm>
@@ -209,6 +210,32 @@ TEST_F(StreamTest, LocalInputStream_ReadAt) {
     EXPECT_TRUE(std::equal(read_data.begin(), read_data.end(), data.begin() + 50));
 }
 
+TEST_F(StreamTest, LocalInputStream_ReadAtAsyncDefersRead) {
+    const std::vector<uint8_t> data = {1, 2, 3, 4, 5, 6};
+    WriteTestFile(data);
+
+    LocalInputStream in(temp_file_);
+    std::vector<uint8_t> read_data(3, 0);
+
+    auto future = in.ReadAtAsync(read_data.data(), 2, read_data.size());
+    EXPECT_EQ(read_data, std::vector<uint8_t>({0, 0, 0}));
+
+    auto bytes_read = std::move(future).via(&folly::InlineExecutor::instance()).get();
+    EXPECT_EQ(bytes_read, read_data.size());
+    EXPECT_EQ(read_data, std::vector<uint8_t>({3, 4, 5}));
+}
+
+TEST_F(StreamTest, LocalInputStream_ReadAtAsyncPropagatesError) {
+    const std::vector<uint8_t> data = {1, 2, 3, 4};
+    WriteTestFile(data);
+
+    LocalInputStream in(temp_file_);
+    std::vector<uint8_t> read_data(3);
+
+    auto future = in.ReadAtAsync(read_data.data(), 2, read_data.size());
+    EXPECT_THROW(std::move(future).via(&folly::InlineExecutor::instance()).get(), std::runtime_error);
+}
+
 TEST_F(StreamTest, LocalInputStream_ReadAtConcurrent) {
     auto data = GenerateTestData(10000);
     WriteTestFile(data);