Fix std::future status query

cpp/include/kvikio/parallel_operation.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2021-2024, NVIDIA CORPORATION.
+ * Copyright (c) 2021-2025, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,8 +17,10 @@
 
 #include <cassert>
 #include <future>
+#include <memory>
 #include <numeric>
 #include <system_error>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -30,12 +32,60 @@ namespace kvikio {
 
 namespace detail {
 
-template <typename F, typename T>
-std::future<std::size_t> submit_task(
-  F op, T buf, std::size_t size, std::size_t file_offset, std::size_t devPtr_offset)
+/**
+ * @brief Utility function to create a copyable callable from a move-only callable.
+ *
+ * The underlying thread pool uses `std::function` (until C++23) or `std::move_only_function`
+ * (since C++23) as the element type of the task queue. For the former case that currently applies,
+ * the `std::function` requires its "target" (associated callable) to be copy-constructible. This
+ * utility function is a workaround for those move-only callables.
+ *
+ * @tparam F Callable type. F shall be move-only.
+ * @param op Callable.
+ * @return A new callable that satisfies the copy-constructible condition.
+ */
+template <typename F>
+auto make_copyable_lambda(F op)
+{
+  // Create the callable on the heap by moving from op. Use a shared pointer to manage its lifetime.
+  auto sp = std::make_shared<F>(std::move(op));
+
+  // Use the copyable closure as the proxy of the move-only callable.
+  return
+    [sp](auto&&... args) -> decltype(auto) { return (*sp)(std::forward<decltype(args)>(args)...); };
+}
+
+/**
+ * @brief Submit the task callable to the underlying thread pool.
+ *
+ * Both the callable and arguments shall satisfy copy-constructible.
+ *
+ * @tparam F Callable type.
+ * @tparam Args Argument type.
+ * @param op Callable.
+ * @param args Arguments to the callable.
+ * @return A future to be used later to check if the operation has finished its execution.
+ */
+template <typename F, typename... Args>
+std::future<std::size_t> submit_task(F&& op, Args&&... args)
 {
-  return defaults::thread_pool().submit_task(
-    [=] { return op(buf, size, file_offset, devPtr_offset); });
+  static_assert(std::is_invocable_r_v<std::size_t, std::decay_t<F>, Args...>);
+  return defaults::thread_pool().submit_task([=] { return op(args...); });
+}
+
+/**
+ * @brief Submit the move-only task callable to the underlying thread pool.
+ *
+ * @tparam F Callable type. F shall be move-only and have no argument.
+ * @param op Callable.
+ * @return A future to be used later to check if the operation has finished its execution.
+ */
+template <typename F>
+std::future<std::size_t> submit_move_only_task(F op_move_only)
+{
+  static_assert(std::is_invocable_r_v<std::size_t, F>);
+  auto op_copyable = make_copyable_lambda(std::move(op_move_only));
+  return defaults::thread_pool().submit_task(op_copyable);
 }
 
 }  // namespace detail
@@ -60,6 +110,8 @@ std::future<std::size_t> parallel_io(F op,
                                      std::size_t task_size,
                                      std::size_t devPtr_offset)
 {
+  static_assert(std::is_invocable_r_v<std::size_t, F, T, std::size_t, std::size_t, std::size_t>);
+
   if (task_size == 0) { throw std::invalid_argument("`task_size` cannot be zero"); }
 
   // Single-task guard
@@ -83,14 +135,14 @@ std::future<std::size_t> parallel_io(F op,
   if (size > 0) { tasks.push_back(detail::submit_task(op, buf, size, file_offset, devPtr_offset)); }
 
   // Finally, we sum the result of all tasks.
-  auto gather_tasks = [](std::vector<std::future<std::size_t>>&& tasks) -> std::size_t {
+  auto gather_tasks = [tasks = std::move(tasks)]() mutable -> std::size_t {
     std::size_t ret = 0;
     for (auto& task : tasks) {
       ret += task.get();
     }
     return ret;
   };
-  return std::async(std::launch::deferred, gather_tasks, std::move(tasks));
+  return detail::submit_move_only_task(std::move(gather_tasks));
 }
 
 }  // namespace kvikio

cpp/include/kvikio/utils.hpp

@@ -15,6 +15,7 @@
  */
 #pragma once
 
+#include <cassert>
 #include <chrono>
 #include <cstring>
 #include <future>
@@ -149,9 +150,43 @@ class PushAndPopContext {
 std::tuple<void*, std::size_t, std::size_t> get_alloc_info(void const* devPtr,
                                                            CUcontext* ctx = nullptr);
 
+/**
+ * @brief Create a shared state in a future object that is immediately ready.
+ *
+ * A partial implementation of the namesake function from the concurrency TS
+ * (https://en.cppreference.com/w/cpp/experimental/make_ready_future). The cases of
+ * std::reference_wrapper and void are not implemented.
+ *
+ * @tparam T Type of the value provided.
+ * @param t Object provided.
+ * @return A future holding a decayed copy of the object provided.
+ */
+template <typename T>
+std::future<std::decay_t<T>> make_ready_future(T&& t)
+{
+  std::promise<std::decay_t<T>> p;
+  auto fut = p.get_future();
+  p.set_value(std::forward<T>(t));
+  return fut;
+}
+
+/**
+ * @brief Check the status of the future object. True indicates that the result is available in the
+ * future's shared state. False otherwise.
+ *
+ * The future shall not be created using `std::async(std::launch::deferred)`. Otherwise, this
+ * function always returns true.
+ *
+ * @tparam T Type of the future.
+ * @param future Instance of the future.
+ * @return Boolean answer indicating if the future is ready or not.
+ */
 template <typename T>
 bool is_future_done(T const& future)
 {
+  if (!future.valid()) {
+    throw std::invalid_argument("The future object does not refer to a valid shared state.");
+  }
   return future.wait_for(std::chrono::seconds(0)) != std::future_status::timeout;
 }
 

cpp/src/file_handle.cpp

@@ -26,6 +26,7 @@
 #include <kvikio/defaults.hpp>
 #include <kvikio/file_handle.hpp>
 #include <kvikio/file_utils.hpp>
+#include "kvikio/utils.hpp"
 
 namespace kvikio {
 
@@ -210,11 +211,11 @@ std::future<std::size_t> FileHandle::pread(void* buf,
 
   // Shortcut that circumvent the threadpool and use the POSIX backend directly.
   if (size < gds_threshold) {
-    auto task = [this, ctx, buf, size, file_offset]() -> std::size_t {
-      PushAndPopContext c(ctx);
-      return detail::posix_device_read(_fd_direct_off.fd(), buf, size, file_offset, 0);
-    };
-    return std::async(std::launch::deferred, task);
+    PushAndPopContext c(ctx);
+    auto bytes_read = detail::posix_device_read(_fd_direct_off.fd(), buf, size, file_offset, 0);
+    // Maintain API consistency while making this trivial case synchronous.
+    // The result in the future is immediately available after the call.
+    return make_ready_future(bytes_read);
   }
 
   // Let's synchronize once instead of in each task.
@@ -260,11 +261,11 @@ std::future<std::size_t> FileHandle::pwrite(void const* buf,
 
   // Shortcut that circumvent the threadpool and use the POSIX backend directly.
   if (size < gds_threshold) {
-    auto task = [this, ctx, buf, size, file_offset]() -> std::size_t {
-      PushAndPopContext c(ctx);
-      return detail::posix_device_write(_fd_direct_off.fd(), buf, size, file_offset, 0);
-    };
-    return std::async(std::launch::deferred, task);
+    PushAndPopContext c(ctx);
+    auto bytes_write = detail::posix_device_write(_fd_direct_off.fd(), buf, size, file_offset, 0);
+    // Maintain API consistency while making this trivial case synchronous.
+    // The result in the future is immediately available after the call.
+    return make_ready_future(bytes_write);
   }
 
   // Let's synchronize once instead of in each task.