diff --git a/python/cudf_polars/cudf_polars/engine/core.py b/python/cudf_polars/cudf_polars/engine/core.py index 79caea6c8653..367e54a93de4 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 32cf1e4444e7..3b8a4ea5aba3 100644 --- a/python/cudf_polars/cudf_polars/engine/dask.py +++ b/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, diff --git a/python/cudf_polars/cudf_polars/engine/options.py b/python/cudf_polars/cudf_polars/engine/options.py index 42d1677227b8..49523d36c7dd 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, @@ -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 ) diff --git a/python/cudf_polars/cudf_polars/engine/ray.py b/python/cudf_polars/cudf_polars/engine/ray.py index 85d2402d64b7..7dd5b6dd3cf7 100644 --- a/python/cudf_polars/cudf_polars/engine/ray.py +++ b/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 diff --git a/python/cudf_polars/cudf_polars/engine/spmd.py b/python/cudf_polars/cudf_polars/engine/spmd.py index 3465f3ee7207..e8a819c8e9b4 100644 --- a/python/cudf_polars/cudf_polars/engine/spmd.py +++ b/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 {} diff --git a/python/cudf_polars/cudf_polars/utils/config.py b/python/cudf_polars/cudf_polars/utils/config.py index a0a06ebb87cb..49bf59d77f2f 100644 --- a/python/cudf_polars/cudf_polars/utils/config.py +++ b/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: diff --git a/python/cudf_polars/tests/test_config.py b/python/cudf_polars/tests/test_config.py index 458b4c35941a..6c337b25ced6 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,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