From 4cc6ae2949458e47f0efe42a86368aa7bfab8ab7 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sat, 29 Aug 2026 16:05:43 +0200 Subject: [PATCH 1/2] Add concurrent.futures API --- CHANGELOG.md | 2 + README.md | 16 +- docs/index.rst | 3 +- docs/source/api_concurrent.rst | 110 ++ docs/source/api_futures.rst | 9 +- docs/source/api_multiprocessing.rst | 2 +- docs/source/functions.md | 2 +- docs/source/notebooks/function_chaining.ipynb | 2 +- examples/function_chaining.py | 2 +- lithops/concurrent/__init__.py | 53 + lithops/concurrent/futures.py | 856 ++++++++++++++ lithops/tests/test_concurrent_futures.py | 1033 +++++++++++++++++ 12 files changed, 2079 insertions(+), 11 deletions(-) create mode 100644 docs/source/api_concurrent.rst create mode 100644 lithops/concurrent/__init__.py create mode 100644 lithops/concurrent/futures.py create mode 100644 lithops/tests/test_concurrent_futures.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d9ea8fdd..66592a57b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ### Added +- [API] Added `lithops.concurrent.futures`, a `concurrent.futures`-compatible executor interface (`submit`, eager `map`, stdlib `wait`/`as_completed`) backed by Lithops. See issue #1427. +- [Docs] Renamed the native executor documentation from "Futures API" to "Core API". - [Tests] Added a unit test suite for all non-backend modules (18 files, 876 tests). - [Core] Added a `log_prefix()` helper for uniform log prefixes across core and backends. - [Core] Added a cache of serialized functions to avoid re-uploading the same function. diff --git a/README.md b/README.md index 974cf4b1c..da18d0788 100644 --- a/README.md +++ b/README.md @@ -148,9 +148,9 @@ Supported backends by platform: ## High-level API -Lithops provides two high-level compute APIs and two high-level storage APIs. +Lithops provides three high-level compute APIs and two high-level storage APIs. -### [Futures API](docs/source/api_futures.rst) +### [Core API](docs/source/api_futures.rst) ```python from lithops import FunctionExecutor @@ -163,6 +163,18 @@ with FunctionExecutor() as fexec: print(f.result()) ``` +### [Concurrent Futures API](docs/source/api_concurrent.rst) + +```python +from lithops.concurrent.futures import ProcessPoolExecutor + +def double(i): + return i * 2 + +with ProcessPoolExecutor() as executor: + print(list(executor.map(double, [1, 2, 3, 4]))) +``` + ### [Multiprocessing API](docs/source/api_multiprocessing.rst) ```python diff --git a/docs/index.rst b/docs/index.rst index 81efd2765..024cfd947 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -238,7 +238,7 @@ Lithops is an open-source project, actively maintained and supported by a commun :maxdepth: 0 :caption: Lithops Compute API - source/api_futures.rst + Core API source/functions.md source/worker_granularity.rst source/notebooks/function_chaining.ipynb @@ -264,6 +264,7 @@ Lithops is an open-source project, actively maintained and supported by a commun :maxdepth: 0 :caption: Integrations + source/api_concurrent.rst source/api_multiprocessing.rst source/api_storage_os.rst source/sklearn_joblib.rst diff --git a/docs/source/api_concurrent.rst b/docs/source/api_concurrent.rst new file mode 100644 index 000000000..34f135cdc --- /dev/null +++ b/docs/source/api_concurrent.rst @@ -0,0 +1,110 @@ +Concurrent Futures API +====================== + +``lithops.concurrent.futures`` is a drop-in for Python's +`concurrent.futures `_ +**Executor interface**. Swap the import and the same client keeps working: + +.. code:: python + + # from concurrent.futures import ProcessPoolExecutor, as_completed, wait + from lithops.concurrent.futures import ProcessPoolExecutor, as_completed, wait + + def same_client(executor): + with executor: + future = executor.submit(pow, 323, 1235) + print(future.result()) + print(list(executor.map(abs, [-1, 2, -3]))) + + same_client(ProcessPoolExecutor()) + +``ThreadPoolExecutor`` is provided under the same name for import compatibility; +both names run tasks on Lithops workers. + +This is separate from the :doc:`Core API ` (``lithops.FunctionExecutor``), +whose ``map()`` returns futures, which has ``call_async()`` instead of ``submit()``, +and whose ``wait()`` is a method with Lithops-specific ``return_when`` values. + + +Implemented standard-library surface +------------------------------------ + +The module exports the names application code actually uses: + +* **Executor** — ``submit()``, eager ``map()`` (results, not futures), + ``shutdown(wait=True, *, cancel_futures=False)``, context manager +* **ProcessPoolExecutor** / **ThreadPoolExecutor** — Lithops-backed pools +* **Future** — subclass of ``concurrent.futures.Future``: + ``result()``, ``exception()``, ``done()``, ``running()``, ``cancel()``, + ``cancelled()``, ``add_done_callback()`` +* **wait** / **as_completed** — same contract and constants + (``FIRST_COMPLETED``, ``FIRST_EXCEPTION``, ``ALL_COMPLETED``) +* **Exceptions** — ``CancelledError``, ``TimeoutError``, ``BrokenExecutor``, + ``InvalidStateError`` + +``InterpreterPoolExecutor`` (Python 3.14) is not implemented: Lithops has no +isolated-interpreter workers. + +``map()`` submits one Lithops ``map()`` job for the whole iterable, not one +``submit()`` per item. Callables travel through a trampoline, so builtins such +as ``abs`` and ``pow`` work. Its ``chunksize`` is how many items each Lithops +worker takes, which is what it means for the standard ``ProcessPoolExecutor`` +too; left unset, the ``chunksize`` of the Lithops configuration applies rather +than the standard library default of one item per worker. + +.. code:: python + + from lithops.concurrent.futures import FunctionExecutor, as_completed + + def load(url): + return url, len(url) + + with FunctionExecutor() as executor: + futures = {executor.submit(load, url): url for url in ('a', 'bb', 'ccc')} + for future in as_completed(futures): + url, size = future.result() + print(url, size) + +Mode-specific subclasses ``LocalhostExecutor``, ``ServerlessExecutor`` and +``StandaloneExecutor`` pin the Lithops execution mode the same way the Core API +executors do. You can wrap an existing Core API executor:: + + import lithops + from lithops.concurrent.futures import FunctionExecutor + + fexec = lithops.FunctionExecutor() + with FunctionExecutor(executor=fexec) as executor: + print(executor.submit(pow, 2, 8).result()) + + +Runtime differences +------------------- + +The *API* matches ``concurrent.futures``. The *runtime* is Lithops: + +* Workers are Lithops activations, not local threads or ``multiprocessing`` processes. + ``mp_context``, ``max_tasks_per_child``, and ``thread_name_prefix`` are ignored. +* ``initializer`` / ``initargs`` are not supported (workers are ephemeral) and raise + ``NotImplementedError`` if provided. +* ``cancel()`` cannot stop a job Lithops has already dispatched. ``submit()`` marks + the future as running immediately, so ``cancel()`` returns ``False``, and + ``shutdown(cancel_futures=True)`` therefore still waits the calls out. +* Lithops job options (``runtime_memory``, ``extra_env``, ``execution_timeout``, + ``include_modules``, ``exclude_modules``) are set on the executor. Keyword + arguments to ``submit(fn, *args, **kwargs)`` are passed to ``fn``. +* Each ``Future`` also exposes ``lithops_future`` and ``stats``. +* ``RetryingFunctionExecutor`` cannot be wrapped. Its retries are driven from + its own ``wait()``, which this adapter never calls, so wrapping it would + quietly give you no retries at all. Wrap the ``FunctionExecutor`` it holds. + +Completion is tracked by the Lithops job monitor, the same one the native +``wait()`` uses: one batched poll per round for the whole job rather than one +status read per call. Results are downloaded off the tracking thread, so a +slow object does not hold up the futures behind it, and ``done()`` never +blocks on storage. + +.. automodule:: lithops.concurrent.futures + :members: FunctionExecutor, LocalhostExecutor, ServerlessExecutor, + StandaloneExecutor, ProcessPoolExecutor, ThreadPoolExecutor, + Future, wait, as_completed + :show-inheritance: diff --git a/docs/source/api_futures.rst b/docs/source/api_futures.rst index 56f5d0c74..c7a81c691 100644 --- a/docs/source/api_futures.rst +++ b/docs/source/api_futures.rst @@ -1,7 +1,8 @@ +.. _core-api: .. _futures-api: -Lithops Futures API -=================== +Lithops Core API +================ The core abstraction in Lithops is the **executor**, which is responsible for orchestrating the execution of your functions across different environments. @@ -41,8 +42,8 @@ By default, executors load configuration from the Lithops configuration file (e. This layered executor design lets Lithops provide a powerful, unified API for parallel function execution — from local development to multi-cloud production deployments, with fault tolerance and retries built in. -Futures API Reference ---------------------- +Core API Reference +------------------ .. automodule:: lithops.executors :members: diff --git a/docs/source/api_multiprocessing.rst b/docs/source/api_multiprocessing.rst index 104334899..f2b1aa84e 100644 --- a/docs/source/api_multiprocessing.rst +++ b/docs/source/api_multiprocessing.rst @@ -13,7 +13,7 @@ Before utilizing this API, you will need to install its dependencies: Process and Pool ---------------- -`Processes `_ and `Pool `_ are the abstractions used in multiprocessing to parallelize computation. They interact directly with Lithops' Futures API. +`Processes `_ and `Pool `_ are the abstractions used in multiprocessing to parallelize computation. They interact directly with Lithops' Core API. .. code:: python diff --git a/docs/source/functions.md b/docs/source/functions.md index 19911d622..06f7dd1ed 100644 --- a/docs/source/functions.md +++ b/docs/source/functions.md @@ -5,7 +5,7 @@ This document describes how to invoke functions based on the *iterdata* variable Reserved parameters ------------------- -Reserved parameters are only accessible when using the [Futures API](./api_futures.rst). +Reserved parameters are only accessible when using the [Core API](./api_futures.rst). - **id**: To get the call id. For instance, if you spawn 10 activations of a function, you will get here a number from 0 to 9, for example: [map.py](https://github.com/lithops-cloud/lithops/blob/master/examples/map.py) diff --git a/docs/source/notebooks/function_chaining.ipynb b/docs/source/notebooks/function_chaining.ipynb index 23916910d..1cc535e09 100644 --- a/docs/source/notebooks/function_chaining.ipynb +++ b/docs/source/notebooks/function_chaining.ipynb @@ -19,7 +19,7 @@ "Lithops does not download the intermediate results to the local client; instead, the intermediate results are read\n", "directly by the next function.\n", "\n", - "It currently works with the Futures API, and you can chain the `map()`, `map_reduce()`, `wait()`, and `get_result()`\n", + "It currently works with the Core API, and you can chain the `map()`, `map_reduce()`, `wait()`, and `get_result()`\n", "methods. Note that the returning value of one function must match the signature of the next function when chaining\n", "multiple `map()` calls. View the next examples:\n", "\n", diff --git a/examples/function_chaining.py b/examples/function_chaining.py index 8d43527c9..7f7826de1 100644 --- a/examples/function_chaining.py +++ b/examples/function_chaining.py @@ -1,6 +1,6 @@ """ Simple Lithops example using the function chaining pattern -in the Futures API. +in the Core API. """ import lithops diff --git a/lithops/concurrent/__init__.py b/lithops/concurrent/__init__.py new file mode 100644 index 000000000..a51f24c00 --- /dev/null +++ b/lithops/concurrent/__init__.py @@ -0,0 +1,53 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +from .futures import ( + ALL_COMPLETED, + FIRST_COMPLETED, + FIRST_EXCEPTION, + BrokenExecutor, + CancelledError, + Executor, + FunctionExecutor, + Future, + InvalidStateError, + LocalhostExecutor, + ProcessPoolExecutor, + ServerlessExecutor, + StandaloneExecutor, + ThreadPoolExecutor, + TimeoutError, + as_completed, + wait, +) + +__all__ = [ + 'ALL_COMPLETED', + 'FIRST_COMPLETED', + 'FIRST_EXCEPTION', + 'BrokenExecutor', + 'CancelledError', + 'Executor', + 'FunctionExecutor', + 'Future', + 'InvalidStateError', + 'LocalhostExecutor', + 'ProcessPoolExecutor', + 'ServerlessExecutor', + 'StandaloneExecutor', + 'ThreadPoolExecutor', + 'TimeoutError', + 'as_completed', + 'wait', +] diff --git a/lithops/concurrent/futures.py b/lithops/concurrent/futures.py new file mode 100644 index 000000000..ebafbde4d --- /dev/null +++ b/lithops/concurrent/futures.py @@ -0,0 +1,856 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +""" +concurrent.futures-compatible executors backed by Lithops. + +The native Lithops executors (``lithops.FunctionExecutor`` and friends) +are intentionally different from ``concurrent.futures``: ``map()`` returns +futures, there is no ``submit()``, and ``wait()`` lives on the executor. +This module is the drop-in interface for code that already talks to +``ThreadPoolExecutor`` / ``ProcessPoolExecutor``: + +.. code-block:: python + + from lithops.concurrent.futures import ProcessPoolExecutor + + with ProcessPoolExecutor() as executor: + future = executor.submit(pow, 2, 8) + print(future.result()) + print(list(executor.map(abs, [-1, 2, -3]))) + +``Future`` subclasses ``concurrent.futures.Future``, so the standard +library's ``wait()`` and ``as_completed()`` work unchanged, including +with futures from other executors. +""" + +from __future__ import annotations + +import collections +import inspect +import logging +import threading +import time +from concurrent.futures import ( + ALL_COMPLETED, + FIRST_COMPLETED, + FIRST_EXCEPTION, + BrokenExecutor, + CancelledError, + Executor as _CfExecutor, + Future as _CfFuture, + InvalidStateError, + ThreadPoolExecutor as _CfThreadPool, + TimeoutError, + as_completed as _cf_as_completed, + wait as _cf_wait, +) + +from lithops.executors import ( + FunctionExecutor as _LithopsFunctionExecutor, + LocalhostExecutor as _LithopsLocalhostExecutor, + ServerlessExecutor as _LithopsServerlessExecutor, + StandaloneExecutor as _LithopsStandaloneExecutor, +) +from lithops.future import ResponseFuture as _ResponseFuture +from lithops.retries import RetryingFunctionExecutor as _RetryingFunctionExecutor + +logger = logging.getLogger(__name__) + +# Re-exported so ``from lithops.concurrent.futures import wait`` is the +# concurrent.futures API, not lithops.wait (which uses a different +# return_when convention). +Executor = _CfExecutor + +__all__ = [ + 'ALL_COMPLETED', + 'FIRST_COMPLETED', + 'FIRST_EXCEPTION', + 'BrokenExecutor', + 'CancelledError', + 'Executor', + 'FunctionExecutor', + 'Future', + 'InvalidStateError', + 'LocalhostExecutor', + 'ProcessPoolExecutor', + 'ServerlessExecutor', + 'StandaloneExecutor', + 'ThreadPoolExecutor', + 'TimeoutError', + 'as_completed', + 'wait', +] + + +# How often the watcher re-reads the state the Lithops job monitor keeps in +# memory. Cheap, so it can be tight +_WATCHER_POLL_SEC = 0.1 + +# How often the watcher asks storage directly, which it only does for an +# executor that has no job monitor of its own +_UNMONITORED_POLL_SEC = 1.0 + +# Downloads of the results run here rather than in the watcher, so that one +# slow object does not hold up every other completion. Same size Lithops +# uses for its own wait() +_RESOLVER_THREADS = 64 + + +def _call(fn, args, kwargs): + """ + Worker-side trampoline for submit(). Lithops binds each iterdata + element to the map function's signature, so *args/**kwargs have to + travel as a single tuple rather than as Lithops extra_args + """ + return fn(*args, **kwargs) + + +def _result_or_cancel(fut, timeout=None): + """ + Same helper concurrent.futures.Executor.map uses: wait for one + result, then drop the reference so a completed future can be freed + """ + try: + try: + return fut.result(timeout) + finally: + fut.cancel() + finally: + del fut + + +def _lithops_finished(lf): + """ + True once Lithops has a terminal or readable status. ResponseFuture + uses properties; duck-typed wrappers (RetryingFuture) expose a subset + """ + return bool( + getattr(lf, 'done', False) + or getattr(lf, 'error', False) + or getattr(lf, 'success', False) + or getattr(lf, 'ready', False) + ) + + +def _lithops_unknown(lf): + """ + Lithops moves a future to Unknown when the job it belonged to was + abandoned, typically by an interrupted native wait(). ``done`` is true + for it, but there is no result behind it and no exception either + """ + return getattr(lf, '_state', None) == _ResponseFuture.State.Unknown + + +def _unwrap(lf): + """The ResponseFuture behind a wrapper such as RetryingFuture""" + return getattr(lf, 'response_future', lf) + + +def _exception_from_lithops(lf): + """ + Turns the (type, value, traceback) tuple Lithops stores into the + exception instance concurrent.futures.Future.set_exception expects + """ + exc = getattr(lf, '_exception', None) + if isinstance(exc, tuple) and len(exc) >= 2: + value = exc[1] + if isinstance(value, BaseException): + return value + # A fresh ResponseFuture holds a bare Exception() as its placeholder, so + # an empty one means the failure was never given a reason + if isinstance(exc, BaseException) and (exc.args or type(exc) is not Exception): + return exc + return Exception(_failure_message(lf)) + + +def _failure_message(lf): + call_id = getattr(lf, 'call_id', None) + where = f' (call {call_id})' if call_id else '' + return f'The Lithops call failed without reporting an exception{where}' + + +def _unknown_state_error(lf): + call_id = getattr(lf, 'call_id', None) + where = f' (call {call_id})' if call_id else '' + return BrokenExecutor( + f'Lithops lost track of the activation{where}: it is marked done ' + 'but never reported a status, so there is no result to read' + ) + + +def _storage_of(executor): + """ + InternalStorage of a FunctionExecutor, or of an executor that wraps one + """ + inner = getattr(executor, 'executor', executor) + return getattr(inner, 'internal_storage', None) + + +def _with_storage(fn, storage, **kwargs): + """ + Calls a Lithops future method, handing it the internal storage handler + only when its signature takes one. Duck-typed futures do not all accept + it, and a blanket ``except TypeError`` would also swallow one raised + inside the call and run it a second time + """ + try: + params = inspect.signature(fn).parameters + except (TypeError, ValueError): + params = {} + accepts_storage = 'internal_storage' in params or any( + param.kind is inspect.Parameter.VAR_KEYWORD + for param in params.values() + ) + if accepts_storage and storage is not None: + kwargs['internal_storage'] = storage + return fn(**kwargs) + + +def _set_result(fut, value): + try: + fut.set_result(value) + except InvalidStateError: + pass + + +def _set_exception(fut, exc): + try: + fut.set_exception(exc) + except InvalidStateError: + pass + + +class Future(_CfFuture): + """ + A concurrent.futures.Future backed by a Lithops ResponseFuture. + + Created by :meth:`FunctionExecutor.submit`. The underlying Lithops + future is available as :attr:`lithops_future` for stats and other + Lithops-specific attributes. + """ + + def __init__(self, lithops_future=None, adapter=None): + super().__init__() + self._lithops_future = lithops_future + self._adapter = adapter + + @property + def lithops_future(self): + """The Lithops ResponseFuture this object is tracking.""" + return self._lithops_future + + @property + def stats(self): + """Execution stats from the Lithops future, once they are available.""" + lf = self._lithops_future + return getattr(lf, 'stats', {}) if lf is not None else {} + + def _sync(self): + """ + Hands the future over for resolution the moment Lithops has a status + for it, so a caller arriving between two watcher rounds does not have + to sit through one. Never blocks: the download happens elsewhere and + the caller goes on to wait on its own condition, timeout included + """ + adapter = self._adapter + if adapter is not None: + adapter._nudge(self) + + def done(self): + self._sync() + return super().done() + + def result(self, timeout=None): + self._sync() + return super().result(timeout) + + def exception(self, timeout=None): + self._sync() + return super().exception(timeout) + + +class FunctionExecutor(_CfExecutor): + """ + concurrent.futures.Executor that runs callables on Lithops workers. + + ``submit(fn, *args, **kwargs)`` and ``map(fn, *iterables)`` follow the + standard library: ``map`` is eager and yields *results*, not futures. + Internally, ``map`` is a single Lithops ``map()`` job rather than one + ``submit`` per item, so a large iterator still benefits from Lithops + batching. + + :param max_workers: Passed through to the Lithops compute backend + :param executor: An existing ``lithops.FunctionExecutor`` (or + compatible object) to wrap. When omitted, one is created from + ``**kwargs`` + :param initializer: Not supported; Lithops workers are ephemeral. + Providing a callable raises ``NotImplementedError`` + :param initargs: Ignored unless ``initializer`` is set + :param runtime_memory: Memory (MB) for every submitted call + :param extra_env: Extra environment variables for every submitted call + :param execution_timeout: Max seconds each function activation may run + :param include_modules: Modules to pickle into the worker payload + :param exclude_modules: Modules to keep out of the worker payload + :param kwargs: Forwarded to the Lithops executor constructor + (``config``, ``backend``, ``storage``, ``log_level``, ...) + """ + + _executor_cls = _LithopsFunctionExecutor + + def __init__( + self, + max_workers=None, + *, + executor=None, + initializer=None, + initargs=(), + runtime_memory=None, + extra_env=None, + execution_timeout=None, + include_modules=None, + exclude_modules=None, + **kwargs, + ): + if initializer is not None: + raise NotImplementedError( + 'initializer is not supported; Lithops workers are ephemeral' + ) + if isinstance(executor, _RetryingFunctionExecutor): + # Its retries are driven from its own wait(), which nothing here + # calls, so wrapping it would quietly give you no retries at all + raise TypeError( + 'RetryingFunctionExecutor is not supported: its retries are ' + 'driven by its own wait(), which this adapter never calls. ' + 'Wrap the FunctionExecutor it holds instead' + ) + + # Drop-in replacements of ProcessPoolExecutor / ThreadPoolExecutor + # pass these; they are not Lithops backend keys + for key in ('mp_context', 'max_tasks_per_child', 'thread_name_prefix'): + kwargs.pop(key, None) + + self._runtime_memory = runtime_memory + self._extra_env = extra_env + self._execution_timeout = execution_timeout + self._include_modules = include_modules + self._exclude_modules = exclude_modules + + self._owns_executor = executor is None + if executor is None: + if max_workers is not None: + kwargs['max_workers'] = max_workers + executor = self._executor_cls(**kwargs) + elif max_workers is not None: + logger.debug( + 'max_workers is ignored when wrapping an existing executor' + ) + self._inner = executor + + self._lock = threading.RLock() + self._is_shutdown = False + self._broken = None + self._torn_down = False + self._pending = {} + self._resolving = set() + self._wake = threading.Event() + self._stop_event = threading.Event() + self._watcher = None + self._resolver = None + self._reaper = None + + @property + def lithops_executor(self): + """The wrapped Lithops FunctionExecutor.""" + return self._inner + + def _lithops_inner(self): + """ + The Lithops FunctionExecutor, unwrapping an executor that holds one + """ + return getattr(self._inner, 'executor', self._inner) + + def _job_monitor(self): + return getattr(self._lithops_inner(), 'job_monitor', None) + + def _job_kwargs(self): + kwargs = {} + if self._runtime_memory is not None: + kwargs['runtime_memory'] = self._runtime_memory + if self._extra_env is not None: + kwargs['extra_env'] = self._extra_env + if self._execution_timeout is not None: + kwargs['timeout'] = self._execution_timeout + if self._include_modules is not None: + kwargs['include_modules'] = self._include_modules + if self._exclude_modules is not None: + kwargs['exclude_modules'] = self._exclude_modules + return kwargs + + # -- completion tracking ------------------------------------------------ + + def _ensure_watcher(self): + """ + Starts the thread that copies Lithops completion into the + concurrent.futures.Future condition, so wait() and as_completed() + wake up without the caller having to poll + """ + with self._lock: + if self._watcher is not None and self._watcher.is_alive(): + return + self._stop_event.clear() + self._watcher = threading.Thread( + target=self._watch, + name='lithops-cf-watcher', + daemon=True, + ) + self._watcher.start() + + def _watch(self): + poll = ( + _WATCHER_POLL_SEC if self._job_monitor() is not None + else _UNMONITORED_POLL_SEC + ) + try: + while not self._stop_event.is_set(): + with self._lock: + pairs = [ + (fut, lf) for fut, lf in self._pending.items() + if fut not in self._resolving + ] + idle = not self._pending + if idle and self._is_shutdown: + break + if pairs: + self._revive_job_monitor([lf for _, lf in pairs]) + for fut, lf in pairs: + if _CfFuture.done(fut): + self._forget(fut) + elif self._is_ready(lf): + self._schedule(fut, lf) + self._wake.wait(timeout=poll) + self._wake.clear() + except BaseException as exc: + self._break(exc) + + def _revive_job_monitor(self, lfs): + """ + The Lithops job monitor is what moves futures into their Ready state, + and the invoker starts one per job. It is a daemon that winds down + once everything it knows about is done, so a job submitted just as it + was exiting can be left unwatched. Native wait() guards the same way + """ + job_monitor = self._job_monitor() + # No monitor thread has ever run: nothing was submitted through the + # invoker, and JobMonitor.is_alive() would fail on the missing one + if job_monitor is None or getattr(job_monitor, 'monitor', None) is None: + return + try: + if job_monitor.is_alive(): + return + unfinished = [ + _unwrap(lf) for lf in lfs if not _lithops_finished(lf) + ] + if unfinished: + job_monitor.start(fs=unfinished) + except Exception: + logger.debug( + 'Could not restart the Lithops job monitor', exc_info=True + ) + + def _is_ready(self, lf): + """ + Whether the Lithops future has a status waiting to be read. + + The job monitor keeps that state up to date in memory for every + executor that has one, so this costs nothing. Only an executor + without a monitor is polled through storage, and then once a second: + a per-future status read on the watcher interval would be one storage + request per future per round + """ + if _lithops_finished(lf): + return True + if self._job_monitor() is not None: + return False + return self._peek_status(lf) is not None + + def _peek_status(self, lf): + status_fn = getattr(lf, 'status', None) + if status_fn is None: + return None + try: + return _with_storage( + status_fn, + _storage_of(self._inner), + throw_except=False, + check_only=True, + ) + except Exception: + logger.debug('Error reading the status of a call', exc_info=True) + return None + + def _nudge(self, fut): + """ + Fast path for a caller that reached the future first. Only looks at + state already in memory, so done() and a timed result() never block + on storage + """ + if _CfFuture.done(fut) or self._broken is not None: + return + lf = fut._lithops_future + if lf is not None and _lithops_finished(lf): + self._schedule(fut, lf) + + def _resolver_pool(self): + with self._lock: + if self._resolver is None: + self._resolver = _CfThreadPool( + max_workers=_RESOLVER_THREADS, + thread_name_prefix='lithops-cf-resolver', + ) + return self._resolver + + def _schedule(self, fut, lf): + """ + Queues the download of one result, at most once per future + """ + with self._lock: + if fut in self._resolving or fut not in self._pending: + return + self._resolving.add(fut) + try: + self._resolver_pool().submit(self._resolve, fut, lf) + except RuntimeError: + # The pool is already shutting down; finish it here instead of + # leaving the future hanging + self._resolve(fut, lf) + + def _resolve(self, fut, lf): + try: + if not _CfFuture.done(fut): + self._apply_lithops_outcome(fut, lf) + except Exception as exc: + _set_exception(fut, exc) + finally: + with self._lock: + self._resolving.discard(fut) + self._pending.pop(fut, None) + self._wake.set() + + def _apply_lithops_outcome(self, fut, lf): + storage = _storage_of(self._inner) + + status_fn = getattr(lf, 'status', None) + if status_fn is not None: + _with_storage(status_fn, storage, throw_except=False) + + if getattr(lf, 'error', False): + _set_exception(fut, _exception_from_lithops(lf)) + return + if _lithops_unknown(lf): + # Counts as done for Lithops, but there is no result behind it. + # Saying so beats handing the caller a silent None + _set_exception(fut, _unknown_state_error(lf)) + return + + value = _with_storage(lf.result, storage, throw_except=False) + # result() marks the future as failed when the output never showed up + if getattr(lf, 'error', False): + _set_exception(fut, _exception_from_lithops(lf)) + return + _set_result(fut, value) + + def _break(self, exc): + """ + The watcher is the only thing that hands futures to the resolver, so + if it dies every result() and wait() would block for good. Fail them + loudly instead + """ + logger.error( + 'The Lithops concurrent.futures watcher thread died', exc_info=exc + ) + broken = BrokenExecutor(f'the Lithops watcher thread died: {exc!r}') + broken.__cause__ = exc + with self._lock: + self._broken = broken + pending = list(self._pending) + self._pending.clear() + self._resolving.clear() + for fut in pending: + _set_exception(fut, broken) + + def _forget(self, fut): + with self._lock: + self._pending.pop(fut, None) + + def _track(self, lfs): + futures = [] + with self._lock: + for lf in lfs: + fut = Future(lithops_future=lf, adapter=self) + # Lithops has already dispatched the call, so the future is + # running as far as the caller is concerned and cancel() has + # to decline + fut.set_running_or_notify_cancel() + self._pending[fut] = lf + futures.append(fut) + self._ensure_watcher() + self._wake.set() + return futures + + def _check_running(self): + if self._broken is not None: + raise self._broken + if self._is_shutdown: + raise RuntimeError('cannot schedule new futures after shutdown') + + # -- concurrent.futures.Executor ---------------------------------------- + + def submit(self, fn, /, *args, **kwargs): + with self._lock: + self._check_running() + payload = (fn, args, kwargs) + job_kwargs = self._job_kwargs() + call_async = getattr(self._inner, 'call_async', None) + if call_async is not None: + lf = call_async(_call, payload, **job_kwargs) + else: + # A duck-typed executor may only expose map() + lf = self._inner.map(_call, [payload], **job_kwargs)[0] + return self._track([lf])[0] + + def map( + self, + fn, + *iterables, + timeout=None, + chunksize=None, + buffersize=None, + ): + """ + Eager map, as in the standard library: it returns an iterator over + the *results*, and every call is submitted before it does. + + ``chunksize`` is how many items each Lithops worker takes, which is + what it means for the standard ``ProcessPoolExecutor`` too. Left + unset, the Lithops configuration decides, rather than the standard + library default of one item per worker overriding it + """ + if chunksize is not None and chunksize < 1: + raise ValueError("chunksize must be >= 1.") + if buffersize is not None and buffersize < 0: + raise ValueError("buffersize must be >= 0") + + end_time = None if timeout is None else timeout + time.monotonic() + iterator = zip(*iterables) + + def take(n): + if n is None: + return list(iterator) + batch = [] + for _ in range(n): + try: + batch.append(next(iterator)) + except StopIteration: + break + return batch + + def submit_batch(items): + if not items: + return [] + with self._lock: + self._check_running() + lfs = self._inner.map( + _call, + [(fn, args, {}) for args in items], + chunksize=chunksize, + **self._job_kwargs() + ) + return self._track(lfs) + + fs = collections.deque( + submit_batch(take(buffersize if buffersize else None)) + ) + + def result_iterator(): + try: + while fs: + remaining = ( + None if end_time is None + else end_time - time.monotonic() + ) + yield _result_or_cancel(fs.popleft(), remaining) + if buffersize: + fs.extend(submit_batch(take(buffersize - len(fs)))) + finally: + for future in fs: + future.cancel() + + return result_iterator() + + def shutdown(self, wait=True, *, cancel_futures=False): + with self._lock: + self._is_shutdown = True + pending = list(self._pending) + if cancel_futures: + # Lithops cannot recall an activation it already dispatched, so + # every future here is running and declines. Kept so that callers + # passing the standard argument still work + for fut in pending: + fut.cancel() + self._wake.set() + + if wait: + self._drain(pending) + elif pending: + self._start_reaper(pending) + else: + self._release() + + def _drain(self, pending): + still = [fut for fut in pending if not _CfFuture.done(fut)] + if still: + _cf_wait(still) + self._release() + + def _start_reaper(self, pending): + """ + shutdown(wait=False) returns right away, but the Lithops executor + still has to be given back once the calls in flight are done, or its + monitor and invoker threads outlive it + """ + with self._lock: + if self._reaper is not None or self._torn_down: + return + self._reaper = threading.Thread( + target=self._drain, + args=(pending,), + name='lithops-cf-reaper', + daemon=True, + ) + self._reaper.start() + + def _release(self): + with self._lock: + if self._torn_down: + return + self._torn_down = True + self._stop_watcher() + self._stop_resolver() + self._clean_job_data() + self._teardown_inner() + + def _stop_watcher(self): + self._stop_event.set() + self._wake.set() + watcher = self._watcher + if watcher is not None and watcher is not threading.current_thread(): + watcher.join(timeout=5) + self._watcher = None + + def _stop_resolver(self): + with self._lock: + pool, self._resolver = self._resolver, None + if pool is not None: + # Not waited on: every future handed to it is already resolved by + # the time we get here, and waiting from inside one of its own + # threads would deadlock + pool.shutdown(wait=False) + + def _clean_job_data(self): + """ + Drops the temporary objects the jobs left in storage, which the + native executor does from wait(). Without it they sit there until the + atexit hook runs, which in a long-lived process can be a long while + """ + inner = self._lithops_inner() + if not self._owns_executor or not getattr(inner, 'data_cleaner', False): + return + try: + inner.clean(clean_cloudobjects=False) + except Exception: + logger.debug('Error cleaning temporary job data', exc_info=True) + + def _teardown_inner(self): + if not self._owns_executor: + return + try: + self._inner.__exit__(None, None, None) + except Exception: + logger.debug( + 'Error shutting down the Lithops executor', exc_info=True + ) + + +class LocalhostExecutor(FunctionExecutor): + """FunctionExecutor pinned to the Lithops localhost backend.""" + + _executor_cls = _LithopsLocalhostExecutor + + +class ServerlessExecutor(FunctionExecutor): + """FunctionExecutor pinned to a Lithops serverless backend.""" + + _executor_cls = _LithopsServerlessExecutor + + +class StandaloneExecutor(FunctionExecutor): + """FunctionExecutor pinned to a Lithops standalone backend.""" + + _executor_cls = _LithopsStandaloneExecutor + + +class ProcessPoolExecutor(FunctionExecutor): + """ + Drop-in replacement for ``concurrent.futures.ProcessPoolExecutor``. + + Tasks run on Lithops workers (localhost, serverless, or standalone) + instead of a local ``multiprocessing`` pool. Constructor arguments + that only apply to the standard library (``mp_context``, + ``max_tasks_per_child``) are ignored. + """ + + +class ThreadPoolExecutor(FunctionExecutor): + """ + Drop-in replacement for ``concurrent.futures.ThreadPoolExecutor``. + + Tasks still run on Lithops workers, not in local threads. Use this + name when swapping ``from concurrent.futures import ThreadPoolExecutor``. + ``thread_name_prefix`` is ignored. + """ + + +def _sync_all(fs): + """ + Starts resolving any future Lithops already has a status for, before + handing off to the standard library, so wait() / as_completed() do not + sit through a watcher interval for work that is already done + """ + for fut in fs: + sync = getattr(fut, '_sync', None) + if sync is not None: + sync() + + +def wait(fs, timeout=None, return_when=ALL_COMPLETED): + """Wait for futures to complete. Same contract as concurrent.futures.wait.""" + _sync_all(fs) + return _cf_wait(fs, timeout=timeout, return_when=return_when) + + +def as_completed(fs, timeout=None): + """Yield futures as they complete. Same contract as concurrent.futures.as_completed.""" + _sync_all(fs) + return _cf_as_completed(fs, timeout=timeout) diff --git a/lithops/tests/test_concurrent_futures.py b/lithops/tests/test_concurrent_futures.py new file mode 100644 index 000000000..152f450e4 --- /dev/null +++ b/lithops/tests/test_concurrent_futures.py @@ -0,0 +1,1033 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +import asyncio +import concurrent.futures as cf +import threading +import time + +import pytest + +import lithops +import lithops.concurrent +from lithops.concurrent.futures import ( + ALL_COMPLETED, + FIRST_COMPLETED, + FIRST_EXCEPTION, + BrokenExecutor, + FunctionExecutor, + Future, + LocalhostExecutor, + ProcessPoolExecutor, + ServerlessExecutor, + StandaloneExecutor, + ThreadPoolExecutor, + TimeoutError, + _call, + _exception_from_lithops, + as_completed, + wait, +) +from lithops.future import ResponseFuture +from lithops.retries import RetryingFunctionExecutor +from lithops.tests.functions import simple_map_function + + +class FakeLithopsFuture: + """ + In-memory stand-in for ResponseFuture. + + It models the parts of the contract the adapter depends on: the state + machine (`ready` once a status exists, `done` once the output has been + read), `status()` returning None while the call is still running, and a + `result()` that can be made to block the way a storage download does + """ + + def __init__(self, value=None, exc=None, finished=True, call_id='00000'): + self.call_id = call_id + self.stats = {'worker_exec_time': 0.01} + self._value = None + self._exception = Exception() + self._state = ResponseFuture.State.Invoked + self._status_calls = 0 + self._result_calls = 0 + self._storage_seen = [] + self._result_gate = None + if finished or exc is not None: + self.finish(value=value, exc=exc) + + # -- the ResponseFuture properties the adapter reads ------------------- + + @property + def ready(self): + return self._state == ResponseFuture.State.Ready + + @property + def error(self): + return self._state == ResponseFuture.State.Error + + @property + def success(self): + return self._state in ( + ResponseFuture.State.Success, ResponseFuture.State.Error + ) + + @property + def done(self): + return self._state in ( + ResponseFuture.State.Done, + ResponseFuture.State.Error, + ResponseFuture.State.Unknown, + ) + + def status(self, throw_except=True, internal_storage=None, check_only=False): + self._status_calls += 1 + self._storage_seen.append(internal_storage) + if self._state == ResponseFuture.State.Invoked: + return None + if self.error and throw_except: + raise self._exception[1] + return {'type': '__end__'} + + def result(self, throw_except=True, internal_storage=None, + retries=10, wait_dur_sec=1): + self._result_calls += 1 + self._storage_seen.append(internal_storage) + if self._result_gate is not None: + self._result_gate.wait() + if self.error: + if throw_except: + raise self._exception[1] + return None + self._state = ResponseFuture.State.Done + return self._value + + # -- test helpers ------------------------------------------------------ + + def finish(self, value=None, exc=None): + if exc is not None: + self._exception = (type(exc), exc, None) + self._state = ResponseFuture.State.Error + else: + self._value = value + self._state = ResponseFuture.State.Ready + + def lose(self): + """Reproduces the state an interrupted native wait() leaves behind""" + self._state = ResponseFuture.State.Unknown + + def block_result(self): + self._result_gate = threading.Event() + return self._result_gate + + +class FakeMonitor: + """The daemon thread JobMonitor owns, which exits once its job is done""" + + def __init__(self): + self.alive = True + self.futures = [] + + def is_alive(self): + return self.alive + + +class FakeJobMonitor: + """ + Stand-in for lithops.monitor.JobMonitor. Records the futures it was + asked to track so a test can tell whether the adapter restarted it + """ + + def __init__(self, started=True): + self.monitor = FakeMonitor() if started else None + self.starts = [] + self.stopped = False + + def start(self, fs, **kwargs): + if self.monitor is None: + self.monitor = FakeMonitor() + self.monitor.alive = True + self.monitor.futures = list(fs) + self.starts.append(list(fs)) + + def is_alive(self): + return self.monitor.is_alive() + + def stop(self): + self.stopped = True + if self.monitor is not None: + self.monitor.alive = False + + +class FakeInnerExecutor: + """ + Stand-in for lithops.FunctionExecutor. A MagicMock would answer every + attribute, which hides whether the adapter reached for the job monitor + or fell back to polling storage + """ + + def __init__(self, call_result=None, map_results=None, job_monitor=True, + data_cleaner=False): + self.internal_storage = object() + self.job_monitor = FakeJobMonitor() if job_monitor else None + self.data_cleaner = data_cleaner + self.call_async_calls = [] + self.map_calls = [] + self.cleaned = [] + self.exited = False + self._call_result = call_result + self._call_results = None + self._map_results = map_results + self._map_side_effect = None + + def call_async(self, func, data, **kwargs): + self.call_async_calls.append((func, data, kwargs)) + if self._call_results is not None: + return self._call_results.pop(0) + if self._call_result is None: + self._call_result = FakeLithopsFuture(value=256) + return self._call_result + + def map(self, map_function, map_iterdata, **kwargs): + self.map_calls.append((map_function, list(map_iterdata), kwargs)) + if self._map_side_effect is not None: + return self._map_side_effect.pop(0) + if self._map_results is None: + self._map_results = [ + FakeLithopsFuture(value=1), + FakeLithopsFuture(value=2), + FakeLithopsFuture(value=3), + ] + return self._map_results + + def clean(self, **kwargs): + self.cleaned.append(kwargs) + + def __exit__(self, exc_type, exc_value, traceback): + self.exited = True + if self.job_monitor is not None: + self.job_monitor.stop() + + +def _adapter(inner=None, **kwargs): + return FunctionExecutor(executor=inner or FakeInnerExecutor(), **kwargs) + + +class TestConcurrentFuturesApiSurface: + + def test_executor_is_concurrent_futures_executor(self): + assert issubclass(FunctionExecutor, cf.Executor) + assert issubclass(ProcessPoolExecutor, cf.Executor) + assert issubclass(ThreadPoolExecutor, FunctionExecutor) + assert issubclass(LocalhostExecutor, FunctionExecutor) + assert issubclass(ServerlessExecutor, FunctionExecutor) + assert issubclass(StandaloneExecutor, FunctionExecutor) + + def test_future_is_concurrent_futures_future(self): + assert issubclass(Future, cf.Future) + + def test_pool_executor_names_are_function_executor(self): + assert issubclass(ProcessPoolExecutor, FunctionExecutor) + assert issubclass(ThreadPoolExecutor, FunctionExecutor) + + def test_mode_subclasses_pin_their_lithops_executor(self): + from lithops import executors as native + assert LocalhostExecutor._executor_cls is native.LocalhostExecutor + assert ServerlessExecutor._executor_cls is native.ServerlessExecutor + assert StandaloneExecutor._executor_cls is native.StandaloneExecutor + assert ProcessPoolExecutor._executor_cls is native.FunctionExecutor + + def test_module_reexports_stdlib_constants(self): + assert ALL_COMPLETED is cf.ALL_COMPLETED + assert FIRST_COMPLETED is cf.FIRST_COMPLETED + assert FIRST_EXCEPTION is cf.FIRST_EXCEPTION + + def test_package_reexports_the_public_names(self): + assert lithops.concurrent.ProcessPoolExecutor is ProcessPoolExecutor + assert lithops.concurrent.wait is wait + assert set(lithops.concurrent.__all__) == set( + lithops.concurrent.futures.__all__ + ) + + def test_call_trampoline_applies_args_and_kwargs(self): + def add(x, y, z=0): + return x + y + z + + assert _call(add, (1, 2), {'z': 3}) == 6 + + def test_exception_from_lithops_tuple(self): + lf = FakeLithopsFuture(exc=ValueError('boom')) + assert isinstance(_exception_from_lithops(lf), ValueError) + assert str(_exception_from_lithops(lf)) == 'boom' + + def test_exception_from_lithops_without_a_reason(self): + lf = FakeLithopsFuture(value=1) + lf._state = ResponseFuture.State.Error + exc = _exception_from_lithops(lf) + assert isinstance(exc, Exception) + assert 'without reporting an exception' in str(exc) + + +class TestSubmit: + + def test_submit_dispatches_call_async_with_args_kwargs(self): + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + fut = ex.submit(pow, 2, 8) + func, data, _ = inner.call_async_calls[0] + assert func is _call + assert data == (pow, (2, 8), {}) + assert isinstance(fut, cf.Future) + assert fut.result(timeout=5) == 256 + assert fut.lithops_future is inner._call_result + + def test_submit_passes_keyword_arguments_to_the_callable(self): + inner = FakeInnerExecutor(FakeLithopsFuture(value=9)) + with _adapter(inner) as ex: + ex.submit(pow, 3, exp=2) + assert inner.call_async_calls[0][1] == (pow, (3,), {'exp': 2}) + + def test_submit_falls_back_to_map_without_call_async(self): + inner = FakeInnerExecutor(map_results=[FakeLithopsFuture(value=9)]) + inner.call_async = None + with _adapter(inner) as ex: + assert ex.submit(pow, 3, 2).result(timeout=5) == 9 + assert len(inner.map_calls) == 1 + + def test_submit_returns_a_done_future_with_stats(self): + lf = FakeLithopsFuture(value=4) + lf.stats = {'worker_exec_time': 1.5} + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(pow, 2, 2) + assert fut.result(timeout=5) == 4 + assert fut.done() + assert fut.exception(timeout=5) is None + assert fut.stats['worker_exec_time'] == 1.5 + + def test_submit_propagates_worker_exception(self): + lf = FakeLithopsFuture(exc=ZeroDivisionError('x')) + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(lambda: 1 / 0) + with pytest.raises(ZeroDivisionError, match='x'): + fut.result(timeout=5) + assert isinstance(fut.exception(timeout=5), ZeroDivisionError) + + def test_submitted_future_is_running_and_declines_cancel(self): + lf = FakeLithopsFuture(finished=False) + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(pow, 2, 2) + assert fut.running() + assert fut.cancel() is False + assert not fut.cancelled() + lf.finish(value=4) + assert fut.result(timeout=5) == 4 + + def test_lost_activation_raises_instead_of_returning_none(self): + """A future Lithops marks Unknown is done, but has no result.""" + lf = FakeLithopsFuture(finished=False) + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(pow, 2, 2) + lf.lose() + with pytest.raises(BrokenExecutor, match='lost track'): + fut.result(timeout=5) + + def test_missing_output_surfaces_as_an_error(self): + """result() flips the future to Error when the output never lands.""" + lf = FakeLithopsFuture(value=None) + + def failing_result(throw_except=True, internal_storage=None, + retries=10, wait_dur_sec=1): + lf._state = ResponseFuture.State.Error + return None + + lf.result = failing_result + with _adapter(FakeInnerExecutor(lf)) as ex: + with pytest.raises(Exception, match='without reporting an exception'): + ex.submit(pow, 2, 2).result(timeout=5) + + def test_job_kwargs_are_forwarded(self): + inner = FakeInnerExecutor() + with _adapter( + inner, + runtime_memory=512, + extra_env={'A': '1'}, + execution_timeout=30, + include_modules=['pkg'], + exclude_modules=['pkg.tests'], + ) as ex: + ex.submit(pow, 2, 2) + kwargs = inner.call_async_calls[0][2] + assert kwargs == { + 'runtime_memory': 512, + 'extra_env': {'A': '1'}, + 'timeout': 30, + 'include_modules': ['pkg'], + 'exclude_modules': ['pkg.tests'], + } + + def test_no_job_kwargs_are_sent_when_none_are_set(self): + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + ex.submit(pow, 2, 2) + assert inner.call_async_calls[0][2] == {} + + +class TestMap: + + def test_map_uses_one_lithops_map_and_yields_results_in_order(self): + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + mapped = ex.map(abs, [-1, 2, -3]) + assert not isinstance(mapped, cf.Future) + assert len(inner.map_calls) == 1 + mapped_fn, payloads, _ = inner.map_calls[0] + assert mapped_fn is _call + assert payloads == [ + (abs, (-1,), {}), + (abs, (2,), {}), + (abs, (-3,), {}), + ] + assert list(mapped) == [1, 2, 3] + + def test_map_zips_multiple_iterables(self): + inner = FakeInnerExecutor(map_results=[ + FakeLithopsFuture(value=10), + FakeLithopsFuture(value=12), + ]) + with _adapter(inner) as ex: + results = list(ex.map(simple_map_function, [4, 5], [6, 7])) + assert inner.map_calls[0][1] == [ + (simple_map_function, (4, 6), {}), + (simple_map_function, (5, 7), {}), + ] + assert results == [10, 12] + + def test_map_stops_at_the_shortest_iterable(self): + inner = FakeInnerExecutor(map_results=[FakeLithopsFuture(value=1)]) + with _adapter(inner) as ex: + list(ex.map(simple_map_function, [1, 2, 3], [9])) + assert len(inner.map_calls[0][1]) == 1 + + def test_map_over_nothing_submits_no_job(self): + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + assert list(ex.map(abs, [])) == [] + assert inner.map_calls == [] + + def test_map_leaves_chunksize_to_the_lithops_config_by_default(self): + """ + Passing the standard library default of 1 would silently override a + chunksize set in the Lithops configuration + """ + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + list(ex.map(abs, [-1, 2, -3])) + assert inner.map_calls[0][2]['chunksize'] is None + + def test_map_forwards_an_explicit_chunksize(self): + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + list(ex.map(abs, [-1, 2, -3], chunksize=2)) + assert inner.map_calls[0][2]['chunksize'] == 2 + + def test_map_rejects_non_positive_chunksize(self): + with _adapter() as ex: + with pytest.raises(ValueError, match='chunksize'): + ex.map(abs, [1], chunksize=0) + + def test_map_rejects_negative_buffersize(self): + with _adapter() as ex: + with pytest.raises(ValueError, match='buffersize'): + ex.map(abs, [1], buffersize=-1) + + def test_map_buffersize_submits_in_windows(self): + inner = FakeInnerExecutor() + inner._map_side_effect = [ + [FakeLithopsFuture(value=1)], + [FakeLithopsFuture(value=2)], + [FakeLithopsFuture(value=3)], + ] + with _adapter(inner) as ex: + assert list(ex.map(abs, [-1, -2, -3], buffersize=1)) == [1, 2, 3] + assert len(inner.map_calls) == 3 + + def test_map_is_eager(self): + """Every call is submitted before the iterator is consumed.""" + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + ex.map(abs, [-1, 2, -3]) + assert len(inner.map_calls) == 1 + + def test_map_propagates_the_first_exception_in_order(self): + inner = FakeInnerExecutor(map_results=[ + FakeLithopsFuture(value=0), + FakeLithopsFuture(exc=ValueError('nope')), + FakeLithopsFuture(value=2), + ]) + with _adapter(inner) as ex: + it = ex.map(abs, [0, 1, 2]) + assert next(it) == 0 + with pytest.raises(ValueError, match='nope'): + next(it) + + def test_map_timeout_raises(self): + lf = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor(map_results=[lf]) + ex = _adapter(inner) + try: + it = ex.map(abs, [1], timeout=0.2) + with pytest.raises(TimeoutError): + next(it) + finally: + lf.finish(value=1) + ex.shutdown(wait=True) + + def test_map_timeout_is_not_extended_by_a_slow_download(self): + """ + The download runs off the caller's thread, so a result that is ready + but slow to fetch must not push result(timeout) past its deadline + """ + lf = FakeLithopsFuture(value=1) + gate = lf.block_result() + inner = FakeInnerExecutor(map_results=[lf]) + ex = _adapter(inner) + try: + it = ex.map(abs, [1], timeout=0.3) + start = time.monotonic() + with pytest.raises(TimeoutError): + next(it) + assert time.monotonic() - start < 2 + finally: + gate.set() + ex.shutdown(wait=True) + + +class TestCompletionTracking: + + def test_watcher_completes_a_future_that_finishes_later(self): + lf = FakeLithopsFuture(finished=False) + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(pow, 2, 10) + assert not cf.Future.done(fut) + lf.finish(value=1024) + assert fut.result(timeout=5) == 1024 + + def test_the_job_monitor_state_is_used_instead_of_polling_storage(self): + """ + Lithops' job monitor already keeps every future's state current with + one batched listing per round. Reading each future's status here as + well would be one storage request per future per round + """ + lf = FakeLithopsFuture(finished=False) + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(pow, 2, 10) + time.sleep(0.5) + assert lf._status_calls == 0 + lf.finish(value=1024) + assert fut.result(timeout=5) == 1024 + # One read of the status, once, to resolve the completed call + assert lf._status_calls == 1 + + def test_storage_is_polled_when_there_is_no_job_monitor(self): + lf = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor(lf, job_monitor=False) + with _adapter(inner) as ex: + fut = ex.submit(pow, 2, 10) + lf.finish(value=1024) + assert fut.result(timeout=5) == 1024 + assert lf._status_calls >= 1 + + def test_the_internal_storage_handler_is_reused(self): + """Otherwise every read builds a new client from the config.""" + lf = FakeLithopsFuture(value=7) + inner = FakeInnerExecutor(lf) + with _adapter(inner) as ex: + assert ex.submit(pow, 7, 1).result(timeout=5) == 7 + assert lf._storage_seen + assert all(seen is inner.internal_storage for seen in lf._storage_seen) + + def test_a_dead_job_monitor_is_restarted(self): + """ + The monitor is a daemon that exits once everything it knows about is + done, so a job submitted as it wound down would never be watched + """ + lf = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor(lf) + with _adapter(inner) as ex: + fut = ex.submit(pow, 2, 10) + inner.job_monitor.monitor.alive = False + deadline = time.monotonic() + 5 + while not inner.job_monitor.starts and time.monotonic() < deadline: + time.sleep(0.05) + assert inner.job_monitor.starts == [[lf]] + lf.finish(value=1024) + assert fut.result(timeout=5) == 1024 + + def test_a_monitor_that_never_ran_is_left_alone(self): + """JobMonitor.is_alive() raises when no monitor thread was created.""" + lf = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor(lf) + inner.job_monitor = FakeJobMonitor(started=False) + with _adapter(inner) as ex: + fut = ex.submit(pow, 2, 10) + time.sleep(0.3) + assert inner.job_monitor.starts == [] + lf.finish(value=1024) + assert fut.result(timeout=5) == 1024 + + def test_downloads_do_not_serialize_behind_each_other(self): + """ + A future whose result is slow to fetch must not hold up the ones + behind it, which a single watcher thread doing the downloads would + """ + slow = FakeLithopsFuture(value='slow') + gate = slow.block_result() + quick = FakeLithopsFuture(value='quick') + inner = FakeInnerExecutor(map_results=[slow, quick]) + with _adapter(inner) as ex: + it = ex.map(abs, [1, 2]) + futures = list(ex._pending) + done, _ = wait(futures, timeout=5, return_when=FIRST_COMPLETED) + assert done + gate.set() + assert list(it) == ['slow', 'quick'] + + def test_a_result_is_downloaded_once(self): + lf = FakeLithopsFuture(value=5) + with _adapter(FakeInnerExecutor(lf)) as ex: + fut = ex.submit(abs, -5) + assert [fut.result(timeout=5) for _ in range(3)] == [5, 5, 5] + assert fut.done() + assert lf._result_calls == 1 + + def test_done_does_not_block_on_a_slow_download(self): + lf = FakeLithopsFuture(value=1) + gate = lf.block_result() + ex = _adapter(FakeInnerExecutor(lf)) + try: + fut = ex.submit(abs, -1) + start = time.monotonic() + for _ in range(5): + fut.done() + assert time.monotonic() - start < 1 + finally: + gate.set() + ex.shutdown(wait=True) + + def test_a_dead_watcher_breaks_the_pending_futures(self): + """ + Nothing else resolves futures, so a watcher that died has to fail + them rather than leave every result() blocked for good + """ + def boom(lf): + raise RuntimeError('the watcher blew up') + + lf = FakeLithopsFuture(finished=False) + ex = _adapter(FakeInnerExecutor(lf)) + fut = ex.submit(pow, 2, 2) + ex._is_ready = boom + ex._wake.set() + with pytest.raises(BrokenExecutor, match='blew up'): + fut.result(timeout=5) + with pytest.raises(BrokenExecutor): + ex.submit(pow, 2, 2) + ex.shutdown(wait=False) + + def test_futures_from_many_threads_all_complete(self): + inner = FakeInnerExecutor() + inner._call_results = [FakeLithopsFuture(value=i) for i in range(30)] + results = [] + with _adapter(inner) as ex: + def go(): + results.append(ex.submit(abs, -1).result(timeout=10)) + + threads = [threading.Thread(target=go) for _ in range(30)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=15) + assert sorted(results) == list(range(30)) + + +class TestWaitAndAsCompleted: + + def test_wait_and_as_completed(self): + inner = FakeInnerExecutor() + inner._call_results = [ + FakeLithopsFuture(value=256), + FakeLithopsFuture(value=8), + ] + with _adapter(inner) as ex: + f1 = ex.submit(pow, 2, 8) + f2 = ex.submit(pow, 2, 3) + done, not_done = wait([f1], return_when=ALL_COMPLETED) + assert f1 in done + assert not not_done + assert {f.result() for f in as_completed([f1, f2])} == {256, 8} + + def test_wait_first_completed(self): + pending = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor() + inner._call_results = [FakeLithopsFuture(value='ok'), pending] + with _adapter(inner) as ex: + finished = ex.submit(str, 'ok') + still = ex.submit(str, 'later') + done, not_done = wait( + [finished, still], return_when=FIRST_COMPLETED, timeout=5 + ) + assert finished in done + assert still in not_done + pending.finish(value='later') + assert still.result(timeout=5) == 'later' + + def test_wait_first_exception(self): + inner = FakeInnerExecutor() + inner._call_results = [ + FakeLithopsFuture(exc=RuntimeError('bad')), + FakeLithopsFuture(finished=False), + ] + with _adapter(inner) as ex: + failed = ex.submit(str, 'a') + still = ex.submit(str, 'b') + done, _ = wait( + [failed, still], return_when=FIRST_EXCEPTION, timeout=5 + ) + assert failed in done + inner._call_results = None + still.lithops_future.finish(value='b') + + def test_wait_times_out_on_a_pending_future(self): + lf = FakeLithopsFuture(finished=False) + ex = _adapter(FakeInnerExecutor(lf)) + try: + fut = ex.submit(str, 'x') + done, not_done = wait([fut], timeout=0.2) + assert not done + assert not_done == {fut} + finally: + lf.finish(value='x') + ex.shutdown(wait=True) + + def test_wait_mixes_in_stdlib_futures(self): + with _adapter() as ex, cf.ThreadPoolExecutor(1) as pool: + ours = ex.submit(pow, 2, 8) + theirs = pool.submit(pow, 2, 8) + done, not_done = wait([ours, theirs], timeout=5) + assert done == {ours, theirs} + assert not not_done + + +class TestLifecycle: + + def test_submit_after_shutdown_raises(self): + ex = _adapter() + ex.shutdown(wait=False) + with pytest.raises(RuntimeError, match='shutdown'): + ex.submit(pow, 2, 2) + + def test_context_manager_shuts_down(self): + with _adapter(FakeInnerExecutor()) as ex: + ex.submit(pow, 2, 8) + with pytest.raises(RuntimeError, match='shutdown'): + ex.submit(pow, 2, 2) + + def test_shutdown_waits_for_the_futures_in_flight(self): + lf = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor(lf) + ex = FunctionExecutor(executor=inner) + fut = ex.submit(pow, 2, 2) + threading.Timer(0.2, lambda: lf.finish(value=4)).start() + ex.shutdown(wait=True) + assert cf.Future.done(fut) + assert fut.result() == 4 + + def test_shutdown_is_idempotent(self): + ex = _owning_adapter() + ex.shutdown() + ex.shutdown() + assert ex._inner.exited + + def test_shutdown_without_waiting_still_releases_the_executor(self): + """ + Otherwise the job monitor and invoker threads outlive the executor + """ + lf = FakeLithopsFuture(finished=False) + ex = _owning_adapter(lf) + inner = ex._inner + ex.submit(pow, 2, 2) + ex.shutdown(wait=False) + assert not inner.exited + lf.finish(value=4) + deadline = time.monotonic() + 5 + while not inner.exited and time.monotonic() < deadline: + time.sleep(0.05) + assert inner.exited + + def test_shutdown_of_an_idle_executor_releases_it_at_once(self): + ex = _owning_adapter() + ex.shutdown(wait=False) + assert ex._inner.exited + + def test_owned_executor_is_torn_down(self): + ex = _owning_adapter() + with ex: + ex.submit(pow, 2, 8) + assert ex._inner.exited + assert ex._inner.job_monitor.stopped + + def test_wrapped_executor_is_not_torn_down(self): + inner = FakeInnerExecutor() + with _adapter(inner): + pass + assert not inner.exited + assert not inner.job_monitor.stopped + + def test_owned_executor_cleans_its_temporary_data(self): + ex = _owning_adapter(data_cleaner=True) + with ex: + ex.submit(pow, 2, 8) + assert ex._inner.cleaned == [{'clean_cloudobjects': False}] + + def test_no_cleaning_when_the_data_cleaner_is_off(self): + ex = _owning_adapter(data_cleaner=False) + with ex: + ex.submit(pow, 2, 8) + assert ex._inner.cleaned == [] + + def test_wrapped_executor_data_is_not_cleaned(self): + inner = FakeInnerExecutor(data_cleaner=True) + with _adapter(inner) as ex: + ex.submit(pow, 2, 8) + assert inner.cleaned == [] + + def test_shutdown_cancel_futures_does_not_hang(self): + """ + Lithops cannot recall a dispatched call, so cancel() declines and + shutdown still has to wait the futures out + """ + lf = FakeLithopsFuture(finished=False) + ex = _adapter(FakeInnerExecutor(lf)) + fut = ex.submit(pow, 2, 2) + threading.Timer(0.2, lambda: lf.finish(value=4)).start() + ex.shutdown(wait=True, cancel_futures=True) + assert not fut.cancelled() + assert fut.result() == 4 + + def test_the_watcher_thread_does_not_outlive_the_executor(self): + with _adapter(FakeInnerExecutor()) as ex: + ex.submit(pow, 2, 8) + watcher = ex._watcher + assert not watcher.is_alive() + + +class TestConstruction: + + def test_initializer_is_rejected(self): + with pytest.raises(NotImplementedError, match='initializer'): + FunctionExecutor( + executor=FakeInnerExecutor(), initializer=lambda: None + ) + + def test_retrying_executor_is_rejected(self): + """Its retries come from its own wait(), which nothing here calls.""" + retrying = RetryingFunctionExecutor.__new__(RetryingFunctionExecutor) + with pytest.raises(TypeError, match='RetryingFunctionExecutor'): + FunctionExecutor(executor=retrying) + + def test_stdlib_pool_kwargs_are_accepted_when_wrapping(self): + ex = FunctionExecutor( + executor=FakeInnerExecutor(), + mp_context=object(), + max_tasks_per_child=2, + thread_name_prefix='t', + ) + ex.shutdown(wait=False) + + def test_stdlib_pool_kwargs_never_reach_the_lithops_executor(self): + seen = {} + + class Recording(FunctionExecutor): + _executor_cls = staticmethod( + lambda **kwargs: seen.update(kwargs) or FakeInnerExecutor() + ) + + Recording( + 4, mp_context=object(), max_tasks_per_child=2, + thread_name_prefix='t', backend='localhost', + ).shutdown(wait=False) + assert seen == {'max_workers': 4, 'backend': 'localhost'} + + def test_max_workers_is_ignored_when_wrapping(self): + inner = FakeInnerExecutor() + with FunctionExecutor(8, executor=inner) as ex: + assert ex.lithops_executor is inner + + def test_lithops_executor_is_exposed(self): + inner = FakeInnerExecutor() + with _adapter(inner) as ex: + assert ex.lithops_executor is inner + + +class TestFutureObject: + + def test_a_future_without_an_adapter_behaves_like_the_stdlib_one(self): + fut = Future() + assert fut.lithops_future is None + assert fut.stats == {} + assert not fut.done() + fut.set_result(3) + assert fut.result() == 3 + + def test_stats_come_from_the_lithops_future(self): + lf = FakeLithopsFuture(value=1) + lf.stats = {'worker_exec_time': 2.5} + assert Future(lithops_future=lf).stats == {'worker_exec_time': 2.5} + + +def _owning_adapter(lf=None, data_cleaner=False): + """An adapter that built its own executor, so it owns the teardown.""" + inner = FakeInnerExecutor(lf, data_cleaner=data_cleaner) + + class Owning(FunctionExecutor): + _executor_cls = staticmethod(lambda **kwargs: inner) + + return Owning() + + +def _same_api(executor): + """The interchangeability check from issue 1427.""" + with executor: + future = executor.submit(pow, 323, 1235) + value = future.result(timeout=30) + mapped = list(executor.map(abs, [-1, 2, -3], timeout=30)) + return value, mapped + + +class TestConcurrentFuturesLive: + """Runs real Lithops jobs. Uses the same config as the rest of the suite.""" + + def test_submit_and_map_match_threadpoolexecutor(self): + expected = _same_api(cf.ThreadPoolExecutor(max_workers=2)) + got = _same_api( + FunctionExecutor(config=pytest.lithops_config, log_level=None) + ) + assert got == expected + + def test_submit_keyword_only_and_positional(self): + def greet(name, suffix='!'): + return f'hello {name}{suffix}' + + with FunctionExecutor(config=pytest.lithops_config, log_level=None) as ex: + assert ex.submit(greet, 'lithops', suffix='.').result(timeout=30) == ( + 'hello lithops.' + ) + assert list(ex.map(greet, ['a', 'b'], ['?', '!'])) == [ + 'hello a?', + 'hello b!', + ] + + def test_as_completed_yields_each_future_once(self): + with FunctionExecutor(config=pytest.lithops_config, log_level=None) as ex: + futures = [ex.submit(abs, n) for n in (-2, 0, 5)] + results = [f.result() for f in as_completed(futures, timeout=30)] + assert sorted(results) == [0, 2, 5] + + def test_wait_all_completed(self): + with FunctionExecutor(config=pytest.lithops_config, log_level=None) as ex: + futures = [ex.submit(pow, 2, n) for n in (3, 4)] + done, not_done = wait(futures, timeout=30, return_when=ALL_COMPLETED) + assert not not_done + assert {f.result() for f in done} == {8, 16} + + def test_map_raises_the_callables_exception(self): + def boom(x): + if x: + raise ValueError('nope') + return x + + with FunctionExecutor(config=pytest.lithops_config, log_level=None) as ex: + it = ex.map(boom, [0, 1]) + assert next(it) == 0 + with pytest.raises(ValueError, match='nope'): + next(it) + + def test_future_is_awaitable_via_asyncio_wrap_future(self): + async def run(): + with FunctionExecutor(config=pytest.lithops_config, log_level=None) as ex: + wrapped = asyncio.wrap_future(ex.submit(pow, 2, 8)) + return await wrapped + + assert asyncio.run(run()) == 256 + + def test_the_executor_is_reusable_across_batches(self): + """ + The Lithops job monitor exits once the first batch is done, so a + second one has to bring it back + """ + with FunctionExecutor(config=pytest.lithops_config, log_level=None) as ex: + assert list(ex.map(abs, [-1, -2])) == [1, 2] + time.sleep(1) + assert list(ex.map(abs, [-3, -4])) == [3, 4] + assert ex.submit(abs, -5).result(timeout=30) == 5 + + def test_costs_no_more_storage_reads_than_the_native_api(self): + """ + Lithops' job monitor already tracks every call with one batched + listing per round. Reading each future's status here as well would + put the adapter's cost at one storage request per future per round + """ + from lithops.storage.storage import InternalStorage + + def count(run): + reads = {'n': 0} + original = InternalStorage.get_call_status + + def counting(self, *args, **kwargs): + reads['n'] += 1 + return original(self, *args, **kwargs) + + InternalStorage.get_call_status = counting + try: + run() + finally: + InternalStorage.get_call_status = original + return reads['n'] + + data = list(range(4)) + expected = [x * 2 for x in data] + + def with_adapter(): + with FunctionExecutor( + config=pytest.lithops_config, log_level=None + ) as ex: + assert sorted(ex.map(_sleep_and_double, data)) == expected + + def with_native_api(self=None): + with lithops.FunctionExecutor( + config=pytest.lithops_config, log_level=None + ) as ex: + futures = ex.map(_sleep_and_double, data) + results = ex.get_result(futures, show_progressbar=False) + assert sorted(results) == expected + + native = count(with_native_api) + adapter = count(with_adapter) + # Same order of magnitude, with room for the jitter of two live runs + assert adapter <= native * 1.5 + 20, (adapter, native) + + +def _sleep_and_double(x): + import time + time.sleep(3) + return x * 2 From a945db27346d2548b0e29e4a5628ba07eb87a05e Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sat, 29 Aug 2026 16:13:35 +0200 Subject: [PATCH 2/2] Add tests --- README.md | 2 +- docs/source/api_concurrent.rst | 2 ++ lithops/concurrent/futures.py | 34 +++++++++++++++++------- lithops/tests/test_concurrent_futures.py | 15 ++++++++--- 4 files changed, 38 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index da18d0788..560b7c29a 100644 --- a/README.md +++ b/README.md @@ -148,7 +148,7 @@ Supported backends by platform: ## High-level API -Lithops provides three high-level compute APIs and two high-level storage APIs. +Lithops provides a native compute API, two drop-in replacements for standard Python APIs, and two storage APIs. ### [Core API](docs/source/api_futures.rst) diff --git a/docs/source/api_concurrent.rst b/docs/source/api_concurrent.rst index 34f135cdc..9f8d55964 100644 --- a/docs/source/api_concurrent.rst +++ b/docs/source/api_concurrent.rst @@ -93,6 +93,8 @@ The *API* matches ``concurrent.futures``. The *runtime* is Lithops: ``include_modules``, ``exclude_modules``) are set on the executor. Keyword arguments to ``submit(fn, *args, **kwargs)`` are passed to ``fn``. * Each ``Future`` also exposes ``lithops_future`` and ``stats``. +* A call Lithops loses track of raises ``RuntimeError`` rather than handing + back a silent ``None``. It fails that one future; the executor stays usable. * ``RetryingFunctionExecutor`` cannot be wrapped. Its retries are driven from its own ``wait()``, which this adapter never calls, so wrapping it would quietly give you no retries at all. Wrap the ``FunctionExecutor`` it holds. diff --git a/lithops/concurrent/futures.py b/lithops/concurrent/futures.py index ebafbde4d..e24e9436f 100644 --- a/lithops/concurrent/futures.py +++ b/lithops/concurrent/futures.py @@ -182,9 +182,11 @@ def _failure_message(lf): def _unknown_state_error(lf): + # Not a BrokenExecutor: one call was lost, the executor itself is fine + # and still takes work call_id = getattr(lf, 'call_id', None) where = f' (call {call_id})' if call_id else '' - return BrokenExecutor( + return RuntimeError( f'Lithops lost track of the activation{where}: it is marked done ' 'but never reported a status, so there is no result to read' ) @@ -198,6 +200,26 @@ def _storage_of(executor): return getattr(inner, 'internal_storage', None) +# One entry per (class, method) pair, so the signature of a Lithops future is +# only ever read once however many calls go through it +_TAKES_STORAGE = {} + + +def _accepts_storage(fn): + key = (type(getattr(fn, '__self__', fn)), getattr(fn, '__name__', None)) + cached = _TAKES_STORAGE.get(key) + if cached is None: + try: + params = inspect.signature(fn).parameters + except (TypeError, ValueError): + params = {} + cached = _TAKES_STORAGE[key] = 'internal_storage' in params or any( + param.kind is inspect.Parameter.VAR_KEYWORD + for param in params.values() + ) + return cached + + def _with_storage(fn, storage, **kwargs): """ Calls a Lithops future method, handing it the internal storage handler @@ -205,15 +227,7 @@ def _with_storage(fn, storage, **kwargs): it, and a blanket ``except TypeError`` would also swallow one raised inside the call and run it a second time """ - try: - params = inspect.signature(fn).parameters - except (TypeError, ValueError): - params = {} - accepts_storage = 'internal_storage' in params or any( - param.kind is inspect.Parameter.VAR_KEYWORD - for param in params.values() - ) - if accepts_storage and storage is not None: + if storage is not None and _accepts_storage(fn): kwargs['internal_storage'] = storage return fn(**kwargs) diff --git a/lithops/tests/test_concurrent_futures.py b/lithops/tests/test_concurrent_futures.py index 152f450e4..8554987a1 100644 --- a/lithops/tests/test_concurrent_futures.py +++ b/lithops/tests/test_concurrent_futures.py @@ -334,13 +334,18 @@ def test_submitted_future_is_running_and_declines_cancel(self): assert fut.result(timeout=5) == 4 def test_lost_activation_raises_instead_of_returning_none(self): - """A future Lithops marks Unknown is done, but has no result.""" + """ + A future Lithops marks Unknown is done, but has no result. It is one + lost call, not a dead executor, so it must not raise BrokenExecutor + """ lf = FakeLithopsFuture(finished=False) with _adapter(FakeInnerExecutor(lf)) as ex: fut = ex.submit(pow, 2, 2) lf.lose() - with pytest.raises(BrokenExecutor, match='lost track'): + with pytest.raises(RuntimeError, match='lost track') as raised: fut.result(timeout=5) + assert not isinstance(raised.value, BrokenExecutor) + assert ex.submit(pow, 2, 2) is not None def test_missing_output_surfaces_as_an_error(self): """result() flips the future to Error when the output never lands.""" @@ -1023,8 +1028,10 @@ def with_native_api(self=None): native = count(with_native_api) adapter = count(with_adapter) - # Same order of magnitude, with room for the jitter of two live runs - assert adapter <= native * 1.5 + 20, (adapter, native) + # Generous, on purpose: the regression this guards against is the + # adapter reading every status every round, which is an order of + # magnitude, not the drift between two live runs on a busy machine + assert adapter <= native * 3 + 50, (adapter, native) def _sleep_and_double(x):