delayed

tpcp.parallel.delayed(func)

Wrap a function to be used in a parallel context.

This is a modified version of joblib.delayed that can run arbitrary callbacks when delayed is called in the main process and when the function is called in the parallel process.

This is useful for example to restore the global config in the parallel process. For this to work, callbacks must be registered using register_global_parallel_callback first.

All uses of delayed in tpcp are using this implementation. This means you can configure your custom callbacks and expect them to work in all tpcp functions that use multiprocessing.

If you need to write your own multiprocessing method using joblib, refer to the example below.

Notes

The getters are called as soon as the delayed function is called in the main process. This means, if you are calling delayed long before the actual parallel execution, the getters might not capture the correct state of the global variables.

Setters might be called multiple times in the same process, if the process pool is reused by multiple jobs. The callbacks should be robust against this and not break.

Active warning_error_context instances are restored automatically for each worker task. Their worker-side lifetime is limited to that task, even when joblib reuses the process. The Python warning filters active when the delayed task is created are restored in process workers. Thread workers share process-global warning filters and therefore use the filters active when the task executes. Use register_parallel_side_channel for custom state that must be restored for one worker task and optionally returned to the parent process.

Examples

This example shows how to use this to make sure the global scikit-learn config is restored in the parallel process used in tpcp. Note, sklearn has a custom workaround for this, which is not compatible with tpcp.

>>> from contextlib import contextmanager
>>> from tpcp.parallel import (
...     Parallel,
...     delayed,
...     register_global_parallel_callback,
...     register_parallel_side_channel,
...     remove_global_parallel_callback,
...     remove_parallel_side_channel,
... )
>>> from sklearn import get_config, set_config
>>>
>>> set_config(assume_finite=True)
>>> def callback():
...     def setter(config):
...         set_config(**config)
...
...     return get_config(), setter
>>>
>>> def worker_func():
...     # This is what would be called in the parallel process
...     # We just return the config here for demonstration purposes
...     config = get_config()
...     return config["assume_finite"]
>>>
>>> # register the callback
>>> name = register_global_parallel_callback(callback)
>>> # call the worker function in parallel
>>> Parallel(n_jobs=2)(delayed(worker_func)() for _ in range(2))
[True, True]
>>> # remove the callback again
>>> remove_global_parallel_callback(name)
>>>
>>> # A side channel can additionally return data from successful worker tasks.
>>> worker_events = []
>>> collected_events = []
>>> @contextmanager
... def restore_events(run_name):
...     worker_events.clear()
...     try:
...         yield lambda: tuple((run_name, event) for event in worker_events)
...     finally:
...         worker_events.clear()
>>> def worker_with_event(value):
...     worker_events.append(f"processed {value}")
...     return value * 2
>>> side_channel = register_parallel_side_channel(
...     lambda: "experiment-1", restore_events, collected_events.extend
... )
>>> Parallel(n_jobs=2)(delayed(worker_with_event)(value) for value in range(2))
[0, 2]
>>> collected_events
[('experiment-1', 'processed 0'), ('experiment-1', 'processed 1')]
>>> remove_parallel_side_channel(side_channel)

Examples using tpcp.parallel.delayed

Caching

Caching

Warning and Error Context

Warning and Error Context