Update Multiprocessing and Monitoring

.github/workflows/tests-all-os.yml

@@ -28,6 +28,9 @@ jobs:
           cache-dependency-path: setup.py
 
       - name: Install Redis
+        # There is no Redis build for Windows. The tests that need a server
+        # skip themselves where none is reachable
+        if: runner.os != 'Windows'
         uses: shogo82148/actions-setup-redis@v1
         with:
           redis-version: '7.x'
@@ -36,16 +39,6 @@ jobs:
         run: |
           pip3 install -U .[tests]
 
-      - name: Create Lithops config file
-        run: |
-          mkdir -p $HOME/.lithops
-          echo "lithops:" >> $HOME/.lithops/config
-          echo "    monitoring_interval: 0.1" >> $HOME/.lithops/config
-          echo "    log_level: DEBUG" >> $HOME/.lithops/config
-          echo "    include_modules: None" >> $HOME/.lithops/config
-          echo "localhost:" >> $HOME/.lithops/config
-          echo "    version: 2" >> $HOME/.lithops/config
-
       - name: Run Lithops tests
         run: |
           cd lithops/tests
@@ -56,9 +49,12 @@ jobs:
         shell: bash
         run: |
           if [ "$RUNNER_OS" == "Linux" ]; then
-            tail -n 500 /tmp/lithops-runner/localhost-runner.log
+            log=/tmp/lithops-runner/localhost-runner.log
           elif [ "$RUNNER_OS" == "macOS" ]; then
-            tail -n 500 $TMPDIR/lithops-runner/localhost-runner.log
-          elif [ "$RUNNER_OS" == "Windows" ]; then
-            tail -n 500 "C:\Users\RUNNER~1\AppData\Local\Temp\lithops-root\localhost-runner.log"
+            log=$TMPDIR/lithops-runner/localhost-runner.log
+          else
+            log="C:\Users\RUNNER~1\AppData\Local\Temp\lithops-root\localhost-runner.log"
           fi
+          # A job that failed before running anything leaves no log, which
+          # should not fail this step on top of the failure being reported
+          tail -n 500 "$log" || echo "No runner log at $log"

lithops/concurrent/futures.py

@@ -373,6 +373,9 @@ def __init__(
         self._inner = executor
 
         self._lock = threading.RLock()
+        # Native submission can block on runtime deployment or storage. Keep
+        # it separate from the lock used to resolve already submitted calls.
+        self._submission_lock = threading.RLock()
         self._is_shutdown = False
         self._broken = None
         self._torn_down = False
@@ -632,17 +635,20 @@ def _check_running(self):
     # -- concurrent.futures.Executor ----------------------------------------
 
     def submit(self, fn, /, *args, **kwargs):
-        with self._lock:
-            self._check_running()
-        payload = (fn, args, kwargs)
-        job_kwargs = self._job_kwargs()
-        call_async = getattr(self._inner, 'call_async', None)
-        if call_async is not None:
-            lf = call_async(_call, payload, **job_kwargs)
-        else:
-            # A duck-typed executor may only expose map()
-            lf = self._inner.map(_call, [payload], **job_kwargs)[0]
-        return self._track([lf])[0]
+        # Shutdown must see every accepted submission in _pending before it
+        # can release the native executor, including while submission blocks.
+        with self._submission_lock:
+            with self._lock:
+                self._check_running()
+            payload = (fn, args, kwargs)
+            job_kwargs = self._job_kwargs()
+            call_async = getattr(self._inner, 'call_async', None)
+            if call_async is not None:
+                lf = call_async(_call, payload, **job_kwargs)
+            else:
+                # A duck-typed executor may only expose map()
+                lf = self._inner.map(_call, [payload], **job_kwargs)[0]
+            return self._track([lf])[0]
 
     def map(
         self,
@@ -683,15 +689,16 @@ def take(n):
         def submit_batch(items):
             if not items:
                 return []
-            with self._lock:
-                self._check_running()
-            lfs = self._inner.map(
-                _call,
-                [(fn, args, {}) for args in items],
-                chunksize=chunksize,
-                **self._job_kwargs()
-            )
-            return self._track(lfs)
+            with self._submission_lock:
+                with self._lock:
+                    self._check_running()
+                lfs = self._inner.map(
+                    _call,
+                    [(fn, args, {}) for args in items],
+                    chunksize=chunksize,
+                    **self._job_kwargs()
+                )
+                return self._track(lfs)
 
         fs = collections.deque(
             submit_batch(take(buffersize if buffersize else None))
@@ -714,9 +721,10 @@ def result_iterator():
         return result_iterator()
 
     def shutdown(self, wait=True, *, cancel_futures=False):
-        with self._lock:
-            self._is_shutdown = True
-            pending = list(self._pending)
+        with self._submission_lock:
+            with self._lock:
+                self._is_shutdown = True
+                pending = list(self._pending)
         if cancel_futures:
             # Lithops cannot recall an activation it already dispatched, so
             # every future here is running and declines. Kept so that callers

lithops/constants.py

@@ -57,6 +57,15 @@
 LOGS_PREFIX = "lithops.logs"
 RUNTIMES_PREFIX = "lithops.runtimes"
 
+# ---------------------------------------------------------------------------
+# Executor environment variables
+# ---------------------------------------------------------------------------
+SESSION_ID_ENV = '__LITHOPS_SESSION_ID'
+TOTAL_EXECUTORS_ENV = '__LITHOPS_TOTAL_EXECUTORS'
+MONITORING_QUEUES_ENV = '__LITHOPS_MONITORING_QUEUES'
+WORKER_ENV = 'LITHOPS_WORKER'
+REDUCE_JOB_ENV = '__LITHOPS_REDUCE_JOB'
+
 # ---------------------------------------------------------------------------
 # Local filesystem
 # ---------------------------------------------------------------------------
@@ -136,7 +145,6 @@ def _in_temp(*parts):
     'aws_batch',
     'gcp_cloudrun',
     'gcp_functions',
-    'cloudrun',
     'azure_functions',
     'azure_containers',
     'aliyun_fc',

lithops/executors.py

@@ -55,6 +55,7 @@
     create_futures_list,
     FuturesList,
     _as_future_list as wrap_as_future_list,
+    _chained_futures,
     log_prefix,
 )
 from lithops.localhost import LocalhostHandlerV1, LocalhostHandlerV2
@@ -330,9 +331,8 @@ def _disable_iterdata_output(iterdata):
         Marks the futures used as input as consumed, so that get_result()
         returns the output of this job only
         """
-        if isinstance(iterdata, FuturesList):
-            for fut in iterdata:
-                fut._produce_output = False
+        for fut in _chained_futures(iterdata) or ():
+            fut._produce_output = False
 
     def _invoke(self, job):
         """

lithops/job/job.py

@@ -22,7 +22,6 @@
 import inspect
 import pickle
 import logging
-import weakref
 from collections.abc import Callable, Iterable, Mapping
 from types import SimpleNamespace
 from typing import Any, Dict, List, Optional, Set, Tuple
@@ -35,14 +34,18 @@
     SerializeIndependent, create_module_data, write_module_data
 )
 from lithops.constants import (
-    MAX_AGG_DATA_SIZE, SERVERLESS, STANDALONE, CUSTOM_RUNTIME_DIR, JOBS_PREFIX
+    MAX_AGG_DATA_SIZE,
+    SERVERLESS,
+    STANDALONE,
+    CUSTOM_RUNTIME_DIR,
+    JOBS_PREFIX,
+    REDUCE_JOB_ENV
 )
 
 
 logger = logging.getLogger(__name__)
 
 FUNCTION_CACHE = set()
-_FUNC_SERIALIZE_CACHE = weakref.WeakKeyDictionary()
 MAX_DATA_IN_PAYLOAD = 8 * 1024  # Per invocation. 8KB
 
 
@@ -57,51 +60,6 @@ def invalidate_function_cache(executor_id: str) -> None:
     )
 
 
-def _freeze_module_set(mods: Optional[Set[str]]) -> Optional[Tuple[str, ...]]:
-    if mods is None:
-        return None
-    return tuple(sorted(mods))
-
-
-def _cached_func_serialize(
-    serializer: Any,
-    func: Callable,
-    inc_modules: Optional[Set[str]],
-    exc_modules: Set[str]
-) -> Tuple[bytes, Set[str]]:
-    """
-    Serializes a function and resolves its modules, reusing the result of a
-    previous job that ran the same function with the same module filters
-    """
-    subkey = (
-        _freeze_module_set(inc_modules),
-        _freeze_module_set(exc_modules),
-    )
-    try:
-        per_func = _FUNC_SERIALIZE_CACHE[func]
-    except TypeError:
-        # Not every callable can be weak referenced, so caching is best effort
-        per_func = None
-    except KeyError:
-        per_func = {}
-        try:
-            _FUNC_SERIALIZE_CACHE[func] = per_func
-        except TypeError:
-            per_func = None
-
-    if per_func is not None:
-        cached = per_func.get(subkey)
-        if cached is not None:
-            return cached
-
-    func_str = serializer.dumps([func])[0]
-    func_paths = serializer.module_paths([func], inc_modules, exc_modules)
-    cached = (func_str, func_paths)
-    if per_func is not None:
-        per_func[subkey] = cached
-    return cached
-
-
 def _serialize_func_and_data(
     serializer: Any,
     func: Callable,
@@ -110,9 +68,10 @@ def _serialize_func_and_data(
     exc_modules: Set[str]
 ) -> Tuple[bytes, List[bytes], Set[str]]:
     """
-    Serializes the function apart from its data, so that the function can be
-    cached across jobs. Falls back to serializing everything in one go for
-    serializers that only expose a call interface
+    Serializes the function and data with the current callable state and
+    target runtime's dependencies. Identical payloads are deduplicated at
+    upload time; callable identity alone cannot detect changed captured state.
+    Falls back to serializers that only expose a call interface
     """
     dumps = getattr(serializer, 'dumps', None)
     module_paths = getattr(serializer, 'module_paths', None)
@@ -122,9 +81,8 @@ def _serialize_func_and_data(
         )
         return ser[0], ser[1:], paths
 
-    func_str, func_paths = _cached_func_serialize(
-        serializer, func, inc_modules, exc_modules
-    )
+    func_str = dumps([func])[0]
+    func_paths = module_paths([func], inc_modules, exc_modules)
     data_strs = dumps(iterdata)
     # The data is only inspected for modules when the module manager is on
     # and no explicit include list was given
@@ -233,7 +191,7 @@ def create_reduce_job(
             offset = end
 
     ext_env = {} if extra_env is None else extra_env.copy()
-    ext_env['__LITHOPS_REDUCE_JOB'] = True
+    ext_env[REDUCE_JOB_ENV] = True
 
     iterdata = utils.verify_args(reduce_function, iterdata, extra_args)
 

lithops/job/partitioner.py

@@ -16,6 +16,7 @@
 #
 
 import os
+import re
 import posixpath
 import logging
 import requests
@@ -29,6 +30,11 @@
 
 logger = logging.getLogger(__name__)
 
+#: A local path on Windows, where nothing starts with a separator: a drive
+#: letter followed by one, or a UNC share. The separator is required, so
+#: that a storage key is not read as a path just for holding a colon
+_WINDOWS_PATH = re.compile(r'^(?:[A-Za-z]:[\\/]|\\\\)')
+
 CHUNK_THRESHOLD = 128 * 1024  # 128KB
 
 # Backends whose listing supports a glob pattern
@@ -58,9 +64,10 @@ def create_partitions(
     logger.debug("Parsing input data")
 
     for elem in map_iterdata:
-        if str(elem['obj']).startswith('http'):
+        obj = str(elem['obj'])
+        if obj.startswith('http'):
             urls.append(elem)
-        elif str(elem['obj']).startswith('/'):
+        elif obj.startswith('/') or _WINDOWS_PATH.match(obj):
             paths.append(elem)
         else:
             objects.append(elem)

lithops/localhost/config.py

@@ -29,7 +29,7 @@
 
 LOCALHOST_EXECUTION_TIMEOUT = 3600
 
-_WINDOWS_PATH = re.compile(r'^[A-Za-z]:\\.*$')
+_WINDOWS_PATH = re.compile(r'^(?:[A-Za-z]:[\\/]|\\\\)')
 # Interpreters like python, python3, python3.12, python.exe — not docker tags
 # such as python:3.12.
 _PYTHON_INTERPRETER = re.compile(

lithops/monitoring/backends/__init__.py

@@ -33,8 +33,8 @@
     lithops.monitoring.backends.<name>
 
 The name of the package is also the name of the config section the backend
-reads and the value ``monitoring:`` selects it by; ``Monitor`` checks that
-its ``backend_name`` agrees with the package it is defined in.
+reads and the value ``monitoring:`` selects it by; ``Monitor`` takes its
+``backend_name`` from the package it is defined in.
 """
 
 import importlib
@@ -44,14 +44,6 @@
 #: backend can act as a monitoring channel, so this always works
 DEFAULT_BACKEND = 'storage'
 
-#: Attribute name each backend package exports, and the base class it has
-#: to be a subclass of. Loaded lazily so that importing this module does
-#: not pull the whole monitoring package in
-_CONTRACT = {
-    'MonitoringBackend': ('lithops.monitoring.monitor', 'Monitor'),
-    'CallStatus': ('lithops.monitoring.status', 'CallStatus'),
-}
-
 
 def resolve_backend(
         config: Optional[Dict[str, Any]] = None,
@@ -95,28 +87,13 @@ def import_backend_module(backend: str, submodule: Optional[str] = None):
 
 def load_backend_attr(backend: str, attr: str):
     """
-    Returns ``attr`` of ``lithops.monitoring.backends.<backend>``, checked
-    against the backend contract so that a package that exports the wrong
-    thing fails here rather than halfway through a job
+    Returns ``attr`` of ``lithops.monitoring.backends.<backend>``, naming
+    the backend when the package does not export it
     """
     module = import_backend_module(backend)
-
     try:
-        value = getattr(module, attr)
+        return getattr(module, attr)
     except AttributeError as exc:
         raise ValueError(
             f"Monitoring backend '{backend}' exports no {attr}"
         ) from exc
-
-    expected = _CONTRACT.get(attr)
-    if expected is None:
-        return value
-
-    base_module, base_name = expected
-    base = getattr(importlib.import_module(base_module), base_name)
-    if not (isinstance(value, type) and issubclass(value, base)):
-        raise ValueError(
-            f"Monitoring backend '{backend}' exports {attr}="
-            f"{value!r}, which is not a {base_name} subclass"
-        )
-    return value

lithops/monitoring/backends/storage/storage.py

@@ -260,8 +260,8 @@ def _poll_and_process_job_status(self):
         )
 
         self._generate_tokens(callids_running, callids_done)
-        self._tag_future_as_running(callids_running)
         self._tag_future_as_ready(callids_done)
+        self._tag_future_as_running(callids_running)
 
         self._print_status_log()