Update multiprocessing

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.

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 <notebooks/function_chaining>`.
+
+.. autoclass:: lithops.utils.FuturesList
+   :members: map, map_reduce, wait, get_result
+   :show-inheritance:

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,13 +44,77 @@ 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
 ---------------------
 
 Lithops also implements all stateful abstractions from Python multiprocessing: Queue, Pipes, Shared memory, Events, etc.
 
 Since FaaS lacks mechanisms for function-to-function communication, a `Redis <https://redis.io/>`_ 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.

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": {

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,

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):
         """

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.<name>` of ported code fail. The class is `DefaultContext`

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):

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.