diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d9ea8fdd..33a8acee9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,10 @@ ### Fixed +- [Core] Fixed `wait()` on futures another executor invoked, which crashed with an `AttributeError` in `JobMonitor.is_alive()` and, once past it, watched the wrong storage prefix and never returned. +- [Chaining] Fixed pickling a `FuturesList` detaching the list being pickled from its executor. +- [Chaining] Fixed a list or a slice of futures of a previous job not being recognised as a chain, which failed with an argument binding error instead. +- [Chaining] `extra_args` now raises at submit time instead of letting every activation of the chained job fail on a missing argument. - [Localhost] Fixed a deadlock on a `map` after `wait()` and `get_result()`, caused by stale work queue sentinels. - [Localhost] Fixed a partial `clear()` tearing down the consumers, tasks and latches of other jobs. - [Localhost] Fixed a task starting after `stop()`, leaving a process nobody kills. diff --git a/docs/source/api_futures.rst b/docs/source/api_futures.rst index 56f5d0c74..85f0d2c0d 100644 --- a/docs/source/api_futures.rst +++ b/docs/source/api_futures.rst @@ -53,3 +53,11 @@ Futures API Reference :members: :undoc-members: :show-inheritance: + +``map()`` and ``map_reduce()`` return a :class:`~lithops.utils.FuturesList`, which +is a list of futures that can be mapped over again. See +:doc:`Function chaining `. + +.. autoclass:: lithops.utils.FuturesList + :members: map, map_reduce, wait, get_result + :show-inheritance: diff --git a/docs/source/api_multiprocessing.rst b/docs/source/api_multiprocessing.rst index 104334899..6b31838b5 100644 --- a/docs/source/api_multiprocessing.rst +++ b/docs/source/api_multiprocessing.rst @@ -30,8 +30,8 @@ Process and Pool .. code:: python - # from multiprocessing import Pool - from lithops.multiprocessing import Pool + # from multiprocessing import Pool, TimeoutError + from lithops.multiprocessing import Pool, TimeoutError def square(x): return x * x @@ -44,6 +44,64 @@ Process and Pool except TimeoutError: print("Timed out!") +.. note:: ``Process`` and ``Pool`` need no Redis instance. Everything under + `Stateful abstractions`_ does. + +What is not supported +--------------------- + +The API is the standard one, but the runtime is not a local operating system, +so a few things of it have no counterpart: + +.. list-table:: + :header-rows: 1 + + * - Call + - Behaviour + * - ``active_children()``, ``parent_process()`` + - Raise ``NotImplementedError``. There is no process tree to walk + * - ``Process.terminate()``, ``Process.is_alive()``, ``Process.exitcode`` + - Raise ``NotImplementedError``. Lithops cannot recall an activation it + already dispatched, nor report on one + * - ``Pool.imap()``, ``Pool.imap_unordered()`` + - Not lazy: every call is submitted and every result collected before the + first one is yielded, so an endless iterable will not work. The results + always come back in the order of the input + * - ``Pool.join()`` + - Returns as soon as the pool is released; it does not wait for the calls + still in flight. Use the ``AsyncResult`` of each call to wait for it + * - ``Pool(maxtasksperchild=...)``, ``Process.daemon``, ``Process.authkey`` + - Accepted and ignored. Workers are ephemeral, so there is nothing to + recycle, nothing to daemonize and no handshake to authenticate + * - ``RLock`` + - Only re-entrant for the object that took it. A copy of it in another + process, or one restored from a pickle, does not know the lock is held + * - ``Semaphore.acquire()``, ``Lock.acquire()`` + - Take ``block``, but no ``timeout`` + * - ``Condition.wait(timeout)`` + - A wait that timed out leaves its token behind, so the next + ``notify()`` may wake nobody. ``notify_all()`` is not affected + * - ``RawArray('c', ...)`` + - Not implemented. Use ``Array('c', ...)`` + * - ``Process.close()`` + - Releases the executor whatever state the call is in. The standard + library refuses to close a process still running; Lithops cannot tell + without asking storage, and the activation outlives the object anyway + * - ``freeze_support()``, ``allow_connection_pickling()``, + ``set_executable()``, ``set_forkserver_preload()`` + - Accepted and do nothing. There is no re-executed parent, no local + interpreter to point at and no fork server + +Everything else that ``multiprocessing`` exports is here under the same name, +including ``ProcessError``, ``BufferTooShort``, ``TimeoutError``, +``AuthenticationError``, ``get_logger()`` and ``log_to_stderr()``, and +``ThreadPool`` under ``lithops.multiprocessing.pool``. + +.. note:: ``TimeoutError`` is the one of this package, not the builtin of the + same name, exactly as in the standard library. Catch it by importing it:: + + from lithops.multiprocessing import Pool, TimeoutError + Stateful abstractions --------------------- @@ -51,6 +109,12 @@ Lithops also implements all stateful abstractions from Python multiprocessing: Q Since FaaS lacks mechanisms for function-to-function communication, a `Redis `_ database instance is used. +.. note:: Redis is required for **every** shared object: ``Pipe``, ``Queue``, + ``SimpleQueue``, ``JoinableQueue``, ``Lock``, ``RLock``, ``Semaphore``, + ``BoundedSemaphore``, ``Condition``, ``Event``, ``Barrier``, ``Value``, + ``Array`` and ``Manager``. Building any of them without a ``redis`` section + in the configuration raises an error. + .. note:: Both the functions and the Lithops orchestrator (local process) must be able to access the Redis instance. For example, deploying it on your local machine won't work, since the cloud functions won't be able to reach it. The Redis credentials (host, password, etc.) are loaded from the ``redis`` section of the Lithops configuration. @@ -123,8 +187,12 @@ Multiprocessing configuration keys - Environment variables for the processes, passed directly to Lithops FunctionExecutor ``extra_env`` argument - ``{}`` * - EXPORT_EXECUTION_DETAILS - - Calls ``lithops.FunctionExecutor.plot()``, pass a path to store the plots, ``None`` to disable it - - ``None`` + - Calls ``lithops.FunctionExecutor.plot()``, pass a path to store the plots, ``False`` to disable it + - ``False`` + +``lithops.multiprocessing.config.reset()`` puts every parameter back to its +default. The parameters are process-wide, so a library that sets one changes +what every pool of that process sees. \* To use nanomsg for Pipes, you must still deploy a Redis instance (used for the pipe directory). Note that this feature only works in environments where functions can open a port and communicate with each other. diff --git a/docs/source/notebooks/function_chaining.ipynb b/docs/source/notebooks/function_chaining.ipynb index 23916910d..311688f00 100644 --- a/docs/source/notebooks/function_chaining.ipynb +++ b/docs/source/notebooks/function_chaining.ipynb @@ -218,6 +218,73 @@ ] } ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## When the next job is invoked\n", + "\n", + "By default a chained `map()` is invoked straight away, without waiting for the previous one. Each worker of the\n", + "new job then blocks reading the result of its predecessor from storage, which is what keeps the intermediate\n", + "results out of the client. The wait happens before the function starts, so it does not show up in\n", + "`worker_func_exec_time`, but it is worker time you pay for: in a chain of two, where the first function takes 6\n", + "seconds, the second activation starts at the same instant as the first and stays idle for those 6 seconds.\n", + "\n", + "Pass `sync=True` to have the client wait for the previous job before invoking the next one. The chain takes about\n", + "the same wall-clock time, and no worker is billed for waiting:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "import lithops\n", + "\n", + "\n", + "def my_func1(x):\n", + " return x + 2, 5\n", + "\n", + "\n", + "def my_func2(x, y):\n", + " return x + y\n", + "\n", + "\n", + "fexec = lithops.FunctionExecutor()\n", + "res = fexec.map(my_func1, [1, 2, 3]).map(my_func2, sync=True).get_result()\n", + "print(res)" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Invoking without waiting is the better choice when the functions are short and the backend is slow to start a\n", + "worker, since the two jobs overlap. Waiting is the better choice when a stage is long, when the fan-out is wide,\n", + "or on a backend that bills for the whole activation.\n", + "\n", + "## What a chained function receives\n", + "\n", + "A chained function is called with the result of the previous one and nothing else, so `extra_args` cannot be used\n", + "together with chaining and raises a `ValueError`. Return the extra values from the previous function instead, or\n", + "get the results of the chain and start a new job with them.\n", + "\n", + "Any list of futures of a previous job works as the input of the next one, not only the list `map()` returned, so\n", + "a slice or a comprehension chains the same way:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "fexec = lithops.FunctionExecutor()\n", + "futures = fexec.map(my_func1, [1, 2, 3])\n", + "print(fexec.map(my_func2, futures[:2]).get_result())" + ] } ], "metadata": { diff --git a/lithops/executors.py b/lithops/executors.py index 0ec69016d..de6d35f8d 100644 --- a/lithops/executors.py +++ b/lithops/executors.py @@ -286,6 +286,17 @@ def _as_future_list(futures): """Keep list subclasses (including FuturesList) unchanged; wrap a single future.""" return wrap_as_future_list(futures) + def _monitor_of(self, futures): + """ + The job monitor to wait with. This executor's own only tracks the + jobs it invoked itself, since it polls the storage prefix of its own + id, so futures from anywhere else need wait() to start the monitors + that match the executors they belong to + """ + if all(fut.executor_id == self.executor_id for fut in futures): + return self.job_monitor + return None + @staticmethod def _disable_iterdata_output(iterdata): """ @@ -704,7 +715,7 @@ def wait( wait( fs=futures, internal_storage=self.internal_storage, - job_monitor=self.job_monitor, + job_monitor=self._monitor_of(futures), download_results=download_results, throw_except=throw_except, return_when=return_when, diff --git a/lithops/monitor.py b/lithops/monitor.py index de6e50147..e5ce7c393 100644 --- a/lithops/monitor.py +++ b/lithops/monitor.py @@ -657,9 +657,11 @@ def start(self, fs, job_id=None, chunksize=None, generate_tokens=False): def is_alive(self): """ - Tells whether the monitor thread is still running + Tells whether the monitor thread is still running. False when none + was ever started, which is what an executor asked to wait on futures + it did not invoke itself has """ - return self.monitor.is_alive() + return self.monitor is not None and self.monitor.is_alive() def remove(self, fs): """ diff --git a/lithops/multiprocessing/__init__.py b/lithops/multiprocessing/__init__.py index e23b9f35e..0d75b5b85 100644 --- a/lithops/multiprocessing/__init__.py +++ b/lithops/multiprocessing/__init__.py @@ -6,12 +6,22 @@ # from .context import ( - CloudContext, + ProcessError, + BufferTooShort, + TimeoutError, + AuthenticationError, cpu_count, get_context, get_all_start_methods, set_start_method, - get_start_method + get_start_method, + freeze_support, + allow_connection_pickling, + set_executable, + set_forkserver_preload, + get_logger, + log_to_stderr, + reducer ) from .context import CloudContext as DefaultContext from .connection import Pipe @@ -36,6 +46,17 @@ __all__ = [ + 'ProcessError', + 'BufferTooShort', + 'TimeoutError', + 'AuthenticationError', + 'allow_connection_pickling', + 'freeze_support', + 'get_logger', + 'log_to_stderr', + 'reducer', + 'set_executable', + 'set_forkserver_preload', 'cpu_count', 'get_context', 'get_all_start_methods', @@ -66,5 +87,6 @@ 'config' ] - -context = CloudContext() +# `lithops.multiprocessing.context` is left as the module it is, the way +# `multiprocessing.context` is. Binding an instance over it here made every +# `mp.context.` of ported code fail. The class is `DefaultContext` diff --git a/lithops/multiprocessing/config.py b/lithops/multiprocessing/config.py index 719dd208e..0ebcc10ad 100644 --- a/lithops/multiprocessing/config.py +++ b/lithops/multiprocessing/config.py @@ -35,7 +35,15 @@ EXPORT_EXECUTION_DETAILS: False } -_config = _DEFAULT_CONFIG +# A copy: without it every set_parameter() would also rewrite the defaults, +# leaving nothing to fall back to and no way to tell what a fresh process sees +_config = dict(_DEFAULT_CONFIG) + + +def reset(): + """Puts every parameter back to its default value""" + _config.clear() + _config.update(_DEFAULT_CONFIG) def update(config_dic=None, **configurations): diff --git a/lithops/multiprocessing/connection.py b/lithops/multiprocessing/connection.py index 71a54dad0..4f2fc56fd 100644 --- a/lithops/multiprocessing/connection.py +++ b/lithops/multiprocessing/connection.py @@ -10,14 +10,12 @@ # import time -import selectors import threading import random import io import logging import cloudpickle -from multiprocessing.context import BufferTooShort try: import pynng @@ -26,6 +24,7 @@ from . import util from . import config as mp_config +from .errors import BufferTooShort from queue import Queue logger = logging.getLogger(__name__) @@ -319,13 +318,12 @@ def _set_expiry(self, key): self._set_expiry = lambda key: None def _close(self, _close=None): - if hasattr(self, '_pubsub'): - if self._pubsub is not None: - self._pubsub.unsubscribe(self._handle) - # older versions of StrictRedis can't be closed - if hasattr(self, '_client'): - if hasattr(self._client, 'close'): - self._client.close() + # Only the subscription belongs to this connection. The client is the + # one every shared object of the process talks through, so closing it + # here would take the rest of them down along with this connection + if getattr(self, '_pubsub', None) is not None: + self._pubsub.unsubscribe(self._handle) + self._pubsub = None def _listwrite(self, handle, buf): self._set_expiry(handle) @@ -591,6 +589,7 @@ class _RedisListener: def __init__(self, address, family=None, backlog=1): logger.debug('Requested creation of Redis listener for address %s', address) self._address = address + self._family = family self._client = util.get_redis_client() self._connect() @@ -659,15 +658,6 @@ def _RedisClient(address): # Wait # -# poll/select have the advantage of not requiring any extra file -# descriptor, contrarily to epoll/kqueue (also, they require a single -# syscall). -if hasattr(selectors, 'PollSelector'): - _WaitSelector = selectors.PollSelector -else: - _WaitSelector = selectors.SelectSelector - - def wait(object_list, timeout=None): """ Wait till an object in object_list is ready/readable. diff --git a/lithops/multiprocessing/context.py b/lithops/multiprocessing/context.py index b2ebb9e98..3f4103850 100644 --- a/lithops/multiprocessing/context.py +++ b/lithops/multiprocessing/context.py @@ -18,20 +18,64 @@ # Exceptions # -class ProcessError(Exception): - pass +from .errors import ( # noqa: E402 (re-exported where the stdlib keeps them) + ProcessError, + BufferTooShort, + TimeoutError, + AuthenticationError, +) -class BufferTooShort(ProcessError): - pass +# +# Module-level helpers of the standard library that a cloud backend has +# nothing to do. They are here so that code ported from multiprocessing +# imports and calls them without an AttributeError +# + +def freeze_support(): + """ + No-op. The standard library re-executes the parent script in a spawned + child and needs this to stop it recursing; Lithops workers never do + """ + + +def allow_connection_pickling(): + """No-op. Lithops connections are picklable to begin with""" + + +def set_executable(executable): + """No-op. There is no local interpreter to point at""" + +def set_forkserver_preload(module_names): + """No-op. There is no fork server""" -class TimeoutError(ProcessError): - pass +def get_logger(): + """The logger of this package, as multiprocessing.get_logger() is""" + return logging.getLogger(__package__) -class AuthenticationError(ProcessError): - pass + +_log_to_stderr = False + + +def log_to_stderr(level=None): + """ + Sends the log of this package to stderr, and returns it. Idempotent, as + in the standard library: calling it twice does not print every line twice + """ + global _log_to_stderr + package_logger = get_logger() + if not _log_to_stderr: + handler = logging.StreamHandler() + handler.setFormatter( + logging.Formatter('[%(levelname)s/%(processName)s] %(message)s') + ) + package_logger.addHandler(handler) + _log_to_stderr = True + if level is not None: + package_logger.setLevel(level) + return package_logger # @@ -107,7 +151,7 @@ def Queue(self, maxsize=0): def JoinableQueue(self, maxsize=0): """Returns a queue object""" from .queues import JoinableQueue - return JoinableQueue() + return JoinableQueue(maxsize) def SimpleQueue(self): """Returns a queue object""" @@ -137,7 +181,17 @@ def Array(self, typecode_or_type, size_or_initializer, *, lock=True): ctx=self.get_context()) def cpu_count(self): - lithops_config = lithops.config.default_config() + """ + How many function calls can run at once: the workers of the backend + times the processes each of them runs. + + Resolved against the configuration lithops.multiprocessing was given, + not the one this machine happens to have, or a Pool sized from this + is sized against the wrong backend + """ + from . import config as mp_config + config_data = mp_config.get_parameter(mp_config.LITHOPS_CONFIG) or None + lithops_config = lithops.config.default_config(config_data=config_data) backend = lithops_config['lithops']['backend'] max_workers = lithops_config[backend]['max_workers'] worker_processes = lithops_config[backend]['worker_processes'] @@ -173,6 +227,10 @@ def _check_available(self): _default_context = CloudContext() +# multiprocessing.reducer is the reduction module of the default context. +# Nothing here reduces objects that way, so it stands at None +reducer = _default_context.reducer + cpu_count = _default_context.cpu_count get_context = _default_context.get_context get_all_start_methods = _default_context.get_all_start_methods diff --git a/lithops/multiprocessing/errors.py b/lithops/multiprocessing/errors.py new file mode 100644 index 000000000..c5e4c8c33 --- /dev/null +++ b/lithops/multiprocessing/errors.py @@ -0,0 +1,43 @@ +# +# The exception types of the multiprocessing API +# +# multiprocessing/context.py +# +# Copyright (c) 2006-2008, R Oudkerk +# Licensed to PSF under a Contributor Agreement. +# +# Modifications Copyright (c) 2020 Cloudlab URV +# + +# In a module of their own, rather than in context.py where the standard +# library keeps them, because context.py imports pool.py and pool.py needs +# TimeoutError + +__all__ = [ + 'ProcessError', + 'BufferTooShort', + 'TimeoutError', + 'AuthenticationError', +] + + +class ProcessError(Exception): + pass + + +class BufferTooShort(ProcessError): + pass + + +class TimeoutError(ProcessError): + """ + Raised when a wait ran out of time. + + Not the builtin of the same name, which is an OSError: this is the one + the standard library raises, so `except multiprocessing.TimeoutError` + behaves as it does there + """ + + +class AuthenticationError(ProcessError): + pass diff --git a/lithops/multiprocessing/pool.py b/lithops/multiprocessing/pool.py index 3ef3584d7..679cc9528 100644 --- a/lithops/multiprocessing/pool.py +++ b/lithops/multiprocessing/pool.py @@ -12,7 +12,6 @@ # # Imports # -import queue import itertools import logging @@ -20,6 +19,9 @@ from . import util from . import config as mp_config +# Aliased so that the builtin TimeoutError, which is what Lithops raises +# when a wait runs out, stays reachable in this module +from .errors import TimeoutError as ProcessTimeoutError from .process import cloud_process_wrapper, CloudProcess logger = logging.getLogger(__name__) @@ -47,25 +49,21 @@ class Pool(object): """ Class which supports an async version of applying functions to arguments. """ - _wrap_exception = True - Process = CloudProcess def __init__(self, processes=None, initializer=None, initargs=None, maxtasksperchild=None, context=None): if initargs is None: initargs = () - self._taskqueue = queue.Queue() - self._cache = {} + if processes is not None and processes < 1: + raise ValueError("Number of processes must be at least 1") + if initializer is not None and not callable(initializer): + raise TypeError('initializer must be a callable') + self._state = RUN self._maxtasksperchild = maxtasksperchild self._initializer = initializer self._initargs = initargs - self._remote_logger = {} - self._logger_stream = None - - if processes is not None and processes < 1: - raise ValueError("Number of processes must be at least 1") lithops_conf = mp_config.get_parameter(mp_config.LITHOPS_CONFIG) @@ -76,24 +74,24 @@ def __init__(self, processes=None, initializer=None, initargs=None, maxtasksperc self._executor = FunctionExecutor(**lithops_conf) self._processes = self._executor.invoker.max_workers - if initializer is not None and not callable(initializer): - raise TypeError('initializer must be a callable') - self._remote_logger, self._logger_stream = util.setup_log_streaming(self._executor) def apply(self, func, args=(), kwds={}): """ Equivalent of `func(*args, **kwds)`. """ - assert self._state == RUN - if kwds and not args: - args = {} + if self._state != RUN: + raise ValueError("Pool not running") return self.apply_async(func, args, kwds).get() def map(self, func, iterable, chunksize=None): """ Apply `func` to each element in `iterable`, collecting the results in a list that is returned. + + ``chunksize`` is how many items one worker takes, which is what it + means for the standard library too. Left unset, the chunksize of the + Lithops configuration applies. """ return self._map_async(func, iterable, chunksize).get() @@ -112,16 +110,21 @@ def starmap_async(self, func, iterable, chunksize=None, callback=None, error_cal return self._map_async(func, iterable, chunksize=chunksize, callback=callback, error_callback=error_callback, starmap=True) - def imap(self, func, iterable, chunksize=1): + def imap(self, func, iterable, chunksize=None): """ - Equivalent of `map()` -- can be MUCH slower than `Pool.map()`. + Equivalent of `map()`. + + Unlike the standard library, this is not lazy: every call is + submitted and every result collected before the first one is + yielded. An iterator that never ends will not work here. """ res = self.map(func, iterable, chunksize=chunksize) return IMapIterator(res) - def imap_unordered(self, func, iterable, chunksize=1): + def imap_unordered(self, func, iterable, chunksize=None): """ - Like `imap()` method but ordering of results is arbitrary. + Like `imap()`, and like it not lazy. The results come back in the + order of the input, which the standard library does not promise. """ res = self.map(func, iterable, chunksize=chunksize) return IMapIterator(res) @@ -133,8 +136,8 @@ def apply_async(self, func, args=(), kwds={}, callback=None, error_callback=None if self._state != RUN: raise ValueError("Pool not running") - self._remote_logger, stream = util.setup_log_streaming(self._executor) extra_env = mp_config.get_parameter(mp_config.ENV_VARS) + stream = self._logger_stream process_name = '-'.join([self._executor.executor_id, func.__name__]) futures = self._executor.call_async(cloud_process_wrapper, @@ -166,6 +169,8 @@ def _map_async(self, func, iterable, chunksize=None, callback=None, error_callba """ if self._state != RUN: raise ValueError("Pool not running") + if chunksize is not None and chunksize < 1: + raise ValueError("chunksize must be >= 1") if not hasattr(iterable, '__len__'): iterable = list(iterable) @@ -183,6 +188,7 @@ def _map_async(self, func, iterable, chunksize=None, callback=None, error_callba futures = self._executor.map(cloud_process_wrapper, fmt_args, + chunksize=chunksize, extra_args=extra_args, extra_env=extra_env) @@ -201,13 +207,28 @@ def close(self): def terminate(self): logger.debug('terminating pool') self._state = TERMINATE - if self._remote_logger: - self._remote_logger.stop() - self._remote_logger = None + self._release() def join(self): logger.debug('joining pool') - assert self._state in (CLOSE, TERMINATE) + if self._state not in (CLOSE, TERMINATE): + raise ValueError('Pool is still running') + self._release() + + def _release(self): + """ + Stops the log feed and gives the Lithops executor back. Without it + the monitor and invoker threads of the executor outlive the pool + """ + if self._remote_logger is not None: + self._remote_logger.stop() + self._remote_logger = None + executor, self._executor = self._executor, None + if executor is not None: + try: + executor.__exit__(None, None, None) + except Exception: + logger.debug('Error shutting down the Lithops executor', exc_info=True) def __enter__(self): return self @@ -216,6 +237,16 @@ def __exit__(self, exc_type, exc_val, exc_tb): self.terminate() +class ThreadPool(Pool): + """ + The name ``multiprocessing.pool`` uses for its thread-backed pool. + + Provided so that ``from multiprocessing.pool import ThreadPool`` keeps + working after the import is swapped; the tasks still run on Lithops + workers rather than in local threads. + """ + + # # Class whose instances are returned by `Pool.apply_async()` # @@ -232,7 +263,11 @@ def __init__(self, executor, futures, callback, error_callback): self._exception = None def ready(self): - return all(fut.done for fut in self._futures) + # A call whose status has arrived is finished as far as the caller is + # concerned; `done` only turns true once its result was downloaded + return all( + fut.success or fut.done or fut.error for fut in self._futures + ) def successful(self): if not self.ready(): @@ -240,35 +275,43 @@ def successful(self): return not any(fut.error for fut in self._futures) def wait(self, timeout=None): + """ + Waits for the calls, reporting nothing, as in the standard library. + A wait that timed out leaves the result there to be fetched later + """ try: self._executor.wait(self._futures, download_results=False, timeout=timeout) - except Exception as e: - self._exception = e + except Exception: + logger.debug('Timed out waiting for the pool results', exc_info=True) - def get(self, timeout=None): - if self._exception: - raise self._exception + def _get_values(self, timeout=None): + """ + The value of every call, in order. - self._value = self._executor.get_result(self._futures, timeout=timeout) + Read from the futures rather than through get_result(), which unwraps + a lone result depending on what the executor was last asked to do. A + map in between would otherwise change the shape of this result, and a + call that returns a list of its own is indistinguishable either way + """ + try: + self._executor.wait( + self._futures, download_results=True, timeout=timeout + ) + except TimeoutError as exc: + # Lithops reports it as the builtin, which is an OSError and so + # not what `except multiprocessing.TimeoutError` catches + raise ProcessTimeoutError(str(exc)) from exc + values = [fut.result() for fut in self._futures] + util.export_execution_details(self._futures, self._executor) + return values + def get(self, timeout=None): + """The value of the single call this result stands for""" + self._value = self._get_values(timeout)[0] if self._callback is not None: self._callback(self._value) - - util.export_execution_details(self._futures, self._executor) - return self._value - def _set(self, i, success_result): - self._success, self._value = success_result - if self._callback and self._success: - self._callback(self._value) - self._callback = None - if self._error_callback and not self._success: - self._error_callback(self._value) - self._callback = None - # self._event.set() - # del self._cache[self._job] - AsyncResult = ApplyResult # create alias @@ -279,10 +322,12 @@ def _set(self, i, success_result): class MapResult(ApplyResult): - def __init__(self, executor, futures, callback, error_callback): - ApplyResult.__init__(self, executor, futures, callback, error_callback) - - self._value = [None] * len(futures) + def get(self, timeout=None): + """The list of values, one per item of the iterable""" + self._value = self._get_values(timeout) + if self._callback is not None: + self._callback(self._value) + return self._value # diff --git a/lithops/multiprocessing/process.py b/lithops/multiprocessing/process.py index 780c7b1c6..eb907223c 100644 --- a/lithops/multiprocessing/process.py +++ b/lithops/multiprocessing/process.py @@ -42,14 +42,43 @@ # Public functions # +class _CurrentProcess: + """ + What current_process() reports from inside a worker. Only the identity + of the running call: building a CloudProcess here would create a Lithops + executor and a Redis client just to read a name back + """ + + def __init__(self, name, pid): + self.name = name + self._pid = pid + self.daemon = False + self.exitcode = None + + @property + def ident(self): + return self._pid + + pid = ident + + def is_alive(self): + return True + + def __repr__(self): + return '<{}(name={}, pid={})>'.format( + type(self).__name__, self.name, self._pid + ) + + def current_process(): """ Return process object representing the current process """ if is_lithops_worker(): - p = CloudProcess(name=os.environ.get('LITHOPS_MP_WORKER_NAME')) - p._pid = os.environ.get('__LITHOPS_SESSION_ID', '-1') - return p + return _CurrentProcess( + name=os.environ.get('LITHOPS_MP_WORKER_NAME'), + pid=os.environ.get('__LITHOPS_SESSION_ID', '-1'), + ) else: return _mp.current_process() @@ -73,8 +102,9 @@ def parent_process(): # def cloud_process_wrapper(data, func, initializer=None, initargs=(), name=None, log_stream=None, op=None): - # Put worker name in envs to get it from within the function - os.environ['LITHOPS_MP_WORKER_NAME'] = 'test' + # Put the worker name in the environment, which is where current_process() + # reads it back from + os.environ['LITHOPS_MP_WORKER_NAME'] = name or 'CloudProcess' # Setup remote logger if log_stream is not None: @@ -89,18 +119,15 @@ def cloud_process_wrapper(data, func, initializer=None, initargs=(), name=None, try: if op == 'apply': - res = func(*data['args'], **data['kwargs']) + return func(*data['args'], **data['kwargs']) elif op == 'map': - res = func(data,) + return func(data,) elif op == 'starmap': - res = func(*data) + return func(*data) else: - exception = Exception(op) - raise exception - return res + raise ValueError('Unknown operation {}'.format(op)) except Exception as e: # Print exception stack trace to remote logging buffer - exception = e header = "---------- {} at {} ({}) ----------".format(e.__class__.__name__, os.environ.get('LITHOPS_MP_WORKER_NAME'), os.environ.get('__LITHOPS_SESSION_ID')) @@ -108,18 +135,12 @@ def cloud_process_wrapper(data, func, initializer=None, initargs=(), name=None, footer = '-' * len(header) if remote_log_buff: remote_log_buff.write('\n'.join([header, exception_body, footer, ''])) + raise finally: if remote_log_buff: remote_log_buff.flush() remote_log_buff.stop() - if exception: - raise exception - - @property - def __name__(self): - return os.environ.get('LITHOPS_MP_WORKER_NAME') - # # CloudProcess Class @@ -143,12 +164,13 @@ def __init__(self, group=None, target=None, name=None, args=None, kwargs=None, * self._pid = None if daemon is not None: self.daemon = daemon - lithops_config = mp_config.get_parameter(mp_config.LITHOPS_CONFIG) - self._executor = FunctionExecutor(**lithops_config) + # The executor is built by start(): a process that is never started + # should not leave a monitor and an invoker thread behind, and a + # process talks to Lithops rather than to Redis + self._executor = None self._future = None self._sentinel = object() self._remote_logger = None - self._redis = util.get_redis_client() def run(self): """ @@ -164,11 +186,13 @@ def start(self): assert not self._pid, 'cannot start a process twice' assert self._parent_pid == os.getpid(), 'can only start a process object created by current process' + lithops_config = mp_config.get_parameter(mp_config.LITHOPS_CONFIG) + self._executor = FunctionExecutor(**lithops_config) self._remote_logger, stream = util.setup_log_streaming(self._executor) extra_env = mp_config.get_parameter(mp_config.ENV_VARS) - process_name = '-'.join(['CloudProcess', str(next(_process_counter)), self._target.__name__]) + process_name = '-'.join([self._name, self._target.__name__]) self._future = self._executor.call_async(cloud_process_wrapper, {'func': self._target, 'data': { @@ -190,6 +214,30 @@ def terminate(self): """ raise NotImplementedError() + def kill(self): + """ + Terminate process; sends SIGKILL signal or uses TerminateProcess() + """ + raise NotImplementedError() + + def close(self): + """ + Releases the resources of the process, which cannot be used again. + + Unlike the standard library this does not refuse to close a process + still running: Lithops cannot tell whether the activation is over + without asking storage, and the call outlives this object either way + """ + if self._remote_logger is not None: + self._remote_logger.stop() + self._remote_logger = None + executor, self._executor = self._executor, None + if executor is not None: + try: + executor.__exit__(None, None, None) + except Exception: + logger.debug('Error shutting down the Lithops executor', exc_info=True) + def join(self, timeout=None): """ Wait until child process terminates @@ -199,12 +247,13 @@ def join(self, timeout=None): exception = None try: - self._executor.wait(fs=[self._future]) + self._executor.wait(fs=[self._future], timeout=timeout) except Exception as e: exception = e finally: if self._remote_logger: self._remote_logger.stop() + self._remote_logger = None util.export_execution_details([self._future], self._executor) diff --git a/lithops/multiprocessing/queues.py b/lithops/multiprocessing/queues.py index da6703a23..ad467308d 100644 --- a/lithops/multiprocessing/queues.py +++ b/lithops/multiprocessing/queues.py @@ -12,6 +12,7 @@ __all__ = ['Queue', 'SimpleQueue', 'JoinableQueue'] import os +import time import cloudpickle import logging @@ -23,6 +24,9 @@ logger = logging.getLogger(__name__) +# How often a bounded queue re-checks whether room came free +_POLL_SEC = 0.05 + # # Queue type using a pipe, buffer and thread @@ -51,13 +55,6 @@ def __setstate__(self, state): self._writer, self._opid, self._ref) = state self._after_fork() - @property - def _notfull(self): - if self._maxsize > 0: - return self.qsize() < self._maxsize - else: - return True - def _after_fork(self): logger.debug('Queue._after_fork()') self._closed = False @@ -67,12 +64,28 @@ def _after_fork(self): self._poll = self._reader.poll def put(self, obj, block=True, timeout=None): + """ + Puts an object on the queue, waiting for room on a bounded one. + + A full queue raises Full rather than dropping the object, which is a + loss the caller has no way of noticing + """ if self._closed: raise ValueError(f"Queue {self!r} is closed") - if self._notfull: - obj = cloudpickle.dumps(obj) - self._send_bytes(obj) + if self._maxsize > 0: + self._wait_for_room(block, timeout) + + self._send_bytes(cloudpickle.dumps(obj)) + + def _wait_for_room(self, block, timeout): + end = None if timeout is None else time.monotonic() + timeout + while self.qsize() >= self._maxsize: + if not block: + raise Full + if end is not None and time.monotonic() >= end: + raise Full + time.sleep(_POLL_SEC) def get(self, block=True, timeout=None): if block and timeout is None: @@ -95,15 +108,15 @@ def empty(self): def full(self): if self._maxsize > 0: - return self.qsize() < self._maxsize + return self.qsize() >= self._maxsize else: return False def get_nowait(self): - return self.get(False) + return self.get(block=False) def put_nowait(self, obj): - return self.put(obj, False) + return self.put(obj, block=False) def close(self): self._closed = True @@ -164,10 +177,10 @@ def full(self): return False def get_nowait(self): - return self.get() + return self.get(block=False) def put_nowait(self, obj): - return self.put(obj) + return self.put(obj, block=False) def close(self): if not self._closed: @@ -180,8 +193,8 @@ def close(self): # class JoinableQueue(Queue): - def __init__(self): - super().__init__() + def __init__(self, maxsize=0): + super().__init__(maxsize) self._unfinished_tasks = synchronize.Semaphore(0) self._cond = synchronize.Condition() diff --git a/lithops/multiprocessing/sharedctypes.py b/lithops/multiprocessing/sharedctypes.py index 90f09f18c..7b0d7ea0d 100644 --- a/lithops/multiprocessing/sharedctypes.py +++ b/lithops/multiprocessing/sharedctypes.py @@ -32,6 +32,8 @@ class SharedCTypeProxy: def __init__(self, ctype, *args, **kwargs): + # The tail of the MRO: the lock and context a synchronized proxy was + # given have been consumed by now self._typeid = ctype.__name__ self._oid = '{}-{}'.format(self._typeid, util.get_uuid()) self._client = util.get_redis_client() @@ -65,7 +67,7 @@ def get_lock(self): class RawValueProxy(SharedCTypeProxy): def __init__(self, ctype, *args, **kwargs): - super().__init__(ctype=ctype) + super().__init__(ctype=ctype, *args, **kwargs) def __setattr__(self, key, value): if key == 'value': @@ -86,7 +88,7 @@ def __getattr__(self, item): value = cloudpickle.loads(obj) return value else: - super().__getattribute__(item) + return super().__getattribute__(item) class SynchronizedValueProxy(RawValueProxy, SynchronizedSharedCTypeProxy): @@ -99,7 +101,7 @@ def get_obj(self): class RawArrayProxy(SharedCTypeProxy): def __init__(self, ctype, *args, **kwargs): - super().__init__(ctype) + super().__init__(ctype=ctype, *args, **kwargs) self._it = 0 def _append(self, value): @@ -179,11 +181,12 @@ def __getattr__(self, item): if item == 'value': return self[:] else: - super().__getattribute__(item) + return super().__getattribute__(item) def __getitem__(self, i): if isinstance(i, slice): start, stop, step = i.indices(self.__len__()) + stop -= 1 # lrange is inclusive on both ends logger.debug('Requested get string slice from %i to %i', start, stop) objl = self._client.lrange(self._oid, start, stop) self._client.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) @@ -206,7 +209,7 @@ def RawValue(typecode_or_type, initial_value=None): logger.debug('Requested creation of resource RawValue') type_ = typecode_to_type.get(typecode_or_type, typecode_or_type) obj = RawValueProxy(type_) - if initial_value: + if initial_value is not None: obj.value = initial_value return obj @@ -240,7 +243,7 @@ def Value(typecode_or_type, initial_value=None, lock=True, ctx=None): """ logger.debug('Requested creation of resource Value') type_ = typecode_to_type.get(typecode_or_type, typecode_or_type) - obj = SynchronizedValueProxy(type_) + obj = SynchronizedValueProxy(type_, lock=lock if lock is not True else None, ctx=ctx) if initial_value is not None: obj.value = initial_value return obj @@ -252,10 +255,11 @@ def Array(typecode_or_type, size_or_initializer, *, lock=True, ctx=None): """ logger.debug('Requested creation of resource Array') type_ = typecode_to_type.get(typecode_or_type, typecode_or_type) + given_lock = lock if lock is not True else None if type_ is ctypes.c_char: - obj = SynchronizedStringProxy(type_) + obj = SynchronizedStringProxy(type_, lock=given_lock, ctx=ctx) else: - obj = SynchronizedArrayProxy(type_) + obj = SynchronizedArrayProxy(type_, lock=given_lock, ctx=ctx) if isinstance(size_or_initializer, list) or isinstance(size_or_initializer, bytes): obj._extend(size_or_initializer) diff --git a/lithops/multiprocessing/synchronize.py b/lithops/multiprocessing/synchronize.py index 59a5a288b..72b43dc72 100644 --- a/lithops/multiprocessing/synchronize.py +++ b/lithops/multiprocessing/synchronize.py @@ -275,9 +275,13 @@ def clear(self): self._client.set(self._flag_handle, '0') def wait(self, timeout=None): + """ + Waits for the flag and reports it, as in the standard library, where + `if event.wait(timeout):` is how a timeout is told from a set flag + """ with self._cond: logger.debug('Request wait for event %s', self._flag_handle) - self._cond.wait_for(self.is_set, timeout) + return bool(self._cond.wait_for(self.is_set, timeout)) # diff --git a/lithops/multiprocessing/util.py b/lithops/multiprocessing/util.py index e4fa31ac2..1b7347ad8 100644 --- a/lithops/multiprocessing/util.py +++ b/lithops/multiprocessing/util.py @@ -105,10 +105,10 @@ def export_execution_details(futures, lithops_executor): def get_network_ip(): - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) - s.connect(('', 0)) - return s.getsockname()[0] + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + s.connect(('', 0)) + return s.getsockname()[0] # @@ -212,15 +212,15 @@ def setup_log_streaming(executor): class RemoteLogIOBuffer: def __init__(self, stream): - self._feeder_thread = threading self._buff = io.StringIO() self._redis = get_redis_client() self._stream = stream + self._old_stdout = None def write(self, log): self._buff.write(log) - # self.flush() - self._old_stdout.write(log) + if self._old_stdout is not None: + self._old_stdout.write(log) def flush(self): log = self._buff.getvalue() @@ -228,14 +228,14 @@ def flush(self): self._buff = io.StringIO() def start(self): - import sys self._old_stdout = sys.stdout sys.stdout = self logger.debug('Starting remote logging feed to stream %s', self._stream) def stop(self): - import sys - sys.stdout = self._old_stdout + if self._old_stdout is not None: + sys.stdout = self._old_stdout + self._old_stdout = None logger.debug('Stopping remote logging feed to stream %s', self._stream) @@ -260,7 +260,8 @@ def _logger_monitor(self, stream): logger.debug('Logger monitor thread for stream %s finished', stream) def start(self): - # self._logger_thread.daemon = True + # A daemon: a feed nobody stopped must not hold up interpreter exit + self._logger_thread.daemon = True self._enabled = True self._logger_thread.start() diff --git a/lithops/tests/mp_fakeredis.py b/lithops/tests/mp_fakeredis.py new file mode 100644 index 000000000..197b5264e --- /dev/null +++ b/lithops/tests/mp_fakeredis.py @@ -0,0 +1,310 @@ +# +# 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. +# + +""" +In-memory stand-in for the Redis client lithops.multiprocessing runs on. + +Everything in that package -- pipes, queues, locks, shared values -- talks to +Redis, so without this there is nothing to test against short of a server. +Only the commands the package actually issues are implemented, and they store +bytes the way a real server does, since some of the code compares against +byte literals. +""" + +import fnmatch +import threading +import time + + +def _to_bytes(value): + """What the server stores: every value becomes bytes""" + if isinstance(value, bytes): + return value + if isinstance(value, (bytearray, memoryview)): + return bytes(value) + if isinstance(value, str): + return value.encode() + return str(value).encode() + + +def _key(name): + """ + The server does not tell a str key from the same bytes key, and the + package reads key names back out of lists, which returns them as bytes + """ + if isinstance(name, (bytes, bytearray, memoryview)): + return bytes(name).decode() + return name + + +# Every client unpickled from the same server keeps talking to it, the way +# reconnecting to one server does +_SERVERS = {} + + +def _server(server_id): + return _SERVERS[server_id] + + +class FakeRedis: + """ + A single keyspace shared by every client built from the same instance, + the way one server is shared by every process + """ + + def __init__(self): + self.strings = {} + self.lists = {} + self.expiries = {} + self.published = [] + self.closed = False + self.commands = [] + self._cond = threading.Condition() + self._id = len(_SERVERS) + _SERVERS[self._id] = self + + def __reduce__(self): + return _server, (self._id,) + + # -- helpers used by the tests ---------------------------------------- + + def keys(self, pattern='*'): + names = set(self.strings) | set(self.lists) + return sorted(k for k in names if fnmatch.fnmatch(k, pattern)) + + def _record(self, name, *args): + self.commands.append((name,) + args) + + # -- strings ----------------------------------------------------------- + + def set(self, key, value, ex=None): + key = _key(key) + self._record('set', key) + with self._cond: + self.strings[key] = _to_bytes(value) + if ex is not None: + self.expiries[key] = ex + return True + + def get(self, key): + key = _key(key) + self._record('get', key) + return self.strings.get(key) + + def incr(self, key, amount=1): + key = _key(key) + with self._cond: + value = int(self.strings.get(key, b'0')) + amount + self.strings[key] = _to_bytes(value) + return value + + def decr(self, key, amount=1): + return self.incr(key, -amount) + + def delete(self, *keys): + removed = 0 + with self._cond: + for key in map(_key, keys): + removed += self.strings.pop(key, None) is not None + removed += self.lists.pop(key, None) is not None + self.expiries.pop(key, None) + return removed + + def expire(self, key, seconds): + key = _key(key) + self._record('expire', key, seconds) + if key in self.strings or key in self.lists: + self.expiries[key] = seconds + return True + return False + + # -- lists ------------------------------------------------------------- + + def rpush(self, key, *values): + key = _key(key) + with self._cond: + items = self.lists.setdefault(key, []) + items.extend(_to_bytes(value) for value in values) + self._cond.notify_all() + return len(items) + + def lpush(self, key, *values): + key = _key(key) + with self._cond: + items = self.lists.setdefault(key, []) + for value in values: + items.insert(0, _to_bytes(value)) + self._cond.notify_all() + return len(items) + + def lpop(self, key): + key = _key(key) + with self._cond: + items = self.lists.get(key) + if not items: + return None + return items.pop(0) + + def blpop(self, keys, timeout=0): + """Blocks until one of the keys has an element, as the server does""" + if isinstance(keys, (str, bytes)): + keys = [keys] + keys = [_key(key) for key in keys] + end = None if not timeout else time.monotonic() + timeout + with self._cond: + while True: + for key in keys: + items = self.lists.get(key) + if items: + return key, items.pop(0) + remaining = None if end is None else end - time.monotonic() + if remaining is not None and remaining <= 0: + return None + self._cond.wait(remaining) + + def llen(self, key): + key = _key(key) + return len(self.lists.get(key, [])) + + def lrange(self, key, start, end): + key = _key(key) + items = self.lists.get(key, []) + if end == -1: + return items[start:] + return items[start:end + 1] + + def lindex(self, key, index): + key = _key(key) + items = self.lists.get(key, []) + try: + return items[index] + except IndexError: + return None + + def lset(self, key, index, value): + key = _key(key) + items = self.lists.setdefault(key, []) + while len(items) <= index: + items.append(b'') + items[index] = _to_bytes(value) + return True + + # -- pub/sub ----------------------------------------------------------- + + def publish(self, channel, message): + self.published.append((channel, _to_bytes(message))) + return 1 + + def pubsub(self): + return FakePubSub(self) + + # -- scripting --------------------------------------------------------- + + def register_script(self, script): + """ + The only script the package registers is the capped release of + SemLock, reimplemented here rather than running Lua + """ + return FakeScript(self) + + # -- pipelines --------------------------------------------------------- + + def pipeline(self, transaction=True): + return FakePipeline(self) + + # -- connection -------------------------------------------------------- + + def ping(self): + if self.closed: + raise ConnectionError('client is closed') + return True + + def close(self): + self.closed = True + + +class FakeScript: + """What register_script() returns: a callable with a client to detach""" + + registered_client = None + + def __init__(self, server): + self._server = server + + def __call__(self, keys, args, client=None): + server = client if client is not None else self._server + name = _key(keys[0]) + max_value = int(args[0]) + with server._cond: + current = len(server.lists.get(name, [])) + if current >= max_value: + return current + server.lists.setdefault(name, []).append(b'') + server._cond.notify_all() + return current + 1 + + +class FakePubSub: + def __init__(self, server): + self._server = server + self.channels = [] + self.closed = False + self._pending = [] + + def subscribe(self, channel): + self.channels.append(channel) + self._pending.append({'type': 'subscribe', 'channel': channel, 'data': 1}) + + def unsubscribe(self, channel=None): + self.channels = [] + + def get_message(self, ignore_subscribe_messages=False, timeout=0): + while self._pending: + message = self._pending.pop(0) + if ignore_subscribe_messages and message['type'] == 'subscribe': + continue + return message + return None + + def listen(self): + while self._pending: + yield self._pending.pop(0) + + def feed(self, channel, data): + self._pending.append( + {'type': 'message', 'channel': channel, 'data': _to_bytes(data)} + ) + + def close(self): + self.closed = True + + +class FakePipeline: + def __init__(self, server): + self._server = server + self._queued = [] + + def __getattr__(self, name): + command = getattr(self._server, name) + + def queue(*args, **kwargs): + self._queued.append((command, args, kwargs)) + return self + + return queue + + def execute(self): + results = [command(*args, **kwargs) for command, args, kwargs in self._queued] + self._queued = [] + return results diff --git a/lithops/tests/test_executors.py b/lithops/tests/test_executors.py index 4759ce75f..1c5a12d7d 100644 --- a/lithops/tests/test_executors.py +++ b/lithops/tests/test_executors.py @@ -278,6 +278,28 @@ def test_dump_cleaner_data_recreates_missing_dir(self, tmp_path, monkeypatch): class TestWaitAndGetResult: + @patch('lithops.executors.wait') + def test_wait_uses_its_own_monitor_for_its_own_futures(self, mock_wait): + executor = _bare_executor() + executor.wait([FakeFuture(executor_id='sess-0')], show_progressbar=False) + assert mock_wait.call_args.kwargs['job_monitor'] is executor.job_monitor + + @patch('lithops.executors.wait') + def test_wait_on_another_executors_futures_starts_its_own_monitors( + self, mock_wait + ): + """ + This monitor polls the storage prefix of this executor id, so handing + it futures invoked elsewhere would leave them unwatched. wait() has + to start one monitor per executor they belong to instead + """ + executor = _bare_executor() + executor.wait( + [FakeFuture(executor_id='sess-0'), FakeFuture(executor_id='sess-1')], + show_progressbar=False, + ) + assert mock_wait.call_args.kwargs['job_monitor'] is None + @patch('lithops.executors.wait') def test_wait_partitions_by_done_when_downloading_results(self, mock_wait): finished = FakeFuture(done=True, success=True) diff --git a/lithops/tests/test_monitor.py b/lithops/tests/test_monitor.py index f4776fe0e..2e1dfd6a0 100644 --- a/lithops/tests/test_monitor.py +++ b/lithops/tests/test_monitor.py @@ -19,7 +19,6 @@ import time from unittest.mock import MagicMock, patch -import pytest from lithops.monitor import ( LOG_INTERVAL, @@ -248,13 +247,17 @@ def test_stop_joins_a_live_monitor(self): live.stop.assert_called_once() live.join.assert_called_once_with(timeout=5) - def test_is_alive_requires_a_started_monitor(self): + def test_is_alive_without_a_started_monitor(self): + """ + wait() asks this of the monitor of the executor it was called on, + which has no thread of its own until something is invoked through it + """ storage = MagicMock() storage.get_storage_config.return_value = {'monitoring_interval': 2} storage.backend = 'localhost' job_monitor = JobMonitor('sess-0', storage) - with pytest.raises(AttributeError): - job_monitor.is_alive() + assert job_monitor.monitor is None + assert job_monitor.is_alive() is False class TestMonitorHelpers: diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py new file mode 100644 index 000000000..8162fff26 --- /dev/null +++ b/lithops/tests/test_multiprocessing.py @@ -0,0 +1,1200 @@ +# +# 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. +# + +""" +Unit tests for lithops.multiprocessing. + +The package talks to Redis for every shared object and to a Lithops +FunctionExecutor for every process, so both are replaced here: Redis by the +in-memory server in mp_fakeredis, and the executor by a fake that records +what was submitted. Nothing in this file needs a Redis server or a backend. +""" + +import ctypes +import pickle +import threading +import time +import types + +import cloudpickle +import pytest + +from lithops.multiprocessing import config as mp_config +from lithops.multiprocessing import util as mp_util +from lithops.tests.mp_fakeredis import FakeRedis + + +@pytest.fixture(autouse=True) +def mp_globals(): + """ + Gives back the module-wide state of lithops.multiprocessing. + + Its configuration, its Redis client and its Lithops configuration are all + process-wide singletons, so a test that touches one of them decides the + outcome of every test that runs afterwards + """ + saved_config = dict(mp_config._config) + saved_client = mp_util.REDIS_CLIENT + saved_lithops_config = mp_util.LITHOPS_CONFIG + yield + mp_config._config.clear() + mp_config._config.update(saved_config) + mp_util.REDIS_CLIENT = saved_client + mp_util.LITHOPS_CONFIG = saved_lithops_config + + +@pytest.fixture +def redis(): + """Installs the in-memory server as the client the whole package shares""" + server = FakeRedis() + mp_util.REDIS_CLIENT = server + mp_util.LITHOPS_CONFIG = {'redis': {'host': 'localhost'}} + return server + + +class FakeFuture: + def __init__(self, value=None, error=False): + self.executor_id = 'sess-0' + self.job_id = 'A000' + self.call_id = '00000' + self.done = True + self.error = error + self.success = not error + self.ready = True + self.stats = {'worker_exec_time': 0.5} + self._value = value + + def result(self, throw_except=True, internal_storage=None): + return self._value + + +class FakeExecutor: + """Stand-in for lithops.FunctionExecutor, recording what it was given""" + + def __init__(self, **kwargs): + self.kwargs = kwargs + self.executor_id = 'sess-0' + self.invoker = type('I', (), {'max_workers': 7})() + self.call_async_calls = [] + self.map_calls = [] + self.wait_calls = [] + self.get_result_calls = [] + self.exited = False + self.results = None + self._result_index = 0 + self.wait_error = None + + def call_async(self, func, data, **kwargs): + self.call_async_calls.append((func, data, kwargs)) + return FakeFuture(self._next_value()) + + def map(self, func, iterdata, **kwargs): + self.map_calls.append((func, list(iterdata), kwargs)) + return [FakeFuture(self._next_value()) for _ in iterdata] + + def _next_value(self): + """Hands out `results` one call at a time, in order""" + if not self.results: + return None + value = self.results[self._result_index % len(self.results)] + self._result_index += 1 + return value + + def wait(self, fs=None, **kwargs): + self.wait_calls.append((fs, kwargs)) + if self.wait_error is not None: + raise self.wait_error + return list(fs or []), [] + + def get_result(self, fs=None, **kwargs): + self.get_result_calls.append((fs, kwargs)) + if self.results is not None: + return self.results + return [None] * len(fs or []) + + def __exit__(self, exc_type, exc_value, traceback): + self.exited = True + + +@pytest.fixture +def executor(monkeypatch): + """Every FunctionExecutor the package builds becomes the same fake""" + built = [] + + def factory(**kwargs): + made = FakeExecutor(**kwargs) + built.append(made) + return made + + monkeypatch.setattr('lithops.multiprocessing.pool.FunctionExecutor', factory) + monkeypatch.setattr('lithops.multiprocessing.process.FunctionExecutor', factory) + return built + + +class TestConfig: + + def test_defaults_are_readable(self): + assert mp_config.get_parameter(mp_config.STREAM_STDOUT) is False + assert mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME) == 3600 + assert mp_config.get_parameter(mp_config.PIPE_CONNECTION_TYPE) == 'redislist' + + def test_set_parameter_rejects_unknown_keys(self): + with pytest.raises(KeyError): + mp_config.set_parameter('NOT_A_PARAMETER', 1) + + def test_update_accepts_a_dict_and_keywords(self): + mp_config.update({mp_config.STREAM_STDOUT: True}, REDIS_EXPIRY_TIME=60) + assert mp_config.get_parameter(mp_config.STREAM_STDOUT) is True + assert mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME) == 60 + + def test_setting_a_parameter_leaves_the_defaults_alone(self): + """ + Otherwise there is no pristine default left to fall back to, and the + table in the documentation stops describing what a fresh process sees + """ + mp_config.set_parameter(mp_config.REDIS_EXPIRY_TIME, 11) + assert mp_config._DEFAULT_CONFIG[mp_config.REDIS_EXPIRY_TIME] == 3600 + + +class TestUtil: + + def test_uuid_length(self): + assert len(mp_util.get_uuid()) == 12 + assert len(mp_util.get_uuid(6)) == 6 + + def test_redis_client_requires_a_redis_section(self, monkeypatch): + monkeypatch.setattr(mp_util, 'REDIS_CLIENT', None) + monkeypatch.setattr(mp_util, 'LITHOPS_CONFIG', {'lithops': {}}) + with pytest.raises(Exception, match='Redis section'): + mp_util.get_redis_client() + + def test_redis_client_is_reused(self, redis): + assert mp_util.get_redis_client() is redis + + def test_picklable_redis_survives_a_round_trip(self): + client = mp_util.PicklableRedis(host='h', port=6379) + restored = pickle.loads(pickle.dumps(client)) + assert restored._kwargs == {'host': 'h', 'port': 6379} + + def test_make_stateless_script_detaches_the_client(self, redis): + script = redis.register_script('return 1') + script.registered_client = redis + assert mp_util.make_stateless_script(script).registered_client is None + + def test_log_streaming_is_off_by_default(self): + assert mp_util.setup_log_streaming(FakeExecutor()) == (None, None) + + def test_export_execution_details_is_off_by_default(self): + # Would raise if it tried to plot the fake futures + mp_util.export_execution_details([FakeFuture()], FakeExecutor()) + + +class TestRemoteReference: + + def test_managed_reference_does_not_count(self, redis): + ref = mp_util.RemoteReference('key-1', managed=True, client=redis) + assert ref.managed is True + assert ref.incref() is None + assert ref.decref() is None + + def test_unmanaged_reference_counts_up_and_down(self, redis): + ref = mp_util.RemoteReference('key-1', client=redis) + assert ref.incref() == 1 + assert ref.incref() == 2 + assert ref.decref() == 1 + + def test_the_counter_key_is_collected_with_the_referenced_ones(self, redis): + ref = mp_util.RemoteReference(['key-1', 'key-2'], client=redis) + redis.set('key-1', 'a') + redis.set('key-2', 'b') + ref.collect() + assert redis.keys('key-*') == [] + + def test_a_string_reference_is_taken_as_one_key(self, redis): + ref = mp_util.RemoteReference('key-1', client=redis) + assert ref._referenced == ['key-1', 'ref-key-1'] + + def test_a_reference_must_be_a_key_or_a_list_of_keys(self, redis): + with pytest.raises(TypeError, match='referenced must be'): + mp_util.RemoteReference(42, client=redis) + + +class TestContext: + + def test_the_default_context_exposes_the_process_and_pool(self): + from lithops.multiprocessing import process, pool + from lithops.multiprocessing.context import _default_context + assert _default_context.Process is process.CloudProcess + assert _default_context.Pool is pool.Pool + + def test_get_context_accepts_the_standard_methods(self): + from lithops.multiprocessing.context import get_context, _default_context + for method in ('spawn', 'fork', 'forkserver', 'cloud'): + assert get_context(method) is _default_context + + def test_get_context_rejects_an_unknown_method(self): + from lithops.multiprocessing.context import get_context + with pytest.raises(ValueError, match='cannot find context'): + get_context('threads') + + def test_start_method_is_always_cloud(self): + from lithops.multiprocessing.context import ( + get_start_method, get_all_start_methods + ) + assert get_start_method() == 'cloud' + assert 'cloud' in get_all_start_methods() + + def test_cpu_count_multiplies_workers_by_processes(self, monkeypatch): + from lithops.multiprocessing.context import cpu_count + monkeypatch.setattr( + 'lithops.config.default_config', + lambda *a, **kw: { + 'lithops': {'backend': 'aws_lambda'}, + 'aws_lambda': {'max_workers': 100, 'worker_processes': 2}, + }, + ) + assert cpu_count() == 200 + + def test_cpu_count_uses_the_configured_lithops_parameters(self, monkeypatch): + """ + Otherwise a Pool sized from cpu_count() is sized against whatever the + machine happens to have configured, not against what the caller set + """ + from lithops.multiprocessing.context import cpu_count + seen = {} + + def fake_default_config(config_data=None, **kwargs): + seen['config_data'] = config_data + return { + 'lithops': {'backend': 'localhost'}, + 'localhost': {'max_workers': 4, 'worker_processes': 1}, + } + + monkeypatch.setattr('lithops.config.default_config', fake_default_config) + mp_config.set_parameter( + mp_config.LITHOPS_CONFIG, {'lithops': {'backend': 'localhost'}} + ) + cpu_count() + assert seen['config_data'] == {'lithops': {'backend': 'localhost'}} + + +class TestPool: + + def test_processes_becomes_max_workers(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=3) + assert pool._processes == 3 + assert executor[0].kwargs['max_workers'] == 3 + + def test_without_processes_the_backend_decides(self, executor): + from lithops.multiprocessing import Pool + assert Pool()._processes == 7 + assert 'max_workers' not in executor[0].kwargs + + def test_the_lithops_config_reaches_the_executor(self, executor): + from lithops.multiprocessing import Pool + mp_config.set_parameter( + mp_config.LITHOPS_CONFIG, {'backend': 'localhost'} + ) + Pool() + assert executor[0].kwargs['backend'] == 'localhost' + + def test_zero_processes_is_rejected(self, executor): + from lithops.multiprocessing import Pool + with pytest.raises(ValueError, match='at least 1'): + Pool(processes=0) + + def test_a_non_callable_initializer_is_rejected(self, executor): + from lithops.multiprocessing import Pool + with pytest.raises(TypeError, match='must be a callable'): + Pool(initializer='nope') + + def test_apply_async_submits_one_call(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + pool.apply_async(pow, (2, 8)) + _, data, _ = executor[0].call_async_calls[0] + assert data['op'] == 'apply' + assert data['data'] == {'args': (2, 8), 'kwargs': {}} + assert data['func'] is pow + + def test_apply_returns_the_value_not_a_list(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [256] + assert pool.apply(pow, (2, 8)) == 256 + + def test_map_submits_one_lithops_map(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [1, 4, 9] + assert pool.map(abs, [1, 2, 3]) == [1, 4, 9] + func, iterdata, kwargs = executor[0].map_calls[0] + assert iterdata == [(1,), (2,), (3,)] + assert kwargs['extra_args'][0] is abs + assert kwargs['extra_args'][-1] == 'map' + + def test_starmap_marks_the_operation(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [3] + assert pool.starmap(pow, [(1, 2)]) == [3] + assert executor[0].map_calls[0][2]['extra_args'][-1] == 'starmap' + + def test_map_accepts_a_generator(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [1, 2] + pool.map(abs, (x for x in (1, 2))) + assert executor[0].map_calls[0][1] == [(1,), (2,)] + + def test_map_chunksize_reaches_lithops(self, executor): + """ + Lithops takes a chunksize of its own, meaning the same thing: how + many items one worker takes. Accepting it and dropping it leaves the + caller thinking they tuned something + """ + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [1, 2, 3, 4] + pool.map(abs, [1, 2, 3, 4], chunksize=2) + assert executor[0].map_calls[0][2].get('chunksize') == 2 + + def test_map_without_chunksize_leaves_it_to_the_lithops_config(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [1] + pool.map(abs, [1]) + assert executor[0].map_calls[0][2].get('chunksize') is None + + def test_env_vars_are_forwarded(self, executor): + from lithops.multiprocessing import Pool + mp_config.set_parameter(mp_config.ENV_VARS, {'A': '1'}) + pool = Pool(processes=1) + pool.apply_async(abs, (1,)) + assert executor[0].call_async_calls[0][2]['extra_env'] == {'A': '1'} + + def test_a_closed_pool_takes_no_more_work(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + pool.close() + with pytest.raises(ValueError, match='not running'): + pool.apply_async(abs, (1,)) + with pytest.raises(ValueError, match='not running'): + pool.map_async(abs, [1]) + + def test_a_pool_cannot_be_pickled(self, executor): + from lithops.multiprocessing import Pool + with pytest.raises(NotImplementedError, match='cannot be passed'): + pickle.dumps(Pool(processes=1)) + + def test_join_requires_close_or_terminate(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + with pytest.raises(ValueError, match='still running'): + pool.join() + pool.close() + pool.join() + + def test_the_context_manager_terminates(self, executor): + from lithops.multiprocessing import Pool, pool as pool_module + with Pool(processes=1) as pool: + pass + assert pool._state == pool_module.TERMINATE + + def test_a_pool_gives_its_executor_back(self, executor): + """Otherwise its job monitor and invoker threads outlive the pool""" + from lithops.multiprocessing import Pool + with Pool(processes=1): + pass + assert executor[0].exited + + +class TestApplyResult: + + def _result(self, executor_fake, futures=None): + from lithops.multiprocessing.pool import ApplyResult + return ApplyResult(executor_fake, futures or [FakeFuture()], None, None) + + def _map_result(self, executor_fake, futures): + from lithops.multiprocessing.pool import MapResult + return MapResult(executor_fake, futures, None, None) + + def test_ready_and_successful(self, executor): + result = self._result(FakeExecutor()) + assert result.ready() is True + assert result.successful() is True + + def test_successful_before_ready_raises(self, executor): + pending = FakeFuture() + pending.done = pending.ready = pending.success = False + result = self._result(FakeExecutor(), [pending]) + with pytest.raises(ValueError, match='not ready'): + result.successful() + + def test_ready_once_the_call_reported_back(self, executor): + """ + A future whose status has arrived but whose result nobody downloaded + yet is finished as far as the caller is concerned + """ + fake = FakeExecutor() + finished = FakeFuture() + finished.done = False + finished.success = True + result = self._result(fake, [finished]) + assert result.ready() is True + + def test_successful_is_false_when_a_call_failed(self, executor): + result = self._result(FakeExecutor(), [FakeFuture(error=True)]) + assert result.successful() is False + + def test_get_calls_back_with_the_value(self, executor): + from lithops.multiprocessing.pool import ApplyResult + seen = [] + result = ApplyResult(FakeExecutor(), [FakeFuture(5)], seen.append, None) + assert result.get() == 5 + assert seen == [5] + + def test_map_result_keeps_the_whole_list(self, executor): + result = self._map_result( + FakeExecutor(), [FakeFuture(n) for n in (1, 2, 3)] + ) + assert result.get() == [1, 2, 3] + + def test_a_single_call_returning_a_list_is_not_unwrapped(self, executor): + """ + What the executor's get_result() cannot tell apart: one result that + happens to be a list, and a list of results + """ + result = self._result(FakeExecutor(), [FakeFuture([1, 2, 3])]) + assert result.get() == [1, 2, 3] + + def test_the_shape_of_a_result_does_not_depend_on_later_calls(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [42] + applied = pool.apply_async(abs, (-42,)) + pool.map_async(abs, [1, 2]) + assert applied.get() == 42 + + def test_wait_forwards_the_timeout(self, executor): + result = self._result(FakeExecutor()) + result.wait(timeout=5) + assert result._executor.wait_calls[0][1]['timeout'] == 5 + + def test_a_timeout_raises_the_multiprocessing_error(self, executor): + """ + Lithops reports a timeout as the builtin, which is an OSError and so + not what `except multiprocessing.TimeoutError` catches + """ + from lithops.multiprocessing import TimeoutError as MpTimeoutError + fake = FakeExecutor() + fake.wait_error = TimeoutError('too slow') + result = self._result(fake, [FakeFuture(1)]) + with pytest.raises(MpTimeoutError): + result.get(timeout=0.01) + + def test_a_wait_that_timed_out_does_not_poison_get(self, executor): + """ + wait() reports nothing in the standard library; the result is still + there to be fetched once it arrives + """ + fake = FakeExecutor() + fake.wait_error = TimeoutError('too slow') + result = self._result(fake, [FakeFuture(7)]) + result.wait(timeout=0.01) + fake.wait_error = None + assert result.get() == 7 + + +class TestIMapIterator: + + def test_iterates_over_the_results(self): + from lithops.multiprocessing.pool import IMapIterator + it = IMapIterator([1, 2]) + assert [next(it), it.next()] == [1, 2] + with pytest.raises(StopIteration): + next(it) + + def test_imap_yields_the_same_results_as_map(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [1, 2, 3] + assert list(pool.imap(abs, [1, 2, 3])) == [1, 2, 3] + + def test_imap_unordered_yields_every_result(self, executor): + from lithops.multiprocessing import Pool + pool = Pool(processes=1) + executor[0].results = [1, 2, 3] + assert sorted(pool.imap_unordered(abs, [1, 2, 3])) == [1, 2, 3] + + +def _double(x): + return x * 2 + + +def _boom(x): + raise ValueError('boom') + + +class TestCloudProcessWrapper: + + def test_apply_calls_with_args_and_kwargs(self): + from lithops.multiprocessing.process import cloud_process_wrapper + assert cloud_process_wrapper( + {'args': (2,), 'kwargs': {}}, _double, op='apply' + ) == 4 + + def test_map_passes_the_single_item(self): + from lithops.multiprocessing.process import cloud_process_wrapper + assert cloud_process_wrapper(3, _double, op='map') == 6 + + def test_starmap_unpacks_the_item(self): + from lithops.multiprocessing.process import cloud_process_wrapper + assert cloud_process_wrapper((2, 8), pow, op='starmap') == 256 + + def test_an_unknown_operation_raises(self): + from lithops.multiprocessing.process import cloud_process_wrapper + with pytest.raises(Exception, match='nonsense'): + cloud_process_wrapper(1, _double, op='nonsense') + + def test_the_function_exception_reaches_the_caller(self): + from lithops.multiprocessing.process import cloud_process_wrapper + with pytest.raises(ValueError, match='boom'): + cloud_process_wrapper(1, _boom, op='map') + + def test_the_initializer_runs_before_the_function(self): + from lithops.multiprocessing.process import cloud_process_wrapper + calls = [] + cloud_process_wrapper( + 1, _double, initializer=calls.append, initargs=('init',), op='map' + ) + assert calls == ['init'] + + def test_the_worker_name_is_the_one_that_was_given(self): + """ + current_process().name reads it back out of the environment, so a + hardcoded one makes every process report the same name + """ + import os + from lithops.multiprocessing.process import cloud_process_wrapper + cloud_process_wrapper(1, _double, name='CloudProcess-7', op='map') + assert os.environ['LITHOPS_MP_WORKER_NAME'] == 'CloudProcess-7' + + +class TestCloudProcess: + + def test_start_submits_the_target(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, args=(4,)) + proc.start() + _, data, _ = executor[0].call_async_calls[0] + assert data['func'] is _double + assert data['data'] == {'args': (4,), 'kwargs': {}} + assert proc.pid == 'sess-0/A000/00000' + + def test_a_process_cannot_be_started_twice(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, args=(1,)) + proc.start() + with pytest.raises(AssertionError, match='twice'): + proc.start() + + def test_run_calls_the_target_in_process(self, executor, redis): + from lithops.multiprocessing import Process + seen = [] + Process(target=seen.append, args=('x',)).run() + assert seen == ['x'] + + def test_the_name_defaults_and_can_be_set(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, name='worker-1') + assert proc.name == 'worker-1' + proc.name = 'worker-2' + assert proc.name == 'worker-2' + with pytest.raises(AssertionError, match='must be a string'): + proc.name = 7 + + def test_an_unnamed_process_gets_a_generated_name(self, executor, redis): + from lithops.multiprocessing import Process + assert Process(target=_double).name.startswith('CloudProcess-') + + def test_the_daemon_flag_round_trips(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, daemon=True) + assert proc.daemon is True + assert Process(target=_double).daemon is False + + def test_grouping_is_rejected(self, executor, redis): + from lithops.multiprocessing import Process + with pytest.raises(AssertionError, match='grouping'): + Process(group='g', target=_double) + + def test_join_before_start_is_rejected(self, executor, redis): + from lithops.multiprocessing import Process + with pytest.raises(AssertionError, match='started process'): + Process(target=_double).join() + + def test_join_waits_for_the_call(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, args=(1,)) + proc.start() + proc.join() + assert executor[0].wait_calls + + def test_join_forwards_the_timeout(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, args=(1,)) + proc.start() + proc.join(timeout=5) + assert executor[0].wait_calls[0][1].get('timeout') == 5 + + def test_the_unsupported_api_says_so(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double) + for call in (proc.terminate, proc.kill, proc.is_alive): + with pytest.raises(NotImplementedError): + call() + with pytest.raises(NotImplementedError): + proc.exitcode + + def test_close_gives_the_executor_back(self, executor, redis): + from lithops.multiprocessing import Process + proc = Process(target=_double, args=(1,)) + proc.start() + proc.close() + assert executor[0].exited + + def test_creating_a_process_does_not_need_redis(self, executor, monkeypatch): + """ + A process talks to Lithops, not to Redis. Reaching for a client at + construction makes Redis a requirement of the plain Process API + """ + def no_redis(*args, **kwargs): + raise AssertionError('should not build a Redis client') + + monkeypatch.setattr(mp_util, 'get_redis_client', no_redis) + from lithops.multiprocessing import Process + Process(target=_double, args=(1,)) + + +class TestQueue: + + def _queue(self, maxsize=0): + from lithops.multiprocessing import Queue + return Queue(maxsize) + + def test_put_and_get_round_trip(self, redis): + queue = self._queue() + queue.put({'a': 1}) + assert queue.get() == {'a': 1} + + def test_qsize_and_empty(self, redis): + queue = self._queue() + assert queue.empty() is True + queue.put(1) + assert queue.qsize() == 1 + assert queue.empty() is False + + def test_full_reports_whether_the_queue_is_full(self, redis): + queue = self._queue(maxsize=1) + assert queue.full() is False + queue.put(1) + assert queue.full() is True + + def test_an_unbounded_queue_is_never_full(self, redis): + queue = self._queue() + queue.put(1) + assert queue.full() is False + + def test_put_nowait_on_a_full_queue_raises(self, redis): + """Dropping the item on the floor loses data with no way to tell""" + from queue import Full + queue = self._queue(maxsize=1) + queue.put(1) + with pytest.raises(Full): + queue.put_nowait(2) + assert queue.qsize() == 1 + + def test_get_nowait_on_an_empty_queue_raises(self, redis): + from queue import Empty + with pytest.raises(Empty): + self._queue().get_nowait() + + def test_get_with_a_timeout_raises_when_nothing_arrives(self, redis): + from queue import Empty + queue = self._queue() + started = time.monotonic() + with pytest.raises(Empty): + queue.get(timeout=0.2) + assert time.monotonic() - started < 2 + + def test_put_on_a_closed_queue_raises(self, redis): + queue = self._queue() + queue.close() + with pytest.raises(ValueError, match='closed'): + queue.put(1) + + def test_a_queue_survives_being_pickled(self, redis): + """It travels to the worker inside the job payload""" + queue = self._queue(maxsize=3) + restored = pickle.loads(pickle.dumps(queue)) + restored.put('from the worker') + assert queue.get() == 'from the worker' + assert restored._maxsize == 3 + + +class TestSimpleQueue: + + def test_put_and_get_round_trip(self, redis): + from lithops.multiprocessing import SimpleQueue + queue = SimpleQueue() + queue.put('x') + assert queue.get() == 'x' + assert queue.full() is False + + def test_get_nowait_does_not_block_on_an_empty_queue(self, redis): + from queue import Empty + from lithops.multiprocessing import SimpleQueue + with pytest.raises(Empty): + SimpleQueue().get_nowait() + + def test_put_on_a_closed_queue_raises(self, redis): + from lithops.multiprocessing import SimpleQueue + queue = SimpleQueue() + queue.close() + with pytest.raises(AssertionError): + queue.put('x') + + +class TestJoinableQueue: + + def test_task_done_counts_down(self, redis): + from lithops.multiprocessing import JoinableQueue + queue = JoinableQueue() + queue.put(1) + assert queue._unfinished_tasks.get_value() == 1 + queue.task_done() + assert queue._unfinished_tasks.get_value() == 0 + + def test_too_many_task_done_raises(self, redis): + from lithops.multiprocessing import JoinableQueue + queue = JoinableQueue() + with pytest.raises(ValueError, match='too many times'): + queue.task_done() + + def test_join_returns_when_nothing_is_outstanding(self, redis): + from lithops.multiprocessing import JoinableQueue + queue = JoinableQueue() + queue.put(1) + queue.get() + queue.task_done() + queue.join() + + def test_maxsize_is_honoured(self, redis): + from lithops.multiprocessing.context import _default_context + queue = _default_context.JoinableQueue(2) + assert queue._maxsize == 2 + + +class TestConnection: + + def test_handle_pairs_are_two_ends_of_one_id(self): + from lithops.multiprocessing import connection + a, b = connection.get_handle_pair(connection.REDIS_LIST_CONN) + assert connection.get_subhandle(a) == b + assert connection.get_subhandle(b) == a + + def test_an_unknown_connection_type_is_rejected(self): + from lithops.multiprocessing import connection + with pytest.raises(Exception, match='Unknown connection type'): + connection.get_handle_pair('carrier-pigeon') + + def test_a_bad_handle_prefix_is_rejected(self): + from lithops.multiprocessing import connection + with pytest.raises(ValueError, match='bad handle prefix'): + connection.get_subhandle('nonsense-1234') + + def test_a_pipe_carries_objects_both_ways(self, redis): + from lithops.multiprocessing import Pipe + left, right = Pipe() + left.send({'a': 1}) + assert right.recv() == {'a': 1} + right.send('back') + assert left.recv() == 'back' + + def test_a_simplex_pipe_is_one_way(self, redis): + from lithops.multiprocessing import Pipe + reader, writer = Pipe(duplex=False) + assert reader.readable and not reader.writable + assert writer.writable and not writer.readable + with pytest.raises(OSError, match='read-only'): + reader.send('x') + with pytest.raises(OSError, match='write-only'): + writer.recv() + + def test_send_bytes_honours_offset_and_size(self, redis): + from lithops.multiprocessing import Pipe + left, right = Pipe() + left.send_bytes(b'0123456789', offset=2, size=3) + assert right.recv_bytes() == b'234' + + def test_send_bytes_validates_its_range(self, redis): + from lithops.multiprocessing import Pipe + left, _ = Pipe() + with pytest.raises(ValueError, match='offset is negative'): + left.send_bytes(b'abc', offset=-1) + with pytest.raises(ValueError, match='buffer length < offset'): + left.send_bytes(b'abc', offset=9) + with pytest.raises(ValueError, match='size is negative'): + left.send_bytes(b'abc', size=-1) + + def test_poll_reports_whether_anything_is_waiting(self, redis): + from lithops.multiprocessing import Pipe + left, right = Pipe() + assert right.poll() is False + left.send('x') + assert right.poll() is True + + def test_a_closed_connection_refuses_to_work(self, redis): + from lithops.multiprocessing import Pipe + left, _ = Pipe() + left.close() + assert left.closed is True + with pytest.raises(OSError, match='handle is closed'): + left.send('x') + + def test_a_connection_survives_being_pickled(self, redis): + from lithops.multiprocessing import Pipe + left, right = Pipe() + restored = pickle.loads(pickle.dumps(right)) + left.send('through the pipe') + assert restored.recv() == 'through the pipe' + + def test_closing_one_end_leaves_the_shared_client_usable(self, redis): + """ + The Redis client is a process-wide singleton, so closing it with one + connection takes every other shared object down with it + """ + from lithops.multiprocessing import Pipe + left, right = Pipe() + left.close() + assert redis.closed is False + other_left, other_right = Pipe() + other_left.send('still working') + assert other_right.recv() == 'still working' + + +class TestSemLock: + + def test_acquire_and_release(self, redis): + from lithops.multiprocessing import Lock + lock = Lock() + assert lock.get_value() == 1 + assert lock.acquire() is True + assert lock.get_value() == 0 + lock.release() + assert lock.get_value() == 1 + + def test_a_non_blocking_acquire_fails_when_taken(self, redis): + from lithops.multiprocessing import Lock + lock = Lock() + lock.acquire() + assert lock.acquire(block=False) is False + + def test_the_context_manager_acquires_and_releases(self, redis): + from lithops.multiprocessing import Lock + lock = Lock() + with lock: + assert lock.get_value() == 0 + assert lock.get_value() == 1 + + def test_a_bounded_semaphore_does_not_go_over_its_value(self, redis): + from lithops.multiprocessing import BoundedSemaphore + sem = BoundedSemaphore(2) + sem.release() + assert sem.get_value() == 2 + + def test_a_semaphore_counts_up_past_its_initial_value(self, redis): + from lithops.multiprocessing import Semaphore + sem = Semaphore(1) + sem.release() + assert sem.get_value() == 2 + + def test_a_semaphore_can_start_empty(self, redis): + from lithops.multiprocessing import Semaphore + assert Semaphore(0).get_value() == 0 + + def test_an_rlock_can_be_taken_again_by_its_owner(self, redis): + from lithops.multiprocessing import RLock + lock = RLock() + assert lock.acquire() is True + assert lock.acquire() is True + + def test_a_lock_survives_being_pickled(self, redis): + from lithops.multiprocessing import Lock + lock = Lock() + restored = pickle.loads(pickle.dumps(lock)) + assert restored.acquire(block=False) is True + assert lock.get_value() == 0 + restored.release() + assert lock.get_value() == 1 + + def test_the_repr_shows_the_value(self, redis): + from lithops.multiprocessing import Lock + assert 'value=1' in repr(Lock()) + + +class TestCondition: + + def test_wait_returns_once_notified(self, redis): + from lithops.multiprocessing import Condition + cond = Condition() + with cond: + threading.Timer(0.1, lambda: _notify(cond)).start() + cond.wait(timeout=5) + + def test_notify_all_wakes_every_waiter(self, redis): + from lithops.multiprocessing import Condition + cond = Condition() + cond.acquire() + handles = [ + redis.rpush(cond._notify_handle, 'w-1'), + redis.rpush(cond._notify_handle, 'w-2'), + ] + assert handles[-1] == 2 + cond.notify_all() + assert redis.llen(cond._notify_handle) == 0 + assert redis.llen('w-1') == 1 + assert redis.llen('w-2') == 1 + + def test_wait_for_returns_at_once_when_already_true(self, redis): + from lithops.multiprocessing import Condition + cond = Condition() + with cond: + assert cond.wait_for(lambda: 'ready') == 'ready' + + def test_a_condition_can_wrap_a_given_lock(self, redis): + from lithops.multiprocessing import Condition, Lock + lock = Lock() + cond = Condition(lock) + with cond: + assert lock.get_value() == 0 + + def test_a_condition_survives_being_pickled(self, redis): + from lithops.multiprocessing import Condition + cond = Condition() + restored = pickle.loads(pickle.dumps(cond)) + assert restored._notify_handle == cond._notify_handle + + +def _notify(cond): + with cond: + cond.notify() + + +class TestEvent: + + def test_an_event_starts_clear(self, redis): + from lithops.multiprocessing import Event + assert Event().is_set() is False + + def test_set_and_clear(self, redis): + from lithops.multiprocessing import Event + event = Event() + event.set() + assert event.is_set() is True + event.clear() + assert event.is_set() is False + + def test_wait_reports_the_flag(self, redis): + """`if event.wait(timeout)` is how the standard library is used""" + from lithops.multiprocessing import Event + event = Event() + event.set() + assert event.wait(timeout=1) is True + + def test_wait_reports_a_timeout(self, redis): + from lithops.multiprocessing import Event + assert Event().wait(timeout=0.2) is False + + +class TestSharedCTypes: + + def test_raw_value_round_trips(self, redis): + from lithops.multiprocessing import RawValue + value = RawValue('i', 7) + assert value.value == 7 + value.value = 9 + assert value.value == 9 + + def test_raw_value_keeps_a_falsy_initial_value(self, redis): + from lithops.multiprocessing import RawValue + value = RawValue('d', 0.0).value + assert value == 0.0 and isinstance(value, float) + + def test_a_missing_attribute_still_raises(self, redis): + from lithops.multiprocessing import RawValue + with pytest.raises(AttributeError): + RawValue('i', 1).nonexistent + + def test_value_round_trips_and_locks(self, redis): + from lithops.multiprocessing import Value + value = Value('i', 3) + assert value.value == 3 + with value: + assert value.get_lock().get_value() == 0 + assert value.get_obj() == 3 + + def test_value_uses_the_lock_it_was_given(self, redis): + from lithops.multiprocessing import Value, Lock + lock = Lock() + value = Value('i', 1, lock=lock) + assert value.get_lock() is lock + + def test_raw_array_from_a_list(self, redis): + from lithops.multiprocessing import RawArray + array = RawArray('i', [1, 2, 3]) + assert len(array) == 3 + assert array[1] == 2 + assert array[:] == [1, 2, 3] + assert list(array) == [1, 2, 3] + + def test_raw_array_from_a_size(self, redis): + from lithops.multiprocessing import RawArray + assert RawArray('i', 3)[:] == [0, 0, 0] + + def test_raw_array_assignment(self, redis): + from lithops.multiprocessing import RawArray + array = RawArray('i', [1, 2, 3]) + array[0] = 9 + assert array[0] == 9 + array[1:3] = [7, 8] + assert array[:] == [9, 7, 8] + + def test_raw_array_rejects_a_bad_initializer(self, redis): + from lithops.multiprocessing import RawArray + with pytest.raises(ValueError, match='Invalid size or initializer'): + RawArray('i', 'nope') + + def test_a_char_raw_array_is_not_supported(self, redis): + from lithops.multiprocessing import RawArray + with pytest.raises(NotImplementedError): + RawArray('c', 3) + + def test_array_round_trips(self, redis): + from lithops.multiprocessing import Array + array = Array('i', [1, 2, 3]) + assert array.get_obj() == [1, 2, 3] + + def test_a_char_array_reads_back_as_bytes(self, redis): + from lithops.multiprocessing import Array + array = Array('c', b'abc') + assert array.value == b'abc' + assert array[0:2] == b'ab' + + def test_the_typecode_table_covers_the_standard_codes(self): + from lithops.multiprocessing import sharedctypes + assert sharedctypes.typecode_to_type['i'] is ctypes.c_int + assert sharedctypes.typecode_to_type['d'] is ctypes.c_double + + +class TestPackageSurface: + + def test_every_exported_name_exists(self): + import lithops.multiprocessing as mp + missing = [name for name in mp.__all__ if not hasattr(mp, name)] + assert missing == [] + + def test_the_whole_standard_library_surface_is_covered(self): + """ + A drop-in has to answer every name multiprocessing exports, or an + import of a ported module fails before any of it runs + """ + import multiprocessing as std + import lithops.multiprocessing as mp + assert [name for name in std.__all__ if not hasattr(mp, name)] == [] + assert set(std.__all__) <= set(mp.__all__) + + def test_the_exception_types_are_the_standard_hierarchy(self): + import lithops.multiprocessing as mp + assert issubclass(mp.ProcessError, Exception) + for error in (mp.BufferTooShort, mp.TimeoutError, mp.AuthenticationError): + assert issubclass(error, mp.ProcessError) + # Not the builtin, which is an OSError, as in the standard library + assert not issubclass(mp.TimeoutError, OSError) + + def test_the_helpers_a_ported_script_calls_are_no_ops(self): + import lithops.multiprocessing as mp + assert mp.freeze_support() is None + assert mp.allow_connection_pickling() is None + assert mp.set_executable('/usr/bin/python3') is None + assert mp.set_forkserver_preload(['os']) is None + + def test_get_logger_returns_the_package_logger(self): + import logging + import lithops.multiprocessing as mp + from lithops.multiprocessing import context + + assert mp.get_logger() is logging.getLogger('lithops.multiprocessing') + assert isinstance(context, types.ModuleType) + saved_flag = context._log_to_stderr + streamed = mp.get_logger() + saved_handlers = list(streamed.handlers) + try: + context._log_to_stderr = False + streamed.handlers.clear() + assert mp.log_to_stderr(logging.WARNING) is streamed + assert streamed.level == logging.WARNING + # Twice must not print every line twice + mp.log_to_stderr() + assert len(streamed.handlers) == 1 + finally: + context._log_to_stderr = saved_flag + streamed.handlers[:] = saved_handlers + streamed.setLevel(logging.NOTSET) + + def test_the_context_submodule_is_not_shadowed(self): + """ + `multiprocessing.context` is the module, so ported code reaching for + `mp.context.` has to find one here too + """ + import lithops.multiprocessing as mp + assert isinstance(mp.context, types.ModuleType) + assert mp.context.CloudContext is mp.DefaultContext + + def test_buffer_too_short_is_the_one_this_package_exports(self): + """ + Raising the standard library's would slip past + `except lithops.multiprocessing.BufferTooShort` + """ + from lithops.multiprocessing import connection + import lithops.multiprocessing as mp + assert connection.BufferTooShort is mp.BufferTooShort + + def test_thread_pool_is_available_under_its_standard_name(self): + from lithops.multiprocessing.pool import Pool, ThreadPool + assert issubclass(ThreadPool, Pool) + + def test_the_unimplemented_process_helpers_say_so(self): + import lithops.multiprocessing as mp + for call in (mp.active_children, mp.parent_process): + with pytest.raises(NotImplementedError): + call() + + def test_the_cloudpickle_round_trip_of_a_shared_object(self, redis): + """cloudpickle is what carries these into the job payload""" + from lithops.multiprocessing import Lock + lock = Lock() + assert cloudpickle.loads(cloudpickle.dumps(lock))._name == lock._name diff --git a/lithops/tests/test_utils.py b/lithops/tests/test_utils.py index 86571790d..046ce6fcf 100644 --- a/lithops/tests/test_utils.py +++ b/lithops/tests/test_utils.py @@ -144,6 +144,12 @@ def test_namedtuple_plus_extra_args_is_not_concatenated(self): assert format_data([pt], (9,)) == [(pt, 9)] +def _response_future(): + """A ResponseFuture with none of the state the constructor would set""" + from lithops.future import ResponseFuture + return ResponseFuture.__new__(ResponseFuture) + + class TestVerifyArgs: def test_futures_list_becomes_future_kwargs(self): @@ -153,6 +159,41 @@ def test_futures_list_becomes_future_kwargs(self): {'future': 'f2'}, ] + def test_a_plain_list_of_futures_is_a_chain_too(self): + """ + A slice of a FuturesList, or one built by a comprehension, is a plain + list. Binding a future as if it were data fails with an error that + says nothing about chaining + """ + futures = [_response_future(), _response_future()] + assert verify_args(lambda x, y: x, futures, None) == [ + {'future': futures[0]}, + {'future': futures[1]}, + ] + + def test_a_slice_of_a_futures_list_still_chains(self): + futures = FuturesList([_response_future() for _ in range(3)]) + assert verify_args(lambda x: x, futures[:2], None) == [ + {'future': futures[0]}, + {'future': futures[1]}, + ] + + def test_futures_mixed_with_plain_data_raises(self): + with pytest.raises(ValueError, match='mixes futures'): + verify_args(lambda x: x, [_response_future(), 7], None) + + def test_extra_args_with_a_chain_raises(self): + """ + The worker binds the previous result to the whole signature, leaving + no room for them. Every activation would fail on a missing argument + """ + futures = FuturesList([_response_future()]) + with pytest.raises(ValueError, match='extra_args is not supported'): + verify_args(lambda x, factor: x, futures, (10,)) + + def test_an_empty_futures_list_submits_nothing(self): + assert verify_args(lambda x: x, FuturesList(), None) == [] + def test_positional_and_dict_binding(self): def fn(a, b): return a + b @@ -597,9 +638,12 @@ def __init__(self): assert fl.get_result() == [1] fl2 = FuturesList([1, 2]) - fl2.executor = object() + executor = object() + fl2.executor = executor dumped = pickle.dumps(fl2) - assert fl2.executor is None + # Pickling reports the list, it does not consume it: the executor of + # the one being pickled has to survive + assert fl2.executor is executor loaded = pickle.loads(dumped) assert list(loaded) == [1, 2] assert loaded.executor is None diff --git a/lithops/utils.py b/lithops/utils.py index 167e61a27..e3e8138e1 100644 --- a/lithops/utils.py +++ b/lithops/utils.py @@ -206,6 +206,24 @@ def _extend_futures(self, fs): self.extend(fs) def map(self, map_function, sync=False, **kwargs): + """ + Chains a new map job that takes the results of this one as its input. + The intermediate results are read by the workers of the new job and + are never downloaded to the client. + + :param map_function: The function to map over the results + :param sync: Wait for this job before invoking the new one. Left + False, the new job is invoked right away and each of its workers + blocks until its own input is ready, which is worker time you pay + for. Waiting costs about the same wall-clock time and no idle + workers, at the price of not overlapping the two invocations + :param kwargs: Passed on to + :meth:`~lithops.executors.FunctionExecutor.map`. ``extra_args`` is + not among them: a chained function is called with the result of + the previous one and nothing else + + :return: This list, now holding the futures of the new job + """ self._create_executor() if sync: self.executor.wait(self) @@ -214,6 +232,19 @@ def map(self, map_function, sync=False, **kwargs): return self def map_reduce(self, map_function, reduce_function, sync=False, **kwargs): + """ + Chains a new map-reduce job that takes the results of this one as the + input of its map stage. + + :param map_function: The function to map over the results + :param reduce_function: The function to reduce the map results with + :param sync: Wait for this job before invoking the new one. See + :meth:`map` + :param kwargs: Passed on to + :meth:`~lithops.executors.FunctionExecutor.map_reduce` + + :return: This list, now holding the futures of the new job + """ self._create_executor() if sync: self.executor.wait(self) @@ -224,17 +255,42 @@ def map_reduce(self, map_function, reduce_function, sync=False, **kwargs): return self def wait(self, **kwargs): + """ + Waits for every job of the chain, not only for the last one. + + :param kwargs: Passed on to + :meth:`~lithops.executors.FunctionExecutor.wait` + + :return: `(fs_done, fs_notdone)` + """ self._create_executor() return self.executor.wait(self._all_futures(), **kwargs) def get_result(self, **kwargs): + """ + Returns the results of the last job of the chain. The intermediate + ones are read by the workers, never by the client. + + :param kwargs: Passed on to + :meth:`~lithops.executors.FunctionExecutor.get_result` + + :return: The results of the last job + """ self._create_executor() return self.executor.get_result(self._all_futures(), **kwargs) def __reduce__(self): - # The executor is not picklable, and a rehydrated list creates its own - self.executor = None - return super().__reduce__() + # The executor is not picklable, and a rehydrated list creates its + # own. Dropped from the pickled state rather than from the object, + # so that pickling a list does not detach the one being pickled + reduced = list(super().__reduce__()) + # A list that never had an attribute set reduces without a state + state = reduced[2] if len(reduced) > 2 else None + if isinstance(state, dict) and 'executor' in state: + state = dict(state) + state.pop('executor') + reduced[2] = state + return tuple(reduced) _MODE_TO_DEFAULT_BACKEND = { @@ -706,15 +762,57 @@ def _user_signature(func) -> inspect.Signature: return func_sig.replace(parameters=user_parameters) +def _chained_futures(iterdata): + """ + The futures of a previous job used as the input of this one, or None + when the iterdata is plain data. + + A slice of a FuturesList, or a list built from one, is a chain too: both + lose the FuturesList type, and binding a future to a parameter as if it + were data fails with an error that says nothing about chaining + """ + from lithops.future import ResponseFuture + + if isinstance(iterdata, FuturesList): + return list(iterdata) + + if not isinstance(iterdata, (list, tuple)) or not iterdata: + return None + + futures = [ + elem for elem in iterdata if isinstance(elem, ResponseFuture) + ] + if not futures: + return None + if len(futures) != len(iterdata): + raise ValueError( + "The iterdata mixes futures of a previous job with plain data. " + "Chaining takes the futures of one job as the whole input of " + "the next one" + ) + return list(iterdata) + + def verify_args(func, iterdata, extra_args): """ Binds every element of the iterdata to the params of the map function, returning one kwargs dict per call """ - if isinstance(iterdata, FuturesList): + chained = _chained_futures(iterdata) + if chained is not None: + if extra_args: + # The worker binds the result of the previous call to the whole + # signature, so there is no room left for these. Said here rather + # than letting every activation fail on a missing argument + raise ValueError( + "extra_args is not supported when chaining jobs: a chained " + "function is called with the result of the previous one and " + "nothing else. Return the extra values from the previous " + "function, or get its results and start a new job with them" + ) # A chained job receives the future of the previous one, which is only # bound to a param once the previous job finishes - return [{'future': f} for f in iterdata] + return [{'future': f} for f in chained] data = format_data(iterdata, extra_args) func_sig = _user_signature(func)