Forward-merge release/26.10 into main

cpp/include/kvikio/defaults.hpp

@@ -487,6 +487,18 @@ class defaults {
    */
   [[nodiscard]] static unsigned int remote_io_num_reactors();
 
+  /**
+   * @brief Set the number of reactor threads used by the `MULTI_POLL` remote I/O backend at
+   * runtime, overriding `KVIKIO_REMOTE_IO_NUM_REACTORS`.
+   *
+   * The pool is created lazily on first use and is never rebuilt or resized.
+   *
+   * @param num_reactors The number of reactor threads. Must be a positive integer.
+   *
+   * @exception std::runtime_error if the `MULTI_POLL` reactor pool has already been created.
+   */
+  static void set_remote_io_num_reactors(unsigned int num_reactors);
+
   /**
    * @brief How sub-ranges of one `pread()` are distributed across reactor threads under the
    * `MULTI_POLL` remote I/O backend.
@@ -500,6 +512,18 @@ class defaults {
    */
   [[nodiscard]] static RemoteReactorDispatch remote_io_reactor_dispatch();
 
+  /**
+   * @brief Set the reactor dispatch policy used by the `MULTI_POLL` remote I/O backend at
+   * runtime, overriding `KVIKIO_REMOTE_IO_REACTOR_DISPATCH`.
+   *
+   * The pool is created lazily on first use and is never rebuilt or resized.
+   *
+   * @param dispatch The reactor dispatch policy.
+   *
+   * @exception std::runtime_error if the `MULTI_POLL` reactor pool has already been created.
+   */
+  static void set_remote_io_reactor_dispatch(RemoteReactorDispatch dispatch);
+
   /**
    * @brief Maximum number of concurrent in-flight requests across all reactor threads under the
    * `MULTI_POLL` remote I/O backend.
@@ -519,6 +543,19 @@ class defaults {
    * @return The configured concurrent-request ceiling, or 0 for unlimited.
    */
   [[nodiscard]] static std::size_t remote_io_max_concurrent_requests();
+
+  /**
+   * @brief Set the maximum number of concurrent in-flight requests across all reactor threads
+   * under the `MULTI_POLL` remote I/O backend at runtime, overriding
+   * `KVIKIO_REMOTE_IO_MAX_CONCURRENT_REQUESTS`.
+   *
+   * The pool is created lazily on first use and is never rebuilt or resized.
+   *
+   * @param max_requests The concurrent-request ceiling. 0 means unlimited.
+   *
+   * @exception std::runtime_error if the `MULTI_POLL` reactor pool has already been created.
+   */
+  static void set_remote_io_max_concurrent_requests(std::size_t max_requests);
 };
 
 }  // namespace kvikio

cpp/include/kvikio/detail/multi_poll_reactor.hpp

@@ -318,6 +318,15 @@ class MultiReactorPool {
    */
   static MultiReactorPool& instance();
 
+  /**
+   * @brief Whether the pool singleton has already been constructed.
+   *
+   * `num_reactors`, the dispatch mode, and the concurrency cap are all captured once in the
+   * pool's constructor, so changing them after this returns `true` would silently have no effect.
+   * Used by `kvikio::defaults` to reject such changes with an exception instead.
+   */
+  [[nodiscard]] static bool is_instantiated() noexcept;
+
   MultiReactorPool(MultiReactorPool const&)            = delete;
   MultiReactorPool& operator=(MultiReactorPool const&) = delete;
   MultiReactorPool(MultiReactorPool&&)                 = delete;

cpp/src/defaults.cpp

@@ -12,6 +12,9 @@
 
 #include <kvikio/compat_mode.hpp>
 #include <kvikio/defaults.hpp>
+#ifdef KVIKIO_LIBCURL_FOUND
+#include <kvikio/detail/multi_poll_reactor.hpp>
+#endif
 #include <kvikio/detail/nvtx.hpp>
 #include <kvikio/detail/utils.hpp>
 #include <kvikio/error.hpp>
@@ -304,13 +307,48 @@ void defaults::set_remote_io_backend(RemoteIOBackend backend)
 
 unsigned int defaults::remote_io_num_reactors() { return instance()->_remote_io_num_reactors; }
 
+void defaults::set_remote_io_num_reactors(unsigned int num_reactors)
+{
+  KVIKIO_EXPECT(
+    num_reactors > 0, "remote_io_num_reactors must be a positive integer", std::invalid_argument);
+#ifdef KVIKIO_LIBCURL_FOUND
+  KVIKIO_EXPECT(!detail::MultiReactorPool::is_instantiated(),
+                "remote_io_num_reactors cannot be changed after the MULTI_POLL reactor pool has "
+                "already started",
+                std::runtime_error);
+#endif
+  instance()->_remote_io_num_reactors = num_reactors;
+}
+
 RemoteReactorDispatch defaults::remote_io_reactor_dispatch()
 {
   return instance()->_remote_io_reactor_dispatch;
 }
 
+void defaults::set_remote_io_reactor_dispatch(RemoteReactorDispatch dispatch)
+{
+#ifdef KVIKIO_LIBCURL_FOUND
+  KVIKIO_EXPECT(!detail::MultiReactorPool::is_instantiated(),
+                "remote_io_reactor_dispatch cannot be changed after the MULTI_POLL reactor pool "
+                "has already started",
+                std::runtime_error);
+#endif
+  instance()->_remote_io_reactor_dispatch = dispatch;
+}
+
 std::size_t defaults::remote_io_max_concurrent_requests()
 {
   return instance()->_remote_io_max_concurrent_requests;
 }
+
+void defaults::set_remote_io_max_concurrent_requests(std::size_t max_requests)
+{
+#ifdef KVIKIO_LIBCURL_FOUND
+  KVIKIO_EXPECT(!detail::MultiReactorPool::is_instantiated(),
+                "remote_io_max_concurrent_requests cannot be changed after the MULTI_POLL "
+                "reactor pool has already started",
+                std::runtime_error);
+#endif
+  instance()->_remote_io_max_concurrent_requests = max_requests;
+}
 }  // namespace kvikio

cpp/src/detail/multi_poll_reactor.cpp

@@ -518,6 +518,15 @@ void MultiPollReactor::fail_all_pending(std::exception_ptr eptr)
   _in_flight.clear();
 }
 
+namespace {
+std::atomic<bool> _pool_instantiated{false};
+}  // namespace
+
+bool MultiReactorPool::is_instantiated() noexcept
+{
+  return _pool_instantiated.load(std::memory_order_acquire);
+}
+
 MultiReactorPool::MultiReactorPool() : _dispatch{defaults::remote_io_reactor_dispatch()}
 {
   // Force LibCurl global init before any reactor opens a multi handle.
@@ -534,6 +543,8 @@ MultiReactorPool::MultiReactorPool() : _dispatch{defaults::remote_io_reactor_dis
   for (unsigned int i = 0; i < n; ++i) {
     _reactors.emplace_back(std::make_unique<MultiPollReactor>(this, per_reactor_max));
   }
+
+  _pool_instantiated.store(true, std::memory_order_release);
 }
 
 MultiReactorPool::~MultiReactorPool() noexcept

docs/source/runtime_settings.rst

@@ -79,13 +79,15 @@ KvikIO supports two backends for remote (HTTP/S3/WebHDFS) reads, selected via th
 
 This setting can be queried (:py:func:`kvikio.defaults.get`) and modified (:py:func:`kvikio.defaults.set`) at runtime using the property name ``remote_io_backend``. :py:func:`kvikio.RemoteFile.pread` reads the setting on every call, so a change applies to subsequent reads while reads already in flight finish on the backend they started on.
 
-The ``MULTI_POLL`` backend honors three additional settings, described in the sections below: ``KVIKIO_REMOTE_IO_NUM_REACTORS``, ``KVIKIO_REMOTE_IO_REACTOR_DISPATCH``, and ``KVIKIO_REMOTE_IO_MAX_CONCURRENT_REQUESTS``. They have no effect under ``EASY_THREADPOOL``. These settings are captured once when the reactor pool is first used, so switching to ``MULTI_POLL`` at runtime picks up the values they had at process startup.
+The ``MULTI_POLL`` backend honors three additional settings, described in the sections below: ``KVIKIO_REMOTE_IO_NUM_REACTORS``, ``KVIKIO_REMOTE_IO_REACTOR_DISPATCH``, and ``KVIKIO_REMOTE_IO_MAX_CONCURRENT_REQUESTS``. They have no effect under ``EASY_THREADPOOL``. These settings are captured once when the reactor pool is first used (i.e. the first ``MULTI_POLL`` remote I/O), and become immutable for the rest of the process lifetime; attempting to change any of them afterwards raises an exception.
 
 Remote I/O Reactor Count ``KVIKIO_REMOTE_IO_NUM_REACTORS``
 ----------------------------------------------------------
 
 Number of reactor threads used by the ``MULTI_POLL`` backend. The default value is ``1``. Each reactor owns one ``CURLM*`` handle and serializes its libcurl-multi calls. Increase beyond ``1`` to spread the in-callback ``memcpy`` cost across cores when one reactor's CPU is the bottleneck. This setting has no effect under ``EASY_THREADPOOL``.
 
+This setting can be queried (:py:func:`kvikio.defaults.get`) and modified (:py:func:`kvikio.defaults.set`) at runtime using the property name ``remote_io_num_reactors``, as long as the ``MULTI_POLL`` reactor pool has not already started.
+
 Remote I/O Reactor Dispatch ``KVIKIO_REMOTE_IO_REACTOR_DISPATCH``
 -----------------------------------------------------------------
 
@@ -94,6 +96,8 @@ Controls how the sub-ranges of a single :py:func:`kvikio.RemoteFile.pread` are d
   * ``PER_CHUNK`` (default): Sub-ranges are routed to reactors round-robin, independently of which :py:func:`kvikio.RemoteFile.pread` they belong to. This maximizes load balance across reactors. Trade-off: two sub-ranges of the same file may land on different reactors, each with its own libcurl connection cache, so they may not share an established TCP/TLS connection.
   * ``PER_PREAD``: All sub-ranges of a single :py:func:`kvikio.RemoteFile.pread` are submitted to the same reactor (the reactor is itself chosen round-robin per :py:func:`kvikio.RemoteFile.pread` call). The sub-ranges then share that reactor's libcurl connection cache, allowing an established TCP/TLS connection to be reused. Best for HTTPS, where the TLS handshake cost is non-trivial.
 
+This setting can be queried (:py:func:`kvikio.defaults.get`) and modified (:py:func:`kvikio.defaults.set`) at runtime using the property name ``remote_io_reactor_dispatch``, as long as the ``MULTI_POLL`` reactor pool has not already started.
+
 Remote I/O Concurrency Cap ``KVIKIO_REMOTE_IO_MAX_CONCURRENT_REQUESTS``
 -----------------------------------------------------------------------
 
@@ -103,6 +107,8 @@ The global budget is divided into an equal private share per reactor (``KVIKIO_R
 
 The even split assumes sub-ranges are spread across reactors, which holds under ``PER_CHUNK``. Under ``PER_PREAD`` all sub-ranges of one large :py:func:`kvikio.RemoteFile.pread` land on a single reactor, so that read is effectively limited to one reactor's share while the others stay idle.
 
+This setting can be queried (:py:func:`kvikio.defaults.get`) and modified (:py:func:`kvikio.defaults.set`) at runtime using the property name ``remote_io_max_concurrent_requests``, as long as the ``MULTI_POLL`` reactor pool has not already started.
+
 CA bundle file and CA directory ``CURL_CA_BUNDLE``, ``SSL_CERT_FILE``, ``SSL_CERT_DIR``
 ---------------------------------------------------------------------------------------
 

python/kvikio/kvikio/__init__.py

@@ -12,7 +12,11 @@
     del libkvikio
 
 
-from kvikio._lib.defaults import CompatMode, RemoteIOBackend  # noqa: F401
+from kvikio._lib.defaults import (  # noqa: F401
+    CompatMode,
+    RemoteIOBackend,
+    RemoteReactorDispatch,
+)
 from kvikio._version import __git_commit__, __version__
 from kvikio.buffer import bounce_buffer_free, memory_deregister, memory_register
 from kvikio.cufile import (

python/kvikio/kvikio/_lib/defaults.pyx

@@ -50,12 +50,27 @@ cdef extern from "<kvikio/defaults.hpp>" namespace "kvikio" nogil:
         "kvikio::defaults::remote_io_backend"() except +
     void cpp_set_remote_io_backend \
         "kvikio::defaults::set_remote_io_backend"(RemoteIOBackend backend) except +
+    unsigned int cpp_remote_io_num_reactors \
+        "kvikio::defaults::remote_io_num_reactors"() except +
+    void cpp_set_remote_io_num_reactors \
+        "kvikio::defaults::set_remote_io_num_reactors"(unsigned int num_reactors) except +
+    RemoteReactorDispatch cpp_remote_io_reactor_dispatch \
+        "kvikio::defaults::remote_io_reactor_dispatch"() except +
+    void cpp_set_remote_io_reactor_dispatch \
+        "kvikio::defaults::set_remote_io_reactor_dispatch"(RemoteReactorDispatch dispatch) except +
+    size_t cpp_remote_io_max_concurrent_requests \
+        "kvikio::defaults::remote_io_max_concurrent_requests"() except +
+    void cpp_set_remote_io_max_concurrent_requests \
+        "kvikio::defaults::set_remote_io_max_concurrent_requests"(size_t max_requests) except +
 
 
 cdef extern from "<kvikio/remote_handle.hpp>" namespace "kvikio" nogil:
     cpdef enum class RemoteIOBackend(uint8_t):
         EASY_THREADPOOL
         MULTI_POLL
+    cpdef enum class RemoteReactorDispatch(uint8_t):
+        PER_CHUNK
+        PER_PREAD
 
 
 def is_compat_mode_preferred() -> bool:
@@ -201,3 +216,41 @@ def remote_io_backend() -> RemoteIOBackend:
 def set_remote_io_backend(backend: RemoteIOBackend) -> None:
     with nogil:
         cpp_set_remote_io_backend(backend)
+
+
+def remote_io_num_reactors() -> int:
+    cdef unsigned int result
+    with nogil:
+        result = cpp_remote_io_num_reactors()
+    return result
+
+
+def set_remote_io_num_reactors(num_reactors: int) -> None:
+    cdef unsigned int cpp_num_reactors = num_reactors
+    with nogil:
+        cpp_set_remote_io_num_reactors(cpp_num_reactors)
+
+
+def remote_io_reactor_dispatch() -> RemoteReactorDispatch:
+    cdef RemoteReactorDispatch result
+    with nogil:
+        result = cpp_remote_io_reactor_dispatch()
+    return result
+
+
+def set_remote_io_reactor_dispatch(dispatch: RemoteReactorDispatch) -> None:
+    with nogil:
+        cpp_set_remote_io_reactor_dispatch(dispatch)
+
+
+def remote_io_max_concurrent_requests() -> int:
+    cdef size_t result
+    with nogil:
+        result = cpp_remote_io_max_concurrent_requests()
+    return result
+
+
+def set_remote_io_max_concurrent_requests(max_requests: int) -> None:
+    cdef size_t cpp_max_requests = max_requests
+    with nogil:
+        cpp_set_remote_io_max_concurrent_requests(cpp_max_requests)

python/kvikio/kvikio/defaults.py

@@ -2,11 +2,24 @@
 # SPDX-License-Identifier: Apache-2.0
 
 
-from typing import Any, overload
+from typing import Any, Literal, overload
 
 import kvikio._lib.defaults
 from kvikio.utils import call_once
 
+# Once the MULTI_POLL reactor pool has started (i.e. the first remote I/O has been
+# issued), these properties are fixed for the remaining process lifetime: there is no
+# value they can be reverted to, since even a plain (non-context-manager) `set()` call
+# would fail at that point too. They are therefore not among the properties
+# `ConfigContextManager` knows how to get/set, and cannot be used with `with`.
+_PROCESS_LIFETIME_PROPERTIES = frozenset(
+    {
+        "remote_io_num_reactors",
+        "remote_io_reactor_dispatch",
+        "remote_io_max_concurrent_requests",
+    }
+)
+
 
 class ConfigContextManager:
     """Context manager allowing the KvikIO configurations to be set upon entering a
@@ -74,11 +87,23 @@ def _property_getter_and_setter(self) -> tuple[dict[str, Any], dict[str, Any]]:
 def set(config: dict[str, Any], /) -> ConfigContextManager: ...
 
 
+@overload
+def set(
+    key: Literal[
+        "remote_io_num_reactors",
+        "remote_io_reactor_dispatch",
+        "remote_io_max_concurrent_requests",
+    ],
+    value: Any,
+    /,
+) -> None: ...
+
+
 @overload
 def set(key: str, value: Any, /) -> ConfigContextManager: ...
 
 
-def set(*config) -> ConfigContextManager:
+def set(*config) -> ConfigContextManager | None:
     """Set KvikIO configurations.
 
     Examples:
@@ -130,11 +155,22 @@ def set(*config) -> ConfigContextManager:
         - ``"auto_direct_io_write"``
         - ``"remote_io_backend"``
 
+        The following are fixed for the remaining process lifetime once the first
+        remote I/O has been issued, so they must be set with
+        ``kvikio.defaults.set(key, value)`` (not the dict form) and cannot be used
+        with a `with` block:
+
+        - ``"remote_io_num_reactors"``
+        - ``"remote_io_reactor_dispatch"``
+        - ``"remote_io_max_concurrent_requests"``
+
     Returns
     -------
-    ConfigContextManager
+    ConfigContextManager or None
        A context manager. If used in a `with` statement, the configuration will revert
-       to its old value upon leaving the block.
+       to its old value upon leaving the block. `None` when setting one of the
+       process-lifetime configurations listed above, which cannot be reverted and thus
+       must be set with the two-argument form, not the dict form.
     """
 
     err_msg = (
@@ -145,11 +181,24 @@ def set(*config) -> ConfigContextManager:
     if len(config) == 1:
         if not isinstance(config[0], dict):
             raise ValueError(err_msg)
-        return ConfigContextManager(config[0])
+        config_dict = config[0]
+        process_lifetime_keys = _PROCESS_LIFETIME_PROPERTIES.intersection(config_dict)
+        if process_lifetime_keys:
+            raise ValueError(
+                f"{sorted(process_lifetime_keys)} are fixed for the process's "
+                "remaining lifetime once the first remote I/O has been issued, so "
+                "they cannot be reverted and must be set individually with "
+                "kvikio.defaults.set(key, value), not as part of a dict."
+            )
+        return ConfigContextManager(config_dict)
     elif len(config) == 2:
         if not isinstance(config[0], str):
             raise ValueError(err_msg)
-        return ConfigContextManager({config[0]: config[1]})
+        key, value = config
+        if key in _PROCESS_LIFETIME_PROPERTIES:
+            getattr(kvikio._lib.defaults, "set_" + key)(value)
+            return None
+        return ConfigContextManager({key: value})
     else:
         raise ValueError(err_msg)
 
@@ -175,12 +224,17 @@ def get(config_name: str) -> Any:
         - ``"auto_direct_io_read"``
         - ``"auto_direct_io_write"``
         - ``"remote_io_backend"``
+        - ``"remote_io_num_reactors"``
+        - ``"remote_io_reactor_dispatch"``
+        - ``"remote_io_max_concurrent_requests"``
 
     Returns
     -------
     Any
         The value of the configuration.
     """
+    if config_name in _PROCESS_LIFETIME_PROPERTIES:
+        return getattr(kvikio._lib.defaults, config_name)()
     context_manager = ConfigContextManager({})
     return context_manager._get_property(config_name)