Remove the cudf-polars kvikio_reactor_dispatch option

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
     ----------

python/cudf_polars/cudf_polars/engine/dask.py

@@ -336,7 +336,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,
     dask_worker: distributed.Worker | None = None,
 ) -> None:
@@ -379,8 +378,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.
     """
@@ -398,7 +395,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 = (
@@ -427,7 +423,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,
         )
 
@@ -560,7 +555,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:
@@ -589,8 +583,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
@@ -603,7 +595,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}"
@@ -1120,7 +1111,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"],
             )
 
@@ -1204,7 +1194,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,

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,
@@ -267,12 +263,6 @@ class StreamingOptions:
         Env: ``CUDF_POLARS__EXECUTOR__KVIKIO_REACTOR_COUNT``.
         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.
-        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
         threads under the ``MULTI_POLL`` remote I/O backend. 0 means unlimited.
@@ -437,11 +427,6 @@ 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_request_ceiling: int | Unspecified = _opt(
         "executor", "CUDF_POLARS__EXECUTOR__KVIKIO_REQUEST_CEILING", int
     )

python/cudf_polars/cudf_polars/engine/ray.py

@@ -261,7 +261,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,
@@ -274,7 +273,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 = (
@@ -403,7 +401,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:
         """
@@ -429,8 +426,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.
         """
@@ -442,7 +437,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
@@ -956,9 +950,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"
                         ],
@@ -1045,7 +1036,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

python/cudf_polars/cudf_polars/engine/spmd.py

@@ -642,7 +642,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"],
         )
 
@@ -1078,7 +1077,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 {}

python/cudf_polars/cudf_polars/utils/config.py

@@ -372,22 +372,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 +423,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 +439,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 +451,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 +463,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 +497,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:
@@ -1221,15 +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
-        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)
     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,9 +1312,6 @@ 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({})
-    )
     kvikio_request_ceiling: int = dataclasses.field(
         default_factory=lambda: resolve_kvikio_request_ceiling({})
     )
@@ -1483,12 +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 not isinstance(self.kvikio_reactor_dispatch, kvikio.RemoteReactorDispatch):
-            object.__setattr__(
-                self,
-                "kvikio_reactor_dispatch",
-                kvikio.RemoteReactorDispatch[str(self.kvikio_reactor_dispatch).upper()],
-            )
         if not isinstance(self.kvikio_request_ceiling, int):
             raise TypeError("kvikio_request_ceiling must be an int")
         if self.kvikio_request_ceiling < 0:

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,8 @@ 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_executor_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -1306,7 +1297,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 +1312,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