Skip to content

pydvl.parallel.futures

init_executor 

init_executor(
    max_workers: Optional[int] = None,
    config: Optional[ParallelConfig] = None,
    **kwargs
) -> Generator[Executor, None, None]

Initializes a futures executor for the given parallel configuration.

PARAMETER DESCRIPTION
max_workers

Maximum number of concurrent tasks.

TYPE: Optional[int] DEFAULT: None

config

instance of ParallelConfig with cluster address, number of cpus, etc.

TYPE: Optional[ParallelConfig] DEFAULT: None

kwargs

Other optional parameter that will be passed to the executor.

DEFAULT: {}

Examples 

from pydvl.parallel.futures import init_executor, ParallelConfig

config = ParallelConfig(backend="ray")
with init_executor(max_workers=1, config=config) as executor:
    future = executor.submit(lambda x: x + 1, 1)
    result = future.result()
assert result == 2

from pydvl.parallel.futures import init_executor
with init_executor() as executor:
    results = list(executor.map(lambda x: x + 1, range(5)))
assert results == [1, 2, 3, 4, 5]

Source code in src/pydvl/parallel/futures/__init__.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@contextmanager
@deprecated(
    target=None,
    deprecated_in="0.9.0",
    remove_in="0.10.0",
)
def init_executor(
    max_workers: Optional[int] = None,
    config: Optional[ParallelConfig] = None,
    **kwargs,
) -> Generator[Executor, None, None]:
    """Initializes a futures executor for the given parallel configuration.

    Args:
        max_workers: Maximum number of concurrent tasks.
        config: instance of [ParallelConfig][pydvl.utils.config.ParallelConfig]
            with cluster address, number of cpus, etc.
        kwargs: Other optional parameter that will be passed to the executor.


    ??? Examples
        ``` python
        from pydvl.parallel.futures import init_executor, ParallelConfig

        config = ParallelConfig(backend="ray")
        with init_executor(max_workers=1, config=config) as executor:
            future = executor.submit(lambda x: x + 1, 1)
            result = future.result()
        assert result == 2
        ```
        ``` python
        from pydvl.parallel.futures import init_executor
        with init_executor() as executor:
            results = list(executor.map(lambda x: x + 1, range(5)))
        assert results == [1, 2, 3, 4, 5]
        ```
    """

    if config is None:
        config = ParallelConfig()

    try:
        cls = ParallelBackend.BACKENDS[config.backend]
        with cls.executor(max_workers=max_workers, config=config, **kwargs) as e:
            yield e
    except KeyError:
        raise NotImplementedError(f"Unexpected parallel backend {config.backend}")