diff --git a/cpp/include/kvikio/defaults.hpp b/cpp/include/kvikio/defaults.hpp index 7f8cf0ee72..220be07433 100644 --- a/cpp/include/kvikio/defaults.hpp +++ b/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 diff --git a/cpp/include/kvikio/detail/multi_poll_reactor.hpp b/cpp/include/kvikio/detail/multi_poll_reactor.hpp index 0342356908..0643260cf1 100644 --- a/cpp/include/kvikio/detail/multi_poll_reactor.hpp +++ b/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; diff --git a/cpp/src/defaults.cpp b/cpp/src/defaults.cpp index 85fc4ac8c2..91854a9071 100644 --- a/cpp/src/defaults.cpp +++ b/cpp/src/defaults.cpp @@ -12,6 +12,9 @@ #include #include +#ifdef KVIKIO_LIBCURL_FOUND +#include +#endif #include #include #include @@ -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 diff --git a/cpp/src/detail/multi_poll_reactor.cpp b/cpp/src/detail/multi_poll_reactor.cpp index bf05fa859d..8860513635 100644 --- a/cpp/src/detail/multi_poll_reactor.cpp +++ b/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 _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(this, per_reactor_max)); } + + _pool_instantiated.store(true, std::memory_order_release); } MultiReactorPool::~MultiReactorPool() noexcept diff --git a/docs/source/runtime_settings.rst b/docs/source/runtime_settings.rst index 9284ec34e9..b3af32547f 100644 --- a/docs/source/runtime_settings.rst +++ b/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`` --------------------------------------------------------------------------------------- diff --git a/python/kvikio/kvikio/__init__.py b/python/kvikio/kvikio/__init__.py index b869000aaa..3bf947680c 100644 --- a/python/kvikio/kvikio/__init__.py +++ b/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 ( diff --git a/python/kvikio/kvikio/_lib/defaults.pyx b/python/kvikio/kvikio/_lib/defaults.pyx index e843bf3311..2ba1584b1e 100644 --- a/python/kvikio/kvikio/_lib/defaults.pyx +++ b/python/kvikio/kvikio/_lib/defaults.pyx @@ -50,12 +50,27 @@ cdef extern from "" 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 "" 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) diff --git a/python/kvikio/kvikio/defaults.py b/python/kvikio/kvikio/defaults.py index f3d60e790a..96346f8383 100644 --- a/python/kvikio/kvikio/defaults.py +++ b/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) diff --git a/python/kvikio/tests/test_defaults.py b/python/kvikio/tests/test_defaults.py index 6dd7c281dc..4ba8102a3e 100644 --- a/python/kvikio/tests/test_defaults.py +++ b/python/kvikio/tests/test_defaults.py @@ -184,3 +184,72 @@ def test_remote_io_backend(): kvikio.RemoteIOBackend.EASY_THREADPOOL ) assert kvikio.defaults.get("remote_io_backend") == before + + +def test_remote_io_num_reactors(): + before = kvikio.defaults.get("remote_io_num_reactors") + after = before + 1 + + try: + result = kvikio.defaults.set("remote_io_num_reactors", after) + assert result is None + assert kvikio.defaults.get("remote_io_num_reactors") == after + finally: + kvikio.defaults.set("remote_io_num_reactors", before) + assert kvikio.defaults.get("remote_io_num_reactors") == before + + with pytest.raises(TypeError, match="context manager protocol"): + with kvikio.defaults.set("remote_io_num_reactors", after): + pass + kvikio.defaults.set("remote_io_num_reactors", before) + + with pytest.raises(ValueError, match="must be set individually"): + kvikio.defaults.set( + { + "remote_io_num_reactors": after, + "task_size": kvikio.defaults.get("task_size"), + } + ) + with pytest.raises(ValueError, match="must be set individually"): + kvikio.defaults.set({"remote_io_num_reactors": after}) + + +def test_remote_io_reactor_dispatch(): + before = kvikio.defaults.get("remote_io_reactor_dispatch") + after = ( + kvikio.RemoteReactorDispatch.PER_CHUNK + if before == kvikio.RemoteReactorDispatch.PER_PREAD + else kvikio.RemoteReactorDispatch.PER_PREAD + ) + assert after != before + + try: + result = kvikio.defaults.set("remote_io_reactor_dispatch", after) + assert result is None + assert kvikio.defaults.get("remote_io_reactor_dispatch") == after + finally: + kvikio.defaults.set("remote_io_reactor_dispatch", before) + assert kvikio.defaults.get("remote_io_reactor_dispatch") == before + + with pytest.raises(TypeError, match="context manager protocol"): + with kvikio.defaults.set("remote_io_reactor_dispatch", after): + pass + kvikio.defaults.set("remote_io_reactor_dispatch", before) + + +def test_remote_io_max_concurrent_requests(): + before = kvikio.defaults.get("remote_io_max_concurrent_requests") + after = before + 1 + + try: + result = kvikio.defaults.set("remote_io_max_concurrent_requests", after) + assert result is None + assert kvikio.defaults.get("remote_io_max_concurrent_requests") == after + finally: + kvikio.defaults.set("remote_io_max_concurrent_requests", before) + assert kvikio.defaults.get("remote_io_max_concurrent_requests") == before + + with pytest.raises(TypeError, match="context manager protocol"): + with kvikio.defaults.set("remote_io_max_concurrent_requests", after): + pass + kvikio.defaults.set("remote_io_max_concurrent_requests", before)