From 64d28f1bb22ea4126384e6699c6d94ed69ef4973 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 19 Oct 2026 02:27:23 +0000 Subject: [PATCH 1/2] Deprecate the cudf-polars kvikio_reactor_dispatch option --- python/cudf_polars/cudf_polars/engine/core.py | 5 +- python/cudf_polars/cudf_polars/engine/dask.py | 11 ---- .../cudf_polars/cudf_polars/engine/options.py | 16 ++--- python/cudf_polars/cudf_polars/engine/ray.py | 10 ---- python/cudf_polars/cudf_polars/engine/spmd.py | 2 - .../cudf_polars/cudf_polars/utils/config.py | 58 ++++++------------- python/cudf_polars/tests/test_config.py | 39 ++++++++----- 7 files changed, 51 insertions(+), 90 deletions(-) diff --git a/python/cudf_polars/cudf_polars/engine/core.py b/python/cudf_polars/cudf_polars/engine/core.py index a23c9f785703..5be6d81a4c97 100644 --- a/python/cudf_polars/cudf_polars/engine/core.py +++ b/python/cudf_polars/cudf_polars/engine/core.py @@ -318,9 +318,8 @@ class StreamingEngine(pl.GPUEngine): environment variable to control the thread count. Under ``MULTI_POLL``, cudf-polars does not resolve a thread-pool size at all (kvikio itself may still honor ``KVIKIO_NTHREADS`` via its own deferred default); remote I/O - concurrency is instead controlled by the ``kvikio_reactor_count``, - ``kvikio_reactor_dispatch``, and ``kvikio_request_ceiling`` executor - options. + concurrency is instead controlled by the ``kvikio_reactor_count`` and + ``kvikio_request_ceiling`` executor options. Parameters ---------- diff --git a/python/cudf_polars/cudf_polars/engine/dask.py b/python/cudf_polars/cudf_polars/engine/dask.py index f76336c7501d..be56cff683cb 100644 --- a/python/cudf_polars/cudf_polars/engine/dask.py +++ b/python/cudf_polars/cudf_polars/engine/dask.py @@ -343,7 +343,6 @@ def _setup_worker( kvikio_task_size: int, kvikio_bounce_buffer_bytes: int, kvikio_reactor_count: int, - kvikio_reactor_dispatch: kvikio.RemoteReactorDispatch, kvikio_request_ceiling: int, quent_collector_address: str | None, dask_worker: distributed.Worker | None = None, @@ -392,8 +391,6 @@ def _setup_worker( Size, in bytes, of the kvikio bounce buffer to configure on this worker process. kvikio_reactor_count Number of ``MULTI_POLL`` reactor threads to configure on this worker process. - kvikio_reactor_dispatch - ``MULTI_POLL`` reactor dispatch policy to configure on this worker process. kvikio_request_ceiling ``MULTI_POLL`` concurrent-request ceiling to configure on this worker process. quent_collector_address @@ -413,7 +410,6 @@ def _setup_worker( task_size=kvikio_task_size, bounce_buffer_bytes=kvikio_bounce_buffer_bytes, reactor_count=kvikio_reactor_count, - reactor_dispatch=kvikio_reactor_dispatch, request_ceiling=kvikio_request_ceiling, ) memory_resource_config = ( @@ -442,7 +438,6 @@ def _setup_worker( task_size=kvikio_task_size, bounce_buffer_bytes=kvikio_bounce_buffer_bytes, reactor_count=kvikio_reactor_count, - reactor_dispatch=kvikio_reactor_dispatch, request_ceiling=kvikio_request_ceiling, ) @@ -581,7 +576,6 @@ def _reset_worker( kvikio_task_size: int, kvikio_bounce_buffer_bytes: int, kvikio_reactor_count: int, - kvikio_reactor_dispatch: kvikio.RemoteReactorDispatch, kvikio_request_ceiling: int, dask_worker: distributed.Worker | None = None, ) -> None: @@ -610,8 +604,6 @@ def _reset_worker( Size, in bytes, of the kvikio bounce buffer to configure on this worker process. kvikio_reactor_count Number of ``MULTI_POLL`` reactor threads to configure on this worker process. - kvikio_reactor_dispatch - ``MULTI_POLL`` reactor dispatch policy to configure on this worker process. kvikio_request_ceiling ``MULTI_POLL`` concurrent-request ceiling to configure on this worker process. dask_worker @@ -624,7 +616,6 @@ def _reset_worker( task_size=kvikio_task_size, bounce_buffer_bytes=kvikio_bounce_buffer_bytes, reactor_count=kvikio_reactor_count, - reactor_dispatch=kvikio_reactor_dispatch, request_ceiling=kvikio_request_ceiling, ) attr = f"_cudf_polars_mp_context_{uid}" @@ -1153,7 +1144,6 @@ def __init__( "kvikio_bounce_buffer_bytes" ], kvikio_reactor_count=executor_options["kvikio_reactor_count"], - kvikio_reactor_dispatch=executor_options["kvikio_reactor_dispatch"], kvikio_request_ceiling=executor_options["kvikio_request_ceiling"], ) if quent_context is not None: @@ -1237,7 +1227,6 @@ def _reset( "kvikio_bounce_buffer_bytes" ], kvikio_reactor_count=executor_options["kvikio_reactor_count"], - kvikio_reactor_dispatch=executor_options["kvikio_reactor_dispatch"], kvikio_request_ceiling=executor_options["kvikio_request_ceiling"], ), rapidsmpf_options_as_bytes, diff --git a/python/cudf_polars/cudf_polars/engine/options.py b/python/cudf_polars/cudf_polars/engine/options.py index 4b948800cd5d..32951f6d8257 100644 --- a/python/cudf_polars/cudf_polars/engine/options.py +++ b/python/cudf_polars/cudf_polars/engine/options.py @@ -48,10 +48,6 @@ def _parse_remote_io_backend(value: str) -> kvikio.RemoteIOBackend: return kvikio.RemoteIOBackend[value.upper()] -def _parse_reactor_dispatch(value: str) -> kvikio.RemoteReactorDispatch: - return kvikio.RemoteReactorDispatch[value.upper()] - - def _opt( category: str, env_var: str | None = None, @@ -263,10 +259,10 @@ class StreamingOptions: Default: ``24``. Category: executor. kvikio_reactor_dispatch - How sub-ranges of one read are distributed across reactor threads under - the ``MULTI_POLL`` remote I/O backend. + **Deprecated**: Has no effect. KvikIO's ``MULTI_POLL`` reactors always + take sub-ranges from one shared queue. Setting it emits a + ``FutureWarning``. Env: ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH``. - Default: ``kvikio.RemoteReactorDispatch.PER_CHUNK``. Category: executor. kvikio_request_ceiling Maximum number of concurrent in-flight requests across all reactor @@ -432,10 +428,8 @@ class StreamingOptions: kvikio_reactor_count: int | Unspecified = _opt( "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_COUNT", int ) - kvikio_reactor_dispatch: kvikio.RemoteReactorDispatch | Unspecified = _opt( - "executor", - "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH", - _parse_reactor_dispatch, + kvikio_reactor_dispatch: Any = _opt( + "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH" ) kvikio_request_ceiling: int | Unspecified = _opt( "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REQUEST_CEILING", int diff --git a/python/cudf_polars/cudf_polars/engine/ray.py b/python/cudf_polars/cudf_polars/engine/ray.py index e776570a0591..ae1ef8e1afaf 100644 --- a/python/cudf_polars/cudf_polars/engine/ray.py +++ b/python/cudf_polars/cudf_polars/engine/ray.py @@ -262,7 +262,6 @@ def __init__( kvikio_task_size: int, kvikio_bounce_buffer_bytes: int, kvikio_reactor_count: int, - kvikio_reactor_dispatch: kvikio.RemoteReactorDispatch, kvikio_request_ceiling: int, hardware_binding: HardwareBindingPolicy, memory_resource_config: MemoryResourceConfig | None, @@ -277,7 +276,6 @@ def __init__( task_size=kvikio_task_size, bounce_buffer_bytes=kvikio_bounce_buffer_bytes, reactor_count=kvikio_reactor_count, - reactor_dispatch=kvikio_reactor_dispatch, request_ceiling=kvikio_request_ceiling, ) memory_resource_config = ( @@ -406,7 +404,6 @@ def reset( kvikio_task_size: int, kvikio_bounce_buffer_bytes: int, kvikio_reactor_count: int, - kvikio_reactor_dispatch: kvikio.RemoteReactorDispatch, kvikio_request_ceiling: int, ) -> None: """ @@ -432,8 +429,6 @@ def reset( Size, in bytes, of the kvikio bounce buffer to configure on this worker process. kvikio_reactor_count Number of ``MULTI_POLL`` reactor threads to configure on this worker process. - kvikio_reactor_dispatch - ``MULTI_POLL`` reactor dispatch policy to configure on this worker process. kvikio_request_ceiling ``MULTI_POLL`` concurrent-request ceiling to configure on this worker process. """ @@ -445,7 +440,6 @@ def reset( task_size=kvikio_task_size, bounce_buffer_bytes=kvikio_bounce_buffer_bytes, reactor_count=kvikio_reactor_count, - reactor_dispatch=kvikio_reactor_dispatch, request_ceiling=kvikio_request_ceiling, ) assert self._comm is not None @@ -966,9 +960,6 @@ def __init__( "kvikio_bounce_buffer_bytes" ], kvikio_reactor_count=executor_options["kvikio_reactor_count"], - kvikio_reactor_dispatch=executor_options[ - "kvikio_reactor_dispatch" - ], kvikio_request_ceiling=executor_options[ "kvikio_request_ceiling" ], @@ -1077,7 +1068,6 @@ def _reset( "kvikio_bounce_buffer_bytes" ], kvikio_reactor_count=executor_options["kvikio_reactor_count"], - kvikio_reactor_dispatch=executor_options["kvikio_reactor_dispatch"], kvikio_request_ceiling=executor_options["kvikio_request_ceiling"], ) for rank in self._rank_actors diff --git a/python/cudf_polars/cudf_polars/engine/spmd.py b/python/cudf_polars/cudf_polars/engine/spmd.py index 14dfe5de6d8f..9f7960afa9be 100644 --- a/python/cudf_polars/cudf_polars/engine/spmd.py +++ b/python/cudf_polars/cudf_polars/engine/spmd.py @@ -594,7 +594,6 @@ def __init__( task_size=executor_options["kvikio_task_size"], bounce_buffer_bytes=executor_options["kvikio_bounce_buffer_bytes"], reactor_count=executor_options["kvikio_reactor_count"], - reactor_dispatch=executor_options["kvikio_reactor_dispatch"], request_ceiling=executor_options["kvikio_request_ceiling"], ) @@ -1003,7 +1002,6 @@ def _reset( task_size=executor_options["kvikio_task_size"], bounce_buffer_bytes=executor_options["kvikio_bounce_buffer_bytes"], reactor_count=executor_options["kvikio_reactor_count"], - reactor_dispatch=executor_options["kvikio_reactor_dispatch"], request_ceiling=executor_options["kvikio_request_ceiling"], ) engine_options = engine_options or {} diff --git a/python/cudf_polars/cudf_polars/utils/config.py b/python/cudf_polars/cudf_polars/utils/config.py index a0a06ebb87cb..117d927cd7ff 100644 --- a/python/cudf_polars/cudf_polars/utils/config.py +++ b/python/cudf_polars/cudf_polars/utils/config.py @@ -27,6 +27,7 @@ import importlib.util import json import os +import warnings from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, TypeVar @@ -372,22 +373,6 @@ def resolve_kvikio_reactor_count(executor_options: dict[str, Any]) -> int: ) -def resolve_kvikio_reactor_dispatch( - executor_options: dict[str, Any], -) -> kvikio.RemoteReactorDispatch: - """Resolve the MULTI_POLL reactor dispatch policy, with env var fallback.""" - value = executor_options.get( - "kvikio_reactor_dispatch", - os.environ.get( - "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH", - os.environ.get("KVIKIO_REMOTE_IO_REACTOR_DISPATCH", "PER_CHUNK"), - ), - ) - if isinstance(value, kvikio.RemoteReactorDispatch): - return value - return kvikio.RemoteReactorDispatch[str(value).upper()] - - def resolve_kvikio_request_ceiling(executor_options: dict[str, Any]) -> int: """Resolve the MULTI_POLL concurrent-request ceiling, with env var fallback.""" return int( @@ -439,10 +424,6 @@ def resolve_kvikio_executor_options(executor_options: dict[str, Any]) -> dict[st executor_options.setdefault( "kvikio_reactor_count", resolve_kvikio_reactor_count(executor_options) ) - executor_options.setdefault( - "kvikio_reactor_dispatch", - resolve_kvikio_reactor_dispatch(executor_options), - ) executor_options.setdefault( "kvikio_request_ceiling", resolve_kvikio_request_ceiling(executor_options), @@ -459,7 +440,6 @@ def resolve_kvikio_executor_options(executor_options: dict[str, Any]) -> dict[st "task_size", "bounce_buffer_size", "remote_io_num_reactors", - "remote_io_reactor_dispatch", "remote_io_max_concurrent_requests", "num_threads", ) @@ -472,7 +452,6 @@ def resolve_kvikio_executor_options(executor_options: dict[str, Any]) -> dict[st KVIKIO_REACTOR_POOL_PROPERTIES = frozenset( { "remote_io_num_reactors", - "remote_io_reactor_dispatch", "remote_io_max_concurrent_requests", } ) @@ -485,7 +464,6 @@ def configure_kvikio( task_size: int | None = None, bounce_buffer_bytes: int = 16 * 1024 * 1024, reactor_count: int = 24, - reactor_dispatch: kvikio.RemoteReactorDispatch = kvikio.RemoteReactorDispatch.PER_CHUNK, request_ceiling: int = 256, ) -> None: """ @@ -520,7 +498,6 @@ def configure_kvikio( # left unset in that case. reactor_settings = { "remote_io_num_reactors": reactor_count, - "remote_io_reactor_dispatch": reactor_dispatch, "remote_io_max_concurrent_requests": request_ceiling, } else: @@ -1222,14 +1199,11 @@ class StreamingExecutor: - the ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_COUNT`` environment variable - the ``KVIKIO_REMOTE_IO_NUM_REACTORS`` environment variable (lower precedence) kvikio_reactor_dispatch - How sub-ranges of one read are distributed across reactor threads under - the ``MULTI_POLL`` remote I/O backend. Ignored when - ``kvikio_remote_io_backend`` is not ``MULTI_POLL``. Defaults to - ``kvikio.RemoteReactorDispatch.PER_CHUNK``. This can be set via - - - ``executor_options`` passed to ``polars.GPUEngine`` - - the ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH`` environment variable - - the ``KVIKIO_REMOTE_IO_REACTOR_DISPATCH`` environment variable (lower precedence) + **Deprecated**: Has no effect. KvikIO no longer makes the ``MULTI_POLL`` + reactor dispatch configurable. Its reactors always take sub-ranges from + one shared queue. Setting this option, either through ``executor_options`` + or the ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH`` environment + variable, emits a ``FutureWarning``. kvikio_request_ceiling Maximum number of concurrent in-flight requests across all reactor threads under the ``MULTI_POLL`` remote I/O backend. 0 means unlimited. @@ -1345,8 +1319,11 @@ class StreamingExecutor: kvikio_reactor_count: int = dataclasses.field( default_factory=lambda: resolve_kvikio_reactor_count({}) ) - kvikio_reactor_dispatch: kvikio.RemoteReactorDispatch = dataclasses.field( - default_factory=lambda: resolve_kvikio_reactor_dispatch({}) + # Deprecated and ignored. `__post_init__` warns if a user passes it. + kvikio_reactor_dispatch: Any = dataclasses.field( + default_factory=_make_default_factory( + f"{_env_prefix}__KVIKIO_REACTOR_DISPATCH", str, default=None + ) ) kvikio_request_ceiling: int = dataclasses.field( default_factory=lambda: resolve_kvikio_request_ceiling({}) @@ -1483,12 +1460,15 @@ def __post_init__(self) -> None: # noqa: D105 raise TypeError("kvikio_reactor_count must be an int") if self.kvikio_reactor_count <= 0: raise ValueError("kvikio_reactor_count must be positive") - if not isinstance(self.kvikio_reactor_dispatch, kvikio.RemoteReactorDispatch): - object.__setattr__( - self, - "kvikio_reactor_dispatch", - kvikio.RemoteReactorDispatch[str(self.kvikio_reactor_dispatch).upper()], + if self.kvikio_reactor_dispatch is not None: + warnings.warn( + "The 'kvikio_reactor_dispatch' option is deprecated and has no " + "effect. KvikIO's MULTI_POLL reactors always take sub-ranges from " + "one shared queue.", + FutureWarning, + stacklevel=2, ) + object.__setattr__(self, "kvikio_reactor_dispatch", None) if not isinstance(self.kvikio_request_ceiling, int): raise TypeError("kvikio_request_ceiling must be an int") if self.kvikio_request_ceiling < 0: diff --git a/python/cudf_polars/tests/test_config.py b/python/cudf_polars/tests/test_config.py index 458b4c35941a..a859edd57bae 100644 --- a/python/cudf_polars/tests/test_config.py +++ b/python/cudf_polars/tests/test_config.py @@ -45,7 +45,6 @@ configure_kvikio, resolve_kvikio_bounce_buffer_bytes, resolve_kvikio_nthreads, - resolve_kvikio_reactor_dispatch, resolve_kvikio_remote_io_backend, resolve_kvikio_task_size, resolve_quent_context, @@ -385,12 +384,6 @@ def test_kvikio_resolvers_accept_enum_values( ) assert resolve_kvikio_nthreads({}) == 32 assert resolve_kvikio_task_size({}) == 64 * 1024 * 1024 - assert ( - resolve_kvikio_reactor_dispatch( - {"kvikio_reactor_dispatch": kvikio.RemoteReactorDispatch.PER_CHUNK} - ) - == kvikio.RemoteReactorDispatch.PER_CHUNK - ) @pytest.mark.parametrize( @@ -427,10 +420,33 @@ def test_streaming_executor_normalizes_kvikio_enum_strings() -> None: executor = StreamingExecutor( cluster=Cluster.DEFAULT_SINGLETON, kvikio_remote_io_backend="EASY_THREADPOOL", - kvikio_reactor_dispatch="PER_CHUNK", ) assert executor.kvikio_remote_io_backend == kvikio.RemoteIOBackend.EASY_THREADPOOL - assert executor.kvikio_reactor_dispatch == kvikio.RemoteReactorDispatch.PER_CHUNK + + +def test_kvikio_reactor_dispatch_default_no_warning( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The "error" warning filter turns any warning on the default path into a failure. + monkeypatch.delenv("CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH", raising=False) + executor = StreamingExecutor(cluster=Cluster.DEFAULT_SINGLETON) + assert executor.kvikio_reactor_dispatch is None + + +@pytest.mark.parametrize("from_env", [False, True]) +def test_kvikio_reactor_dispatch_deprecated( + monkeypatch: pytest.MonkeyPatch, *, from_env: bool +) -> None: + kwargs: dict[str, Any] = {} + if from_env: + monkeypatch.setenv( + "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH", "PER_PREAD" + ) + else: + kwargs["kvikio_reactor_dispatch"] = "PER_PREAD" + with pytest.warns(FutureWarning, match="kvikio_reactor_dispatch"): + executor = StreamingExecutor(cluster=Cluster.DEFAULT_SINGLETON, **kwargs) + assert executor.kvikio_reactor_dispatch is None def test_executor_from_env(monkeypatch: pytest.MonkeyPatch) -> None: @@ -1306,7 +1322,6 @@ def record_set(*args): configure_kvikio(42) reactor_settings = { "remote_io_num_reactors": 24, - "remote_io_reactor_dispatch": kvikio.RemoteReactorDispatch.PER_CHUNK, "remote_io_max_concurrent_requests": 256, } assert all( @@ -1322,10 +1337,6 @@ def record_set(*args): assert (key, value) in calls assert kvikio.defaults.get("remote_io_backend") == kvikio.RemoteIOBackend.MULTI_POLL assert kvikio.defaults.get("remote_io_num_reactors") == 24 - assert ( - kvikio.defaults.get("remote_io_reactor_dispatch") - == kvikio.RemoteReactorDispatch.PER_CHUNK - ) assert kvikio.defaults.get("remote_io_max_concurrent_requests") == 256 assert kvikio.defaults.get("bounce_buffer_size") == 16 * 1024 * 1024 assert kvikio.defaults.get("task_size") == 16 * 1024 * 1024 From c9e42529de33a6423d8f63f80b34a76321ba2c6e Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Thu, 22 Oct 2026 14:32:39 +0000 Subject: [PATCH 2/2] Remove the kvikio_reactor_dispatch option --- .../cudf_polars/cudf_polars/engine/options.py | 9 ------- .../cudf_polars/cudf_polars/utils/config.py | 22 ---------------- python/cudf_polars/tests/test_config.py | 25 ------------------- 3 files changed, 56 deletions(-) diff --git a/python/cudf_polars/cudf_polars/engine/options.py b/python/cudf_polars/cudf_polars/engine/options.py index 5a6cbd4bb74d..49523d36c7dd 100644 --- a/python/cudf_polars/cudf_polars/engine/options.py +++ b/python/cudf_polars/cudf_polars/engine/options.py @@ -263,12 +263,6 @@ class StreamingOptions: Env: ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_COUNT``. Default: ``24``. Category: executor. - kvikio_reactor_dispatch - **Deprecated**: Has no effect. KvikIO's ``MULTI_POLL`` reactors always - take sub-ranges from one shared queue. Setting it emits a - ``FutureWarning``. - Env: ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH``. - Category: executor. kvikio_request_ceiling Maximum number of concurrent in-flight requests across all reactor threads under the ``MULTI_POLL`` remote I/O backend. 0 means unlimited. @@ -433,9 +427,6 @@ class StreamingOptions: kvikio_reactor_count: int | Unspecified = _opt( "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_COUNT", int ) - kvikio_reactor_dispatch: Any = _opt( - "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH" - ) kvikio_request_ceiling: int | Unspecified = _opt( "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REQUEST_CEILING", int ) diff --git a/python/cudf_polars/cudf_polars/utils/config.py b/python/cudf_polars/cudf_polars/utils/config.py index 117d927cd7ff..49bf59d77f2f 100644 --- a/python/cudf_polars/cudf_polars/utils/config.py +++ b/python/cudf_polars/cudf_polars/utils/config.py @@ -27,7 +27,6 @@ import importlib.util import json import os -import warnings from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, TypeVar @@ -1198,12 +1197,6 @@ class StreamingExecutor: - ``executor_options`` passed to ``polars.GPUEngine`` - the ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_COUNT`` environment variable - the ``KVIKIO_REMOTE_IO_NUM_REACTORS`` environment variable (lower precedence) - kvikio_reactor_dispatch - **Deprecated**: Has no effect. KvikIO no longer makes the ``MULTI_POLL`` - reactor dispatch configurable. Its reactors always take sub-ranges from - one shared queue. Setting this option, either through ``executor_options`` - or the ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH`` environment - variable, emits a ``FutureWarning``. kvikio_request_ceiling Maximum number of concurrent in-flight requests across all reactor threads under the ``MULTI_POLL`` remote I/O backend. 0 means unlimited. @@ -1319,12 +1312,6 @@ class StreamingExecutor: kvikio_reactor_count: int = dataclasses.field( default_factory=lambda: resolve_kvikio_reactor_count({}) ) - # Deprecated and ignored. `__post_init__` warns if a user passes it. - kvikio_reactor_dispatch: Any = dataclasses.field( - default_factory=_make_default_factory( - f"{_env_prefix}__KVIKIO_REACTOR_DISPATCH", str, default=None - ) - ) kvikio_request_ceiling: int = dataclasses.field( default_factory=lambda: resolve_kvikio_request_ceiling({}) ) @@ -1460,15 +1447,6 @@ def __post_init__(self) -> None: # noqa: D105 raise TypeError("kvikio_reactor_count must be an int") if self.kvikio_reactor_count <= 0: raise ValueError("kvikio_reactor_count must be positive") - if self.kvikio_reactor_dispatch is not None: - warnings.warn( - "The 'kvikio_reactor_dispatch' option is deprecated and has no " - "effect. KvikIO's MULTI_POLL reactors always take sub-ranges from " - "one shared queue.", - FutureWarning, - stacklevel=2, - ) - object.__setattr__(self, "kvikio_reactor_dispatch", None) if not isinstance(self.kvikio_request_ceiling, int): raise TypeError("kvikio_request_ceiling must be an int") if self.kvikio_request_ceiling < 0: diff --git a/python/cudf_polars/tests/test_config.py b/python/cudf_polars/tests/test_config.py index a859edd57bae..6c337b25ced6 100644 --- a/python/cudf_polars/tests/test_config.py +++ b/python/cudf_polars/tests/test_config.py @@ -424,31 +424,6 @@ def test_streaming_executor_normalizes_kvikio_enum_strings() -> None: assert executor.kvikio_remote_io_backend == kvikio.RemoteIOBackend.EASY_THREADPOOL -def test_kvikio_reactor_dispatch_default_no_warning( - monkeypatch: pytest.MonkeyPatch, -) -> None: - # The "error" warning filter turns any warning on the default path into a failure. - monkeypatch.delenv("CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH", raising=False) - executor = StreamingExecutor(cluster=Cluster.DEFAULT_SINGLETON) - assert executor.kvikio_reactor_dispatch is None - - -@pytest.mark.parametrize("from_env", [False, True]) -def test_kvikio_reactor_dispatch_deprecated( - monkeypatch: pytest.MonkeyPatch, *, from_env: bool -) -> None: - kwargs: dict[str, Any] = {} - if from_env: - monkeypatch.setenv( - "CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_DISPATCH", "PER_PREAD" - ) - else: - kwargs["kvikio_reactor_dispatch"] = "PER_PREAD" - with pytest.warns(FutureWarning, match="kvikio_reactor_dispatch"): - executor = StreamingExecutor(cluster=Cluster.DEFAULT_SINGLETON, **kwargs) - assert executor.kvikio_reactor_dispatch is None - - def test_executor_from_env(monkeypatch: pytest.MonkeyPatch) -> None: with monkeypatch.context() as m: m.setenv("CUDF_POLARS__EXECUTOR", "in-memory")