From e613f5941a9d4a485b6c31a02a8d01f44fb5cdd0 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Mon, 7 Sep 2026 23:16:48 +0200 Subject: [PATCH 01/10] Fix executor id --- CHANGELOG.md | 1 + lithops/tests/test_utils.py | 94 +++++++++++++++++++++++++++++++++++- lithops/utils.py | 95 ++++++++++++++++++++++++++++++------- 3 files changed, 171 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e02b0772..6b20cf182 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,7 @@ - [Multiprocessing] `imap()` and `imap_unordered()` now default to the configured chunksize. ### Fixed +- [Core] Fixed executor IDs repeating in a process whose environment is reset between executors, which made two of them share their storage keys and read back each other's results. - [Redis] Fixed `put_object()` rejecting file-like objects, which also made `upload_file()` and `download_file()` always fail. - [Redis] Fixed `head_object()`, which reported every key as missing on Redis 7 and up because it relied on the now-disabled `DEBUG OBJECT` command. diff --git a/lithops/tests/test_utils.py b/lithops/tests/test_utils.py index bbe7c3055..38f7dd73b 100644 --- a/lithops/tests/test_utils.py +++ b/lithops/tests/test_utils.py @@ -15,6 +15,7 @@ import io import logging import pickle +import os import threading import zipfile from collections import namedtuple @@ -23,6 +24,7 @@ import pytest from lithops import constants +from lithops import utils as lithops_utils from lithops.utils import ( CountDownLatch, CURRENT_PY_VERSION, @@ -356,9 +358,16 @@ def test_as_future_list_and_future_id(self): assert _as_future_list(plain) is plain assert _future_id(future) == ('e', 'j', 'c') - def test_create_executor_id_reuses_session_and_increments(self, monkeypatch): + @staticmethod + def _fresh_session(monkeypatch): + """A process that has not created an executor yet""" + monkeypatch.setattr(lithops_utils, '_SESSION_ID', None) + monkeypatch.setattr(lithops_utils, '_EXECUTOR_COUNT', 0) monkeypatch.delenv('__LITHOPS_SESSION_ID', raising=False) monkeypatch.delenv('__LITHOPS_TOTAL_EXECUTORS', raising=False) + + def test_create_executor_id_reuses_session_and_increments(self, monkeypatch): + self._fresh_session(monkeypatch) first = create_executor_id(lenght=4) second = create_executor_id(lenght=4) session, num = first.rsplit('-', 1) @@ -367,6 +376,89 @@ def test_create_executor_id_reuses_session_and_increments(self, monkeypatch): assert second == f'{session}-1' assert get_executor_id() == second + def test_create_executor_id_exports_the_session(self, monkeypatch): + self._fresh_session(monkeypatch) + executor_id = create_executor_id() + session, num = executor_id.rsplit('-', 1) + + # Exported so that a process spawned from this one joins the session + assert os.environ['__LITHOPS_SESSION_ID'] == session + assert os.environ['__LITHOPS_TOTAL_EXECUTORS'] == num + + def test_create_executor_id_survives_a_reset_environment(self, monkeypatch): + """ + The counter is what tells two executors of one process apart, and + it used to live only in the environment. Anything that saves and + restores os.environ -- this test suite does, between every test -- + put the counter back to zero, so a process kept handing out the + same ID with nothing but six random characters keeping the storage + keys of one executor apart from the next + """ + self._fresh_session(monkeypatch) + ids = [] + for _ in range(4): + ids.append(create_executor_id()) + os.environ.pop('__LITHOPS_SESSION_ID', None) + os.environ.pop('__LITHOPS_TOTAL_EXECUTORS', None) + + assert len(set(ids)) == len(ids) + sessions = {executor_id.rsplit('-', 1)[0] for executor_id in ids} + assert len(sessions) == 1 + assert [executor_id.rsplit('-', 1)[1] for executor_id in ids] == [ + '0', '1', '2', '3' + ] + + def test_create_executor_id_joins_a_session_it_is_given(self, monkeypatch): + # What a worker does before it runs a task, and what the remote + # invoker does for the job it spawns + self._fresh_session(monkeypatch) + create_executor_id() + + monkeypatch.setenv('__LITHOPS_SESSION_ID', 'job-key-00000') + assert create_executor_id() == 'job-key-00000-0' + + monkeypatch.setenv('__LITHOPS_SESSION_ID', 'job-key-00001') + assert create_executor_id() == 'job-key-00001-0' + + def test_create_executor_id_continues_the_count_it_inherits(self, monkeypatch): + # A process spawned by one that had already created eight executors + self._fresh_session(monkeypatch) + monkeypatch.setenv('__LITHOPS_SESSION_ID', 'parent') + monkeypatch.setenv('__LITHOPS_TOTAL_EXECUTORS', '7') + + assert create_executor_id() == 'parent-8' + assert create_executor_id() == 'parent-9' + + def test_create_executor_id_ignores_an_unreadable_count(self, monkeypatch): + self._fresh_session(monkeypatch) + monkeypatch.setenv('__LITHOPS_SESSION_ID', 'parent') + monkeypatch.setenv('__LITHOPS_TOTAL_EXECUTORS', 'not-a-number') + + assert create_executor_id() == 'parent-0' + + def test_create_executor_id_is_unique_under_concurrency(self, monkeypatch): + self._fresh_session(monkeypatch) + ids = [] + lock = threading.Lock() + + def make(): + executor_id = create_executor_id() + with lock: + ids.append(executor_id) + + threads = [threading.Thread(target=make) for _ in range(32)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert len(set(ids)) == 32 + + def test_get_executor_id_without_an_executor(self, monkeypatch): + self._fresh_session(monkeypatch) + with pytest.raises(KeyError): + get_executor_id() + def test_monitoring_queue_chain_matches_the_shapes_in_use(self, monkeypatch): # These are the chains the id-derived formula produced, and every id # shape Lithops builds today. The chain must not change for them diff --git a/lithops/utils.py b/lithops/utils.py index bbdc2ea5b..d92dec8c9 100644 --- a/lithops/utils.py +++ b/lithops/utils.py @@ -78,25 +78,76 @@ def _future_id(fut): return (fut.executor_id, fut.job_id, fut.call_id) +#: The session this process belongs to, and how many executors it has +#: created. Held here rather than only in the environment: the counter is +#: what makes two executor IDs of one process different, and an +#: environment can be rewritten under a running process +_SESSION_LOCK = threading.Lock() +_SESSION_ID = None +_EXECUTOR_COUNT = 0 + +SESSION_ID_ENV = '__LITHOPS_SESSION_ID' +TOTAL_EXECUTORS_ENV = '__LITHOPS_TOTAL_EXECUTORS' + + +def _inherited_executor_count(): + """ + The number the session that this process joined had reached, so that + the executors created here carry on from it instead of repeating it + """ + try: + return int(os.environ[TOTAL_EXECUTORS_ENV]) + 1 + except (KeyError, TypeError, ValueError): + return 0 + + def create_executor_id(lenght=6): """ - Creates the ID of a new executor. Executors of the same session share the - session ID and are told apart by a counter, both kept in the environment - so that they survive across processes + Creates the ID of a new executor. Executors of the same session share + the session ID and are told apart by a counter. + + Both are exported to the environment, so that a process spawned from + this one joins the same session, and a worker can put a session of its + own there for the task it is about to run. They are *read* back only + when the environment names a session other than the one this process + is already in. + + That last part is the difference between an ID that is unique and one + that is merely unlikely to repeat. The counter is what tells two + executors of the same process apart; a process that lost the variable + -- anything that saves and restores ``os.environ`` does that -- used + to start counting from zero again, which left the six characters of + the session ID as the only thing keeping its storage keys apart from + those of every executor it had already created. """ - if '__LITHOPS_SESSION_ID' in os.environ: - session_id = os.environ['__LITHOPS_SESSION_ID'] - else: - session_id = uuid_str().replace('/', '')[:lenght] - os.environ['__LITHOPS_SESSION_ID'] = session_id + global _SESSION_ID, _EXECUTOR_COUNT + + with _SESSION_LOCK: + env_session = os.environ.get(SESSION_ID_ENV) + if _SESSION_ID is None: + # The first executor of this process. It joins the session the + # process that spawned this one left in the environment, + # carrying on from wherever that had got to, or starts one + if env_session: + _SESSION_ID = env_session + _EXECUTOR_COUNT = _inherited_executor_count() + else: + _SESSION_ID = uuid_str().replace('/', '')[:lenght] + _EXECUTOR_COUNT = 0 + elif env_session and env_session != _SESSION_ID: + # The process has been put in a different session, which is + # what a worker does before each task it runs. The count of + # the session it left does not carry over + _SESSION_ID = env_session + _EXECUTOR_COUNT = 0 - if '__LITHOPS_TOTAL_EXECUTORS' in os.environ: - exec_num = int(os.environ['__LITHOPS_TOTAL_EXECUTORS']) + 1 - else: - exec_num = 0 - os.environ['__LITHOPS_TOTAL_EXECUTORS'] = str(exec_num) + exec_num = _EXECUTOR_COUNT + _EXECUTOR_COUNT += 1 + + os.environ[SESSION_ID_ENV] = _SESSION_ID + os.environ[TOTAL_EXECUTORS_ENV] = str(exec_num) - return f'{session_id}-{exec_num}' + return f'{_SESSION_ID}-{exec_num}' # Carries the monitoring queues of an executor down to the workers, so that an @@ -150,10 +201,18 @@ def monitoring_queues(executor_id: str) -> List[str]: def get_executor_id(): - """Returns the ID of the last executor created in this session""" - session_id = os.environ['__LITHOPS_SESSION_ID'] - exec_num = os.environ['__LITHOPS_TOTAL_EXECUTORS'] - return f'{session_id}-{exec_num}' + """ + Returns the ID of the last executor created in this session. + + Read from this process rather than from the environment, so that it + still answers after something has rewritten the environment, and + raises the same KeyError as it always did when no executor has been + created at all + """ + with _SESSION_LOCK: + if _SESSION_ID is None: + raise KeyError(SESSION_ID_ENV) + return f'{_SESSION_ID}-{_EXECUTOR_COUNT - 1}' def iterchunks(lst, n): From 2d7de9ff19dfe80ac578707571eb6b5274a4adf0 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Tue, 8 Sep 2026 12:42:34 +0200 Subject: [PATCH 02/10] Update constants --- CHANGELOG.md | 145 ++++++++++------------------ lithops/constants.py | 9 ++ lithops/job/job.py | 9 +- lithops/multiprocessing/process.py | 5 +- lithops/storage/storage.py | 10 +- lithops/telemetry/exporter.py | 3 +- lithops/tests/test_constants.py | 13 +++ lithops/tests/test_invokers.py | 4 +- lithops/tests/test_job.py | 12 ++- lithops/tests/test_storage_layer.py | 11 ++- lithops/tests/test_telemetry.py | 3 +- lithops/tests/test_utils.py | 71 ++++++-------- lithops/tests/test_worker.py | 31 +++--- lithops/utils.py | 105 +++++++------------- lithops/worker/handler.py | 17 +++- lithops/worker/invoker.py | 10 +- lithops/worker/jobrunner.py | 3 +- 17 files changed, 218 insertions(+), 243 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b20cf182..74ec8a725 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,136 +10,95 @@ - [Core] Added `clean_jobs` to `wait()`, to keep the temporary data until the results are read. - [AWS Batch] Added the `instance_types` config option for EC2/SPOT compute environments. - [Multiprocessing] Added `timeout` to `acquire()`, and `_getvalue()`, `_callmethod()` and `copy_proxy()` to the manager proxies. -- [Multiprocessing] Added `ThreadPool`, `Manager`, the standard error classes and the module-level helpers (`freeze_support()`, `get_logger()`, `log_to_stderr()`). +- [Multiprocessing] Added `ThreadPool`, the standard error classes and the module-level helpers (`freeze_support()`, `get_logger()`, `log_to_stderr()`). - [Telemetry] Added `lithops.telemetry`, which turns the call statuses the monitor reads into 29 Prometheus or OpenTelemetry metrics. - [Tests] Added a unit test suite for all non-backend modules. ### Changed -- [Monitoring] Reorganised job monitoring as pluggable backends. -- [Monitoring] The message backends now delete their queue on cleanup instead of on every `stop()`, so a later `map()` can reuse it. -- [Monitoring] The queue, topic or subscription an executor creates is now deleted on exit even with `data_cleaner: False`. -- [Monitoring] The storage backend now lists only the prefixes of the jobs it still watches. +- [Core] `wait()` now returns two empty lists instead of `None` for empty input and reuses a single thread pool across polls, and `get_result()` deletes the temporary data once the results are in, not during the wait. +- [Core] Stopping an executor now waits for the invocations already in flight instead of returning while its invoker threads still run. +- [Monitoring] Reorganised job monitoring as pluggable backends: the message backends now delete their queue on cleanup instead of on every `stop()`, so a later `map()` can reuse it, and the storage backend lists only the prefixes of the jobs it still watches. - [Monitoring] Status lines (Pending/Running/Done) are now logged every 30s instead of on every activation. -- [Core] Results under 8KB now travel in the call status instead of a separate storage object. -- [Core] `wait()` now returns two empty lists for empty input instead of `None`. -- [Core] `wait()` now reuses a single thread pool instead of creating one per poll. -- [Core] `get_result()` now deletes the temporary data once the results are in, not during the wait. -- [Core] Stopping an executor now waits for the invocations already in flight instead of sleeping five seconds. -- [Worker] Replaced the `multiprocessing` Manager queue of the worker pool with a POSIX pipe. -- [Worker] A failed task or job process now logs its output instead of only its return code. -- [CLI] `job list`, `worker list`, `image delete` and `image list` now reject unknown flags. -- [Storage] `CloudFileProxy.walk()` now yields nothing for a missing path, like `os.walk`. -- [Storage] `cloud_open()` now raises `ValueError` on an unsupported mode. -- [Joblib] `lithops_args` is now applied to the pool that runs the batches. -- [Joblib] Capped the shared-argument upload and download pools at 32 threads. -- [Standalone] `docker login` now reads the password from stdin and quotes its arguments. -- [Localhost] A job that ended cleanly is no longer killed on cleanup, so its runner log is kept. -- [Multiprocessing] Manager proxies now follow the standard library API more closely. -- [Multiprocessing] Shared objects now refresh their expiry when read, not only when written. -- [Multiprocessing] Connection polling now backs off from 1ms instead of waiting a fixed 100ms. +- [Multiprocessing] Manager proxies now follow the standard library API more closely, and `Manager()` returns a started manager instead of the class itself. +- [Multiprocessing] Shared objects now refresh their expiry when read, not only when written, and connection polling backs off from 1ms instead of waiting a fixed 100ms. - [Multiprocessing] `imap()` and `imap_unordered()` now default to the configured chunksize. +- [Localhost] A job that ended cleanly is no longer killed on cleanup, so its runner log is kept. +- [Standalone] `docker login` now reads the password from stdin and quotes its arguments. +- [Storage] `CloudFileProxy.walk()` now yields nothing for a missing path, like `os.walk`, and `cloud_open()` raises `ValueError` on an unsupported mode. +- [Worker] Replaced the `multiprocessing` Manager queue of the worker pool with a POSIX pipe, and a failed task or job process now logs its output instead of only its return code. +- [CLI] `job list`, `worker list`, `image delete` and `image list` now reject unknown flags. +- [Joblib] `lithops_args` is now applied to the pool that runs the batches, and the shared-argument upload and download pools are capped at 32 threads. ### Fixed -- [Core] Fixed executor IDs repeating in a process whose environment is reset between executors, which made two of them share their storage keys and read back each other's results. -- [Redis] Fixed `put_object()` rejecting file-like objects, which also made `upload_file()` and `download_file()` always fail. -- [Redis] Fixed `head_object()`, which reported every key as missing on Redis 7 and up because it relied on the now-disabled `DEBUG OBJECT` command. -- [Redis] Fixed `list_objects()` returning the object bodies instead of their keys and sizes, and skip keys whose value is gone. -- [Redis] Fixed `head_bucket()` returning a bool instead of the bucket metadata, and `delete_objects()` raising on an empty list. -- [Redis] Fixed the `bytes=L-` and `bytes=-N` forms of the `Range` argument raising `ValueError`, and a ranged read of a missing key returning an empty result instead of raising. -- [Redis] `list_keys()` now walks the key space one pipelined round trip per level instead of one per directory, which cost a round trip per activation when listing a job. -- [Infinispan] Fixed an empty object being reported as a missing key, which also made `list_objects()` fail for the whole bucket. -- [Infinispan] Fixed the `bytes=L-` and `bytes=-N` forms of the `Range` argument raising `ValueError`. -- [Infinispan] Fixed `head_bucket()` raising `NotImplementedError`, and `put_object()` ignoring a failed request. -- [Infinispan] Fixed the documented `mech` config key being ignored, so `mech: BASIC` silently authenticated with DIGEST. -- [Infinispan] Fixed `list_objects()` reading every value one after the other, which made a listing cost one round trip per key. -- [Core] Fixed `wait()` on futures another executor invoked, which crashed instead of waiting for them. -- [Core] Fixed `wait()` leaving behind the monitors it started for futures of other executors. -- [Core] Fixed `wait()` with a fractional timeout raising `TypeError` from `signal.alarm()` instead of waiting. +- [Core] Fixed executor IDs repeating in a process whose environment is reset between executors. +- [Core] Fixed `wait()` watching futures of another executor with a monitor that never sees them, leaving behind the monitors it started, and raising `TypeError` from `signal.alarm()` on a fractional timeout. - [Core] Fixed `result()` returning `None` instead of re-raising when the call had already failed. -- [Core] Fixed module inspection crashing on a function whose `__module__` is `None`. -- [Core] Fixed `SerializeIndependent` appending `lithops` to the preinstalled module list on every job. -- [Core] Fixed a hand-built `FuturesList` raising `AttributeError` instead of creating its executor. +- [Core] Fixed module inspection crashing on a function whose `__module__` is `None`, and `SerializeIndependent` appending `lithops` to the preinstalled module list on every job. +- [Core] Fixed a hand-built `FuturesList` raising `AttributeError` instead of creating its executor, and pickling one detaching it from the executor it has. - [Core] Fixed `find_free_port()` setting `SO_REUSEADDR` after the bind. - [Core] Fixed `chunksize=0` and `execution_timeout=0` falling back to the config value. -- [Core] Fixed a second Ctrl+C after a failed call turning into `Error in sys.excepthook`. -- [Core] Fixed logging at interpreter shutdown raising on an already closed stream. -- [Core] Fixed `runtime_include_function` leaving the process in the build directory when the build failed. -- [Core] Fixed the function package carrying `.pytest_cache` directories and stale zips. -- [Core] Fixed a failed function package build leaving a partial zip behind. -- [Core] Fixed the Prometheus exporter raising `KeyError` when `__LITHOPS_SESSION_ID` is not set. -- [Chaining] Fixed pickling a `FuturesList` detaching the list from its executor. -- [Chaining] Fixed a list or a slice of futures of a previous job not being recognised as a chain. -- [Chaining] `extra_args` now raises at submit time instead of failing every activation of the chained job. -- [Job] Fixed a glob pattern in the object name raising `TypeError` instead of listing the objects. -- [Job] Fixed the last byte of an object being left out of its partitions. -- [Job] Fixed folder markers being counted as objects, returning empty partitions. -- [Job] Fixed a `head_object()` without `content-length` raising a bare `KeyError`. +- [Core] Fixed a second Ctrl+C after a failed call turning into `Error in sys.excepthook`, and logging at interpreter shutdown raising on an already closed stream. +- [Core] Fixed the function package carrying `.pytest_cache` directories and stale zips, a failed build leaving a partial zip behind, and `runtime_include_function` leaving the process in the build directory when the build failed. +- [Chaining] Fixed a list or a slice of futures of a previous job not being recognised as a chain, and `extra_args` failing every activation of the chained job instead of raising at submit time. +- [Job] Fixed a glob pattern in the object name raising `TypeError` instead of listing the objects, and a `head_object()` without `content-length` raising a bare `KeyError`. +- [Job] Fixed the last byte of an object being left out of its partitions, and folder markers being counted as objects, returning empty partitions. - [Monitoring] Fixed a status message lost in transit turning into a bogus timeout, or hanging `wait()` for ever. -- [Monitoring] Fixed a nested executor publishing statuses to a queue nobody declares. -- [Monitoring] Fixed failed RabbitMQ publishes being dropped with nothing in the log. +- [Monitoring] Fixed a nested executor publishing statuses to a queue nobody declares, and failed RabbitMQ publishes being dropped with nothing in the log. +- [Redis] Fixed `put_object()` rejecting file-like objects, which made `upload_file()` always fail, and `head_object()` reporting every key as missing on Redis 7 and up, where the `DEBUG OBJECT` command it relied on is disabled. +- [Redis] Fixed `list_objects()` returning the object bodies instead of their keys and sizes and not skipping keys whose value is gone, `head_bucket()` returning a bool instead of the bucket metadata, and `delete_objects()` raising on an empty list. +- [Redis] Fixed the `bytes=L-` form of the `Range` argument raising `ValueError` and `bytes=-N` returning a single byte instead of the last N, and a ranged read of a missing key returning an empty result instead of raising. +- [Redis] `list_keys()` now walks the key space one pipelined round trip per level instead of one per directory, which cost a round trip per activation when listing a job. +- [Infinispan] Fixed `list_objects()` failing for the whole bucket when one of its objects was empty, since an empty object was reported as a missing key, and reading every value one after the other, at one round trip per key. +- [Infinispan] Fixed the `bytes=L-` and `bytes=-N` forms of the `Range` argument raising `ValueError`. +- [Infinispan] Fixed `head_bucket()` raising `NotImplementedError`, and `put_object()` ignoring a failed request. +- [Infinispan] Fixed the documented `mech` config key being ignored, so `mech: BASIC` silently authenticated with DIGEST. - [Multiprocessing] Fixed `error_callback` never being called by `apply_async()`, `map_async()` and `starmap_async()`. -- [Multiprocessing] Fixed a full bounded `Queue` silently discarding what was put on it. It now waits, and raises `Full`. -- [Multiprocessing] Fixed shared list writes being dropped or misplaced through slices, `remove()`, `index()`, `pop()` and `del`. +- [Multiprocessing] Fixed a full bounded `Queue` silently discarding what was put on it, which now waits and raises `Full`, and `Queue.empty()` always saying True over a pynng connection. +- [Multiprocessing] Fixed shared list writes being dropped or misplaced through slices, `remove()`, `index()`, `pop()` and `del`, and a slice of a shared array returning one element too many. - [Multiprocessing] Fixed manager proxies not raising the `KeyError`, `ValueError` and `IndexError` the standard library raises. -- [Multiprocessing] Fixed concurrent updates to a shared object overwriting each other. +- [Multiprocessing] Fixed concurrent updates to a shared object overwriting each other, and a shared object being deleted while on its way to a worker. - [Multiprocessing] Fixed `Condition.wait()` never reporting a notify, so every wait looked like a timeout. -- [Multiprocessing] Fixed over-releasing a lock or bounded semaphore passing silently. -- [Multiprocessing] Fixed a re-entrant `RLock` giving back a token it never took. -- [Multiprocessing] Fixed `Queue.empty()` always saying True over a pynng connection. -- [Multiprocessing] Fixed a shared object being deleted while on its way to a worker. +- [Multiprocessing] Fixed over-releasing a lock or bounded semaphore passing silently, and a re-entrant `RLock` giving back a token it never took. - [Multiprocessing] Fixed `Value()` and `Array()` ignoring their `lock` argument. -- [Multiprocessing] Fixed a slice of a shared array returning one element too many. - [Multiprocessing] Fixed `lithops.multiprocessing.context` being shadowed by a context instance, which broke every `mp.context.`. - [Multiprocessing] Fixed closing one connection closing the Redis client the whole process shares. -- [Multiprocessing] Fixed a closed `Pool` leaving the monitor and invoker threads of its executor running. +- [Multiprocessing] Fixed a closed `Pool` leaving the monitor and invoker threads of its executor running, and the remote log feed keeping the interpreter alive at exit. - [Multiprocessing] Fixed `AsyncResult.get()` raising the builtin `TimeoutError` instead of `multiprocessing.TimeoutError`. -- [Multiprocessing] Fixed `current_process()` in a worker creating an executor and a Redis client just to read a name. -- [Multiprocessing] Fixed `set_parameter()` rewriting the defaults it falls back to. -- [Multiprocessing] Fixed the remote log feed keeping the interpreter alive at exit. -- [Localhost] Fixed a job cleared mid-task leaving a latch closed, spinning the v2 job manager on a full core. +- [Multiprocessing] Fixed `current_process()` in a worker creating an executor and a Redis client just to read a name, and `set_parameter()` rewriting the defaults it falls back to. +- [Localhost] Fixed the v2 job manager spinning a full core: on a job cleared mid-task that left a latch closed, and while an invocation was queueing. - [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. -- [Localhost] Fixed the v2 job manager spinning a core while an invocation was queueing. -- [Localhost] Fixed two concurrent `invoke()` calls clearing each other's in-progress flag. +- [Localhost] Fixed a task starting after `stop()`, leaving a process nobody kills, and two concurrent `invoke()` calls clearing each other's in-progress flag. - [Localhost] Fixed the v2 container being removed while other jobs were still running in it. -- [Localhost] Fixed v1 and v2 sharing one runner file, so a job could run under the other version's runner. +- [Localhost] Fixed v1 and v2 sharing one runner file, so a job could run under the other version's runner, and the runner exiting with success on an unknown command or a crash. - [Localhost] Fixed a container image whose name starts with `python`, such as `python:3.12`, being run as a local interpreter. -- [Localhost] Fixed the runner exiting with success on an unknown command or a crash. - [Standalone] Fixed a dict race that killed the budget keeper and left the VM running. - [Standalone] Fixed a file descriptor leak of the runner log, one per task. -- [Standalone] Fixed the worker `/stop` endpoint iterating the process map while it changed. -- [Standalone] Fixed `cancel_job_process()` raising on an emptied queue or a job with no queue. -- [Standalone] Fixed the master dropping the errors of its parallel worker and job requests. -- [Standalone] Fixed the SSH client keeping a client that failed to connect. -- [Standalone] Fixed the SSH client rejecting every private key that is not RSA. +- [Standalone] Fixed the worker `/stop` endpoint iterating the process map while it changed, and `cancel_job_process()` raising on an emptied queue or a job with no queue. +- [Standalone] Fixed the master dropping the errors of its parallel worker and job requests, and a failed consume-mode worker setup script passing unnoticed. +- [Standalone] Fixed the SSH client keeping a client that failed to connect, and rejecting every private key that is not RSA. - [Standalone] Fixed a reuse-mode worker blocking for ever on a stale queue connection instead of taking the next job. -- [Standalone] Fixed a failed consume-mode worker setup script passing unnoticed. - [Standalone] Fixed the worker service running a `python:*` container image with the local interpreter. -- [Storage] Fixed `delete_cloudobjects()` deleting part of the list before rejecting a foreign object. -- [Storage] Fixed `delete_cloudobjects()` deleting the keys of one bucket from another when the objects spanned several. +- [Storage] Fixed `delete_cloudobjects()` deleting the keys of one bucket from another when the objects spanned several, and deleting part of the list before rejecting a foreign object. - [Storage] Fixed `CloudFileProxy.listdir()` returning nothing for its default argument. -- [Worker] Fixed the memory monitor reporting a peak of zero where usage cannot be read. - [Worker] Fixed the remote invoker returning before its invocations in flight were done. -- [Worker] Fixed the function process being aborted on macOS from the second call on, by setting Apple's fork-safety flag. -- [Worker] Fixed a function process killed by the OOM killer, or by any signal, being reported as a missing result. +- [Worker] Fixed the function process being aborted on macOS from the second call on, by setting Apple's fork-safety flag, and one killed by the OOM killer, or by any signal, being reported as a missing result. - [Azure] Fixed the `az` CLI calls deadlocking when a command filled the stderr pipe. -- [Azure Containers] Fixed a deploy racing a provisioning operation already in progress. -- [Azure Containers] Fixed a container app left in `Failed` state never being recreated. +- [Azure Containers] Fixed a deploy racing a provisioning operation already in progress, and a container app left in `Failed` state never being recreated. - [CLI] Fixed `job list` and `worker list` crashing when there was nothing to list. - [CLI] Fixed `lithops clean` deleting the local temp directory of the jobs running at the same time on the same machine. -- [Cleaner] Fixed two cleaners racing for the pid file. -- [Cleaner] Fixed the cleaner skipping the requests it was started for. -- [Cleaner] Fixed the cleaner reading a request another process was still writing. -- [Cleaner] Fixed the cleaner looping forever on a request it could not read or classify. +- [Cleaner] Fixed two cleaners racing for the pid file, and the cleaner skipping the requests it was started for. +- [Cleaner] Fixed the cleaner reading a request another process was still writing, and looping forever on one it could not read or classify. - [Joblib] Fixed the backend being unused with joblib 1.4+, which renamed `apply_async` to `submit`. -- [Joblib] Fixed a `KeyError` on shared arguments over 32KB, from a check-then-read on the disk cache. -- [Joblib] Fixed shared arguments going to the default storage instead of the configured one. -- [Joblib] Fixed a race losing one of two shared arguments proxied in the same call. +- [Joblib] Fixed shared arguments going to the default storage instead of the configured one, a `KeyError` on those over 32KB from a check-then-read on the disk cache, and a race losing one of two proxied in the same call. - [Joblib] Fixed `lithops[joblib]` missing `redis`, needed to import the backend. - [IBM] Fixed the COS token manager raising if `ibm_botocore` hides the private expiry attribute. +### Removed + +- [Storage] Removed the `infinispan_hotrod` storage backend. + ## [v3.7.0] ### Added diff --git a/lithops/constants.py b/lithops/constants.py index 9062a6046..3de6b2421 100644 --- a/lithops/constants.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/lithops/job/job.py b/lithops/job/job.py index 7684e9772..fef84454b 100644 --- a/lithops/job/job.py +++ b/lithops/job/job.py @@ -35,7 +35,12 @@ 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 ) @@ -233,7 +238,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) diff --git a/lithops/multiprocessing/process.py b/lithops/multiprocessing/process.py index eb907223c..32287f186 100644 --- a/lithops/multiprocessing/process.py +++ b/lithops/multiprocessing/process.py @@ -20,6 +20,7 @@ import multiprocessing as _mp from lithops import FunctionExecutor +from lithops.constants import SESSION_ID_ENV from lithops.utils import is_lithops_worker from . import config as mp_config from . import util @@ -77,7 +78,7 @@ def current_process(): if is_lithops_worker(): return _CurrentProcess( name=os.environ.get('LITHOPS_MP_WORKER_NAME'), - pid=os.environ.get('__LITHOPS_SESSION_ID', '-1'), + pid=os.environ.get(SESSION_ID_ENV, '-1'), ) else: return _mp.current_process() @@ -130,7 +131,7 @@ def cloud_process_wrapper(data, func, initializer=None, initargs=(), name=None, # Print exception stack trace to remote logging buffer header = "---------- {} at {} ({}) ----------".format(e.__class__.__name__, os.environ.get('LITHOPS_MP_WORKER_NAME'), - os.environ.get('__LITHOPS_SESSION_ID')) + os.environ.get(SESSION_ID_ENV)) exception_body = traceback.format_exc() footer = '-' * len(header) if remote_log_buff: diff --git a/lithops/storage/storage.py b/lithops/storage/storage.py index 50f568cf5..d55855bdb 100644 --- a/lithops/storage/storage.py +++ b/lithops/storage/storage.py @@ -22,7 +22,13 @@ import importlib from typing import Optional, List, Union, Dict, TextIO, BinaryIO, Any, Iterable -from lithops.constants import CACHE_DIR, RUNTIMES_PREFIX, JOBS_PREFIX, TEMP_PREFIX +from lithops.constants import ( + CACHE_DIR, + JOBS_PREFIX, + RUNTIMES_PREFIX, + SESSION_ID_ENV, + TEMP_PREFIX, +) from lithops.utils import is_lithops_worker from lithops.storage import utils from lithops.config import extract_storage_config, default_storage_config @@ -271,7 +277,7 @@ def put_cloudobject(self, body: Union[str, bytes, TextIO, BinaryIO], :return: CloudObject instance """ - prefix = os.environ.get('__LITHOPS_SESSION_ID', '') + prefix = os.environ.get(SESSION_ID_ENV, '') coid = hex(next(COBJECTS_INDEX))[2:] coname = f'cloudobject_{coid}' name = '/'.join([prefix, coname]) if prefix else coname diff --git a/lithops/telemetry/exporter.py b/lithops/telemetry/exporter.py index 45e32351b..f70d0f51d 100644 --- a/lithops/telemetry/exporter.py +++ b/lithops/telemetry/exporter.py @@ -35,6 +35,7 @@ from typing import Any, Dict, Optional from lithops.telemetry import metrics as M +from lithops.constants import WORKER_ENV from lithops.version import __version__ from lithops.telemetry.backends import ( import_backend_module, @@ -548,7 +549,7 @@ def get_telemetry( if backend_name is None: return NOOP - if 'LITHOPS_WORKER' in os.environ: + if WORKER_ENV in os.environ: # Metrics are a client concern, and a worker gets the client's # whole config. Without this, every worker would try to reach a # metrics system it most likely has no route to. The remote diff --git a/lithops/tests/test_constants.py b/lithops/tests/test_constants.py index dae790c67..1ba0d8e7a 100644 --- a/lithops/tests/test_constants.py +++ b/lithops/tests/test_constants.py @@ -105,6 +105,19 @@ def test_default_config_keys(self): assert constants.LITHOPS_DEFAULT_CONFIG_KEYS['telemetry'] is False assert constants.LITHOPS_DEFAULT_CONFIG_KEYS['telemetry_interval'] == 10 + def test_executor_environment_variable_names(self): + """ + These names are read by runtimes built from earlier Lithops + versions, so changing one is a breaking change and not a rename. + Pinned here, in the one place that does not go through the + constant, so that changing a value fails rather than passing + """ + assert constants.SESSION_ID_ENV == '__LITHOPS_SESSION_ID' + assert constants.TOTAL_EXECUTORS_ENV == '__LITHOPS_TOTAL_EXECUTORS' + assert constants.MONITORING_QUEUES_ENV == '__LITHOPS_MONITORING_QUEUES' + assert constants.WORKER_ENV == 'LITHOPS_WORKER' + assert constants.REDUCE_JOB_ENV == '__LITHOPS_REDUCE_JOB' + def test_get_mode_and_default_backend_round_trip(self): assert get_mode(constants.LOCALHOST) == constants.LOCALHOST assert get_mode(constants.SERVERLESS_BACKEND_DEFAULT) == constants.SERVERLESS diff --git a/lithops/tests/test_invokers.py b/lithops/tests/test_invokers.py index adb16a151..20a076f64 100644 --- a/lithops/tests/test_invokers.py +++ b/lithops/tests/test_invokers.py @@ -23,7 +23,7 @@ import pytest -from lithops.constants import LOCALHOST, SERVERLESS +from lithops.constants import LOCALHOST, SERVERLESS, WORKER_ENV from lithops.future import ResponseFuture from lithops.invokers import ( BatchInvoker, @@ -608,7 +608,7 @@ def test_init_reads_pool_threads_and_remote_invoker(self): inv.executor.shutdown(wait=False) def test_init_disables_remote_invoker_inside_worker(self, monkeypatch): - monkeypatch.setenv('LITHOPS_WORKER', '1') + monkeypatch.setenv(WORKER_ENV, '1') inv = FaaSInvoker( self._config(remote_invoker=True), 'sess-0', diff --git a/lithops/tests/test_job.py b/lithops/tests/test_job.py index 44b155352..65161de3e 100644 --- a/lithops/tests/test_job.py +++ b/lithops/tests/test_job.py @@ -22,7 +22,13 @@ import pytest -from lithops.constants import LOCALHOST, MAX_AGG_DATA_SIZE, SERVERLESS, STANDALONE +from lithops.constants import ( + LOCALHOST, + MAX_AGG_DATA_SIZE, + REDUCE_JOB_ENV, + SERVERLESS, + STANDALONE, +) from lithops.job import create_map_job, create_reduce_job from lithops.job.job import ( FUNCTION_CACHE, @@ -397,7 +403,7 @@ def test_single_iterdata_without_parts(self): ) assert job.job_id == 'r0' assert job.total_calls == 1 - assert job.extra_env['__LITHOPS_REDUCE_JOB'] == 'True' + assert job.extra_env[REDUCE_JOB_ENV] == 'True' _, _, objs = _CapturingSerializer.last assert objs[1]['results'] == list(range(6)) @@ -419,7 +425,7 @@ def test_reduce_by_key_slices_futures(self): ) assert job.total_calls == 3 assert job.extra_env['K'] == 'False' - assert job.extra_env['__LITHOPS_REDUCE_JOB'] == 'True' + assert job.extra_env[REDUCE_JOB_ENV] == 'True' _, _, objs = _CapturingSerializer.last assert [o['results'] for o in objs[1:]] == [ ['a', 'b'], ['c', 'd', 'e'], ['f'] diff --git a/lithops/tests/test_storage_layer.py b/lithops/tests/test_storage_layer.py index 99551fb98..2c22b13a8 100644 --- a/lithops/tests/test_storage_layer.py +++ b/lithops/tests/test_storage_layer.py @@ -19,7 +19,12 @@ import pytest -from lithops.constants import JOBS_PREFIX, RUNTIMES_PREFIX, TEMP_PREFIX +from lithops.constants import ( + JOBS_PREFIX, + RUNTIMES_PREFIX, + SESSION_ID_ENV, + TEMP_PREFIX, +) from lithops.storage.cloud_proxy import ( CloudFileProxy, CloudStorage, @@ -165,7 +170,7 @@ def sleep(seconds): class TestStorageCloudObjects: def test_put_cloudobject_uses_temp_prefix_and_hex_id(self, monkeypatch): - monkeypatch.delenv('__LITHOPS_SESSION_ID', raising=False) + monkeypatch.delenv(SESSION_ID_ENV, raising=False) storage = _bare_storage() cloudobject = storage.put_cloudobject(b'data') key = storage.storage_handler.put_object.call_args[0][1] @@ -175,7 +180,7 @@ def test_put_cloudobject_uses_temp_prefix_and_hex_id(self, monkeypatch): assert cloudobject.bucket == 'storage' def test_put_cloudobject_prefixes_session_id(self, monkeypatch): - monkeypatch.setenv('__LITHOPS_SESSION_ID', 'sess') + monkeypatch.setenv(SESSION_ID_ENV, 'sess') storage = _bare_storage() cloudobject = storage.put_cloudobject(b'data') key = storage.storage_handler.put_object.call_args[0][1] diff --git a/lithops/tests/test_telemetry.py b/lithops/tests/test_telemetry.py index 931c60143..f3ef63eef 100644 --- a/lithops/tests/test_telemetry.py +++ b/lithops/tests/test_telemetry.py @@ -18,6 +18,7 @@ import pytest +from lithops.constants import WORKER_ENV from lithops.telemetry import metrics as M from lithops.telemetry.backend import MetricsBackend from lithops.telemetry.backends import DEFAULT_BACKEND, resolve_backend @@ -727,7 +728,7 @@ def test_one_process_exports_through_one_exporter(self): def test_a_worker_exports_nothing(self, monkeypatch): # Workers get the client's whole config, and the remote invoker # builds a JobMonitor of its own - monkeypatch.setenv('LITHOPS_WORKER', 'True') + monkeypatch.setenv(WORKER_ENV, 'True') config = {'lithops': {'telemetry': 'prometheus'}} assert get_telemetry(config) is NOOP diff --git a/lithops/tests/test_utils.py b/lithops/tests/test_utils.py index 38f7dd73b..579a9ae60 100644 --- a/lithops/tests/test_utils.py +++ b/lithops/tests/test_utils.py @@ -24,6 +24,12 @@ import pytest from lithops import constants +from lithops.constants import ( + MONITORING_QUEUES_ENV, + SESSION_ID_ENV, + TOTAL_EXECUTORS_ENV, + WORKER_ENV, +) from lithops import utils as lithops_utils from lithops.utils import ( CountDownLatch, @@ -56,7 +62,6 @@ is_podman, is_unix_system, iterchunks, - MONITORING_QUEUES_ENV, monitoring_queue_name, monitoring_queues, log_prefix, @@ -252,9 +257,9 @@ def test_convert_bools_to_string_mutates_in_place(self): assert env == {'flag': 'True', 'count': 1, 'name': 'x'} def test_is_lithops_worker(self, monkeypatch): - monkeypatch.delenv('LITHOPS_WORKER', raising=False) + monkeypatch.delenv(WORKER_ENV, raising=False) assert is_lithops_worker() is False - monkeypatch.setenv('LITHOPS_WORKER', '1') + monkeypatch.setenv(WORKER_ENV, '1') assert is_lithops_worker() is True def test_version_str_and_current_py_version(self): @@ -363,8 +368,8 @@ def _fresh_session(monkeypatch): """A process that has not created an executor yet""" monkeypatch.setattr(lithops_utils, '_SESSION_ID', None) monkeypatch.setattr(lithops_utils, '_EXECUTOR_COUNT', 0) - monkeypatch.delenv('__LITHOPS_SESSION_ID', raising=False) - monkeypatch.delenv('__LITHOPS_TOTAL_EXECUTORS', raising=False) + monkeypatch.delenv(SESSION_ID_ENV, raising=False) + monkeypatch.delenv(TOTAL_EXECUTORS_ENV, raising=False) def test_create_executor_id_reuses_session_and_increments(self, monkeypatch): self._fresh_session(monkeypatch) @@ -375,65 +380,49 @@ def test_create_executor_id_reuses_session_and_increments(self, monkeypatch): assert num == '0' assert second == f'{session}-1' assert get_executor_id() == second - - def test_create_executor_id_exports_the_session(self, monkeypatch): - self._fresh_session(monkeypatch) - executor_id = create_executor_id() - session, num = executor_id.rsplit('-', 1) - - # Exported so that a process spawned from this one joins the session - assert os.environ['__LITHOPS_SESSION_ID'] == session - assert os.environ['__LITHOPS_TOTAL_EXECUTORS'] == num + # Exported so a process spawned from this one joins the session + assert os.environ[SESSION_ID_ENV] == session + assert os.environ[TOTAL_EXECUTORS_ENV] == '1' def test_create_executor_id_survives_a_reset_environment(self, monkeypatch): - """ - The counter is what tells two executors of one process apart, and - it used to live only in the environment. Anything that saves and - restores os.environ -- this test suite does, between every test -- - put the counter back to zero, so a process kept handing out the - same ID with nothing but six random characters keeping the storage - keys of one executor apart from the next - """ + # The counter tells two executors of one process apart and used to + # live only in the environment. Anything that saves and restores + # os.environ -- this suite does, between every test -- put it back to + # zero, and the executor IDs repeated self._fresh_session(monkeypatch) ids = [] for _ in range(4): ids.append(create_executor_id()) - os.environ.pop('__LITHOPS_SESSION_ID', None) - os.environ.pop('__LITHOPS_TOTAL_EXECUTORS', None) + os.environ.pop(SESSION_ID_ENV, None) + os.environ.pop(TOTAL_EXECUTORS_ENV, None) - assert len(set(ids)) == len(ids) sessions = {executor_id.rsplit('-', 1)[0] for executor_id in ids} assert len(sessions) == 1 - assert [executor_id.rsplit('-', 1)[1] for executor_id in ids] == [ - '0', '1', '2', '3' - ] + assert [i.rsplit('-', 1)[1] for i in ids] == ['0', '1', '2', '3'] def test_create_executor_id_joins_a_session_it_is_given(self, monkeypatch): - # What a worker does before it runs a task, and what the remote - # invoker does for the job it spawns + # What a worker does before running a task: name a session and clear + # the count, as function_handler() does self._fresh_session(monkeypatch) - create_executor_id() + monkeypatch.setenv(SESSION_ID_ENV, 'job-key-00000') - monkeypatch.setenv('__LITHOPS_SESSION_ID', 'job-key-00000') assert create_executor_id() == 'job-key-00000-0' - monkeypatch.setenv('__LITHOPS_SESSION_ID', 'job-key-00001') + monkeypatch.setenv(SESSION_ID_ENV, 'job-key-00001') + monkeypatch.delenv(TOTAL_EXECUTORS_ENV) assert create_executor_id() == 'job-key-00001-0' def test_create_executor_id_continues_the_count_it_inherits(self, monkeypatch): # A process spawned by one that had already created eight executors self._fresh_session(monkeypatch) - monkeypatch.setenv('__LITHOPS_SESSION_ID', 'parent') - monkeypatch.setenv('__LITHOPS_TOTAL_EXECUTORS', '7') - + monkeypatch.setenv(SESSION_ID_ENV, 'parent') + monkeypatch.setenv(TOTAL_EXECUTORS_ENV, '7') assert create_executor_id() == 'parent-8' - assert create_executor_id() == 'parent-9' - def test_create_executor_id_ignores_an_unreadable_count(self, monkeypatch): + # A count that cannot be read is no count at all self._fresh_session(monkeypatch) - monkeypatch.setenv('__LITHOPS_SESSION_ID', 'parent') - monkeypatch.setenv('__LITHOPS_TOTAL_EXECUTORS', 'not-a-number') - + monkeypatch.setenv(SESSION_ID_ENV, 'parent') + monkeypatch.setenv(TOTAL_EXECUTORS_ENV, 'not-a-number') assert create_executor_id() == 'parent-0' def test_create_executor_id_is_unique_under_concurrency(self, monkeypatch): diff --git a/lithops/tests/test_worker.py b/lithops/tests/test_worker.py index 584731812..4fe36a5a5 100644 --- a/lithops/tests/test_worker.py +++ b/lithops/tests/test_worker.py @@ -28,7 +28,14 @@ import pytest import lithops.worker.handler as handler_module -from lithops.constants import JOBS_PREFIX, MODULES_DIR +from lithops.constants import ( + JOBS_PREFIX, + MODULES_DIR, + REDUCE_JOB_ENV, + SESSION_ID_ENV, + TOTAL_EXECUTORS_ENV, + WORKER_ENV, +) from lithops.storage.utils import CloudObject, CloudObjectLocal, CloudObjectUrl from lithops.utils import bytes_to_b64str, is_unix_system from lithops.worker import function_handler, function_invoker @@ -354,18 +361,18 @@ def test_removes_module_path_and_total_executors(self): job = _task(worker_processes=1, call_ids=['00000'], data=[b'd']) module_path = os.path.join(MODULES_DIR, job.job_key) sys.path.append(module_path) - os.environ['__LITHOPS_TOTAL_EXECUTORS'] = '2' + os.environ[TOTAL_EXECUTORS_ENV] = '2' try: with patch('lithops.worker.handler.create_job', return_value=job): with patch('lithops.worker.handler.setup_lithops_logger'): with patch('lithops.worker.handler.task_consumer'): function_handler({}) assert module_path not in sys.path - assert '__LITHOPS_TOTAL_EXECUTORS' not in os.environ + assert TOTAL_EXECUTORS_ENV not in os.environ finally: if module_path in sys.path: sys.path.remove(module_path) - os.environ.pop('__LITHOPS_TOTAL_EXECUTORS', None) + os.environ.pop(TOTAL_EXECUTORS_ENV, None) class TestPrepareAndRunTask: @@ -382,7 +389,7 @@ def test_sets_env_creates_dir_and_clears_extra_env( with patch('lithops.worker.handler.run_task') as run: prepare_and_run_task(task) run.assert_called_once_with(task) - assert os.environ['LITHOPS_WORKER'] == 'True' + assert os.environ[WORKER_ENV] == 'True' assert os.environ['PYTHONUNBUFFERED'] == 'True' assert 'FOO' not in os.environ assert os.path.isdir(task.task_dir) @@ -592,7 +599,7 @@ def test_does_not_mutate_extra_env_with_session_id(self, tmp_path): conn.poll.return_value = True self._patch_run(task, jrp, conn) assert extra == {} - assert '__LITHOPS_SESSION_ID' not in extra + assert SESSION_ID_ENV not in extra assert 'LITHOPS_CONFIG' not in extra @@ -1128,7 +1135,7 @@ class TestJobRunner: @pytest.fixture(autouse=True) def _session_id(self, monkeypatch, tmp_path): - monkeypatch.setenv('__LITHOPS_SESSION_ID', 'sid-1') + monkeypatch.setenv(SESSION_ID_ENV, 'sid-1') self.stats = str(tmp_path / 'stats.txt') def _runner(self, func, data, **job_kwargs): @@ -1298,7 +1305,7 @@ def test_new_futures_list_skips_result_upload(self): jr.internal_storage.put_data.assert_not_called() def test_reduce_job_waits_for_futures(self, monkeypatch): - monkeypatch.setenv('__LITHOPS_REDUCE_JOB', 'True') + monkeypatch.setenv(REDUCE_JOB_ENV, 'True') jr = self._runner(_reduce_fn, {'results': []}) with patch.object(jr, '_wait_futures') as wait_f: jr.run() @@ -1345,7 +1352,7 @@ def test_callable_class_function_name(self): class TestFunctionInvoker: def test_function_invoker_wires_handlers(self, monkeypatch): - monkeypatch.delenv('LITHOPS_WORKER', raising=False) + monkeypatch.delenv(WORKER_ENV, raising=False) payload = { 'config': _job_config( monitoring='storage', backend='aws_lambda' @@ -1375,7 +1382,7 @@ def test_function_invoker_wires_handlers(self, monkeypatch): ): function_invoker(payload) invoker.run_job.assert_called_once() - assert os.environ['LITHOPS_WORKER'] == 'True' + assert os.environ[WORKER_ENV] == 'True' assert payload['config']['aws_lambda']['invoke_pool_threads'] == 128 def test_the_remote_invoker_takes_a_queue_of_its_own(self, monkeypatch): @@ -1384,14 +1391,14 @@ def test_the_remote_invoker_takes_a_queue_of_its_own(self, monkeypatch): the client's queue: the statuses it took would never reach the client. It reads its own, and the calls report to both """ + from lithops.constants import MONITORING_QUEUES_ENV from lithops.utils import ( - MONITORING_QUEUES_ENV, monitoring_queue_name, monitoring_queues, remote_invoker_queue_name, ) - monkeypatch.delenv('LITHOPS_WORKER', raising=False) + monkeypatch.delenv(WORKER_ENV, raising=False) monkeypatch.delenv(MONITORING_QUEUES_ENV, raising=False) payload = { 'config': _job_config( diff --git a/lithops/utils.py b/lithops/utils.py index d92dec8c9..d85414aae 100644 --- a/lithops/utils.py +++ b/lithops/utils.py @@ -78,83 +78,51 @@ def _future_id(fut): return (fut.executor_id, fut.job_id, fut.call_id) -#: The session this process belongs to, and how many executors it has -#: created. Held here rather than only in the environment: the counter is -#: what makes two executor IDs of one process different, and an -#: environment can be rewritten under a running process +# The session this process belongs to, and how many executors it has created. +# Kept here and not only in the environment, which can be reset under a +# running process _SESSION_LOCK = threading.Lock() _SESSION_ID = None _EXECUTOR_COUNT = 0 -SESSION_ID_ENV = '__LITHOPS_SESSION_ID' -TOTAL_EXECUTORS_ENV = '__LITHOPS_TOTAL_EXECUTORS' - - -def _inherited_executor_count(): - """ - The number the session that this process joined had reached, so that - the executors created here carry on from it instead of repeating it - """ - try: - return int(os.environ[TOTAL_EXECUTORS_ENV]) + 1 - except (KeyError, TypeError, ValueError): - return 0 - def create_executor_id(lenght=6): """ - Creates the ID of a new executor. Executors of the same session share - the session ID and are told apart by a counter. - - Both are exported to the environment, so that a process spawned from - this one joins the same session, and a worker can put a session of its - own there for the task it is about to run. They are *read* back only - when the environment names a session other than the one this process - is already in. - - That last part is the difference between an ID that is unique and one - that is merely unlikely to repeat. The counter is what tells two - executors of the same process apart; a process that lost the variable - -- anything that saves and restores ``os.environ`` does that -- used - to start counting from zero again, which left the six characters of - the session ID as the only thing keeping its storage keys apart from - those of every executor it had already created. + Creates the ID of a new executor. Executors of the same session share the + session ID and are told apart by a counter. + + Both are exported, so that a process spawned from this one joins the same + session and a worker can name a session for the task it runs. The counter + is also kept in the process: an environment that is restored under a + running process used to restart it, and two executors then shared an ID, + and with it their storage keys. """ global _SESSION_ID, _EXECUTOR_COUNT with _SESSION_LOCK: - env_session = os.environ.get(SESSION_ID_ENV) - if _SESSION_ID is None: - # The first executor of this process. It joins the session the - # process that spawned this one left in the environment, - # carrying on from wherever that had got to, or starts one - if env_session: - _SESSION_ID = env_session - _EXECUTOR_COUNT = _inherited_executor_count() - else: - _SESSION_ID = uuid_str().replace('/', '')[:lenght] + session_id = os.environ.get(constants.SESSION_ID_ENV) + if session_id and session_id != _SESSION_ID: + # A session named for this process, by the one that spawned it or + # by a worker about to run a task. Carry on from where it got to + _SESSION_ID = session_id + try: + _EXECUTOR_COUNT = int( + os.environ[constants.TOTAL_EXECUTORS_ENV] + ) + 1 + except (KeyError, ValueError): _EXECUTOR_COUNT = 0 - elif env_session and env_session != _SESSION_ID: - # The process has been put in a different session, which is - # what a worker does before each task it runs. The count of - # the session it left does not carry over - _SESSION_ID = env_session - _EXECUTOR_COUNT = 0 + elif _SESSION_ID is None: + _SESSION_ID = uuid_str().replace('/', '')[:lenght] exec_num = _EXECUTOR_COUNT _EXECUTOR_COUNT += 1 - os.environ[SESSION_ID_ENV] = _SESSION_ID - os.environ[TOTAL_EXECUTORS_ENV] = str(exec_num) + os.environ[constants.SESSION_ID_ENV] = _SESSION_ID + os.environ[constants.TOTAL_EXECUTORS_ENV] = str(exec_num) return f'{_SESSION_ID}-{exec_num}' -# Carries the monitoring queues of an executor down to the workers, so that an -# executor created inside one of them can extend the chain -MONITORING_QUEUES_ENV = '__LITHOPS_MONITORING_QUEUES' - - def monitoring_queue_name(executor_id: str) -> str: """Returns the name of the queue an executor is monitored through""" return f'lithops-{executor_id}' @@ -182,13 +150,14 @@ def monitoring_queues(executor_id: str) -> List[str]: job, so the same number of tokens can stand for different chains """ parent_queues = [] - raw_queues = os.environ.get(MONITORING_QUEUES_ENV) + raw_queues = os.environ.get(constants.MONITORING_QUEUES_ENV) if raw_queues: try: parent_queues = list(json.loads(raw_queues)) except ValueError: logger.warning( - f'Ignoring a malformed {MONITORING_QUEUES_ENV}: {raw_queues}' + 'Ignoring a malformed ' + f'{constants.MONITORING_QUEUES_ENV}: {raw_queues}' ) queue = monitoring_queue_name(executor_id) @@ -201,18 +170,10 @@ def monitoring_queues(executor_id: str) -> List[str]: def get_executor_id(): - """ - Returns the ID of the last executor created in this session. - - Read from this process rather than from the environment, so that it - still answers after something has rewritten the environment, and - raises the same KeyError as it always did when no executor has been - created at all - """ - with _SESSION_LOCK: - if _SESSION_ID is None: - raise KeyError(SESSION_ID_ENV) - return f'{_SESSION_ID}-{_EXECUTOR_COUNT - 1}' + """Returns the ID of the last executor created in this session""" + if _SESSION_ID is None: + raise KeyError(constants.SESSION_ID_ENV) + return f'{_SESSION_ID}-{_EXECUTOR_COUNT - 1}' def iterchunks(lst, n): @@ -565,7 +526,7 @@ def is_linux_system() -> bool: def is_lithops_worker() -> bool: """Checks if the current execution is within a lithops worker""" - return 'LITHOPS_WORKER' in os.environ + return constants.WORKER_ENV in os.environ def is_object_processing_function(map_function) -> bool: diff --git a/lithops/worker/handler.py b/lithops/worker/handler.py index 2020f6a6a..fab3b8c7d 100644 --- a/lithops/worker/handler.py +++ b/lithops/worker/handler.py @@ -43,9 +43,16 @@ LogStream, custom_redirection, get_function_and_modules, get_function_data, SystemMonitor ) -from lithops.constants import JOBS_PREFIX, LITHOPS_TEMP_DIR, MODULES_DIR -from lithops.utils import ( +from lithops.constants import ( + JOBS_PREFIX, + LITHOPS_TEMP_DIR, + MODULES_DIR, MONITORING_QUEUES_ENV, + SESSION_ID_ENV, + TOTAL_EXECUTORS_ENV, + WORKER_ENV, +) +from lithops.utils import ( setup_lithops_logger, is_unix_system, ) @@ -237,7 +244,7 @@ def function_handler(payload: Dict[str, Any]) -> None: if module_path in sys.path: sys.path.remove(module_path) - os.environ.pop('__LITHOPS_TOTAL_EXECUTORS', None) + os.environ.pop(TOTAL_EXECUTORS_ENV, None) def task_consumer( @@ -296,7 +303,7 @@ def prepare_and_run_task(task: SimpleNamespace) -> None: act_id = str(uuid.uuid4()).replace('-', '')[:12] os.environ['__LITHOPS_ACTIVATION_ID'] = act_id - os.environ['LITHOPS_WORKER'] = 'True' + os.environ[WORKER_ENV] = 'True' os.environ['PYTHONUNBUFFERED'] = 'True' os.environ.update(task.extra_env) @@ -461,7 +468,7 @@ def run_task(task: SimpleNamespace) -> None: injected_env = { 'LITHOPS_CONFIG': json.dumps(task.config), - '__LITHOPS_SESSION_ID': '-'.join([task.job_key, task.call_id]), + SESSION_ID_ENV: '-'.join([task.job_key, task.call_id]), # An executor created by the user function reports to these queues as # well as to its own, which is how a nested job reaches the client MONITORING_QUEUES_ENV: json.dumps( diff --git a/lithops/worker/invoker.py b/lithops/worker/invoker.py index b9e4f401b..c02de9109 100644 --- a/lithops/worker/invoker.py +++ b/lithops/worker/invoker.py @@ -20,13 +20,17 @@ from types import SimpleNamespace from typing import Any, Dict +from lithops.constants import ( + MONITORING_QUEUES_ENV, + SESSION_ID_ENV, + WORKER_ENV, +) from lithops.serverless import ServerlessHandler from lithops.monitoring import JobMonitor from lithops.storage import InternalStorage from lithops.config import extract_serverless_config, extract_storage_config from lithops.invokers import FaaSInvoker from lithops.utils import ( - MONITORING_QUEUES_ENV, monitoring_queues, remote_invoker_queue_name, ) @@ -50,9 +54,9 @@ def function_invoker(job_payload: Dict[str, Any]) -> None: invoker_queue = remote_invoker_queue_name(job.executor_id) os.environ.update({ - 'LITHOPS_WORKER': 'True', + WORKER_ENV: 'True', 'PYTHONUNBUFFERED': 'True', - '__LITHOPS_SESSION_ID': job.job_key, + SESSION_ID_ENV: job.job_key, # The job this invoker spawns reports to the queues of the client and # to this invoker's own, and an executor created here extends that # chain rather than replacing it diff --git a/lithops/worker/jobrunner.py b/lithops/worker/jobrunner.py index 5557604d7..5ec190c7d 100644 --- a/lithops/worker/jobrunner.py +++ b/lithops/worker/jobrunner.py @@ -39,6 +39,7 @@ except ModuleNotFoundError: pass +from lithops.constants import REDUCE_JOB_ENV from lithops.storage import Storage from lithops.wait import wait from lithops.future import ResponseFuture @@ -364,7 +365,7 @@ def run(self) -> None: func = pickle.loads(self.job.func) data = pickle.loads(self.job.data) - if ast.literal_eval(os.environ.get('__LITHOPS_REDUCE_JOB', 'False')): + if ast.literal_eval(os.environ.get(REDUCE_JOB_ENV, 'False')): self._wait_futures(data) elif is_object_processing_function(func): self._load_object(data) From 0fd6d7e3a153258685e31fdfa069e56e0984d461 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Wed, 9 Sep 2026 14:39:50 +0200 Subject: [PATCH 03/10] Fix issues --- CHANGELOG.md | 10 +- lithops/concurrent/futures.py | 54 +++++---- lithops/constants.py | 1 - lithops/executors.py | 6 +- lithops/job/job.py | 59 +--------- .../monitoring/backends/storage/storage.py | 2 +- lithops/monitoring/monitor.py | 76 ++++++++++--- lithops/multiprocessing/synchronize.py | 59 ++++++++-- lithops/tests/test_concurrent_futures.py | 104 ++++++++++++++++++ lithops/tests/test_constants.py | 25 +++++ lithops/tests/test_job.py | 58 ++++++++-- lithops/tests/test_map.py | 13 +++ lithops/tests/test_monitor.py | 80 ++++++++++++++ lithops/tests/test_multiprocessing.py | 55 +++++++++ lithops/tests/test_worker.py | 71 ++++++++---- lithops/worker/handler.py | 51 ++++++--- 16 files changed, 571 insertions(+), 153 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 74ec8a725..36e35c17e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,6 @@ - [API] Added `lithops.concurrent.futures`, a `concurrent.futures`-compatible executor interface backed by Lithops. - [Monitoring] Added Redis, AWS SQS, GCP Pub/Sub and Azure Queue Storage monitoring backends. -- [Core] Added a cache of serialized functions to avoid re-uploading the same function. - [Core] Added `clean_jobs` to `wait()`, to keep the temporary data until the results are read. - [AWS Batch] Added the `instance_types` config option for EC2/SPOT compute environments. - [Multiprocessing] Added `timeout` to `acquire()`, and `_getvalue()`, `_callmethod()` and `copy_proxy()` to the manager proxies. @@ -32,6 +31,7 @@ ### Fixed +- [Core] Fixed a repeated submission reusing the first serialization of a callable, freezing its captured state and the dependencies of the first runtime it ran on. - [Core] Fixed executor IDs repeating in a process whose environment is reset between executors. - [Core] Fixed `wait()` watching futures of another executor with a monitor that never sees them, leaving behind the monitors it started, and raising `TypeError` from `signal.alarm()` on a fractional timeout. - [Core] Fixed `result()` returning `None` instead of re-raising when the call had already failed. @@ -41,10 +41,11 @@ - [Core] Fixed `chunksize=0` and `execution_timeout=0` falling back to the config value. - [Core] Fixed a second Ctrl+C after a failed call turning into `Error in sys.excepthook`, and logging at interpreter shutdown raising on an already closed stream. - [Core] Fixed the function package carrying `.pytest_cache` directories and stale zips, a failed build leaving a partial zip behind, and `runtime_include_function` leaving the process in the build directory when the build failed. -- [Chaining] Fixed a list or a slice of futures of a previous job not being recognised as a chain, and `extra_args` failing every activation of the chained job instead of raising at submit time. +- [Chaining] Fixed a plain list, tuple or slice of futures not being recognised as a chain, leaving its consumed results in `get_result()`, and `extra_args` failing every activation instead of raising at submit time. +- [Concurrent] Fixed `shutdown()` returning before a submission already in progress had been registered. - [Job] Fixed a glob pattern in the object name raising `TypeError` instead of listing the objects, and a `head_object()` without `content-length` raising a bare `KeyError`. - [Job] Fixed the last byte of an object being left out of its partitions, and folder markers being counted as objects, returning empty partitions. -- [Monitoring] Fixed a status message lost in transit turning into a bogus timeout, or hanging `wait()` for ever. +- [Monitoring] Fixed a lost status message or an unread stored status turning into a bogus timeout, hanging `wait()` for ever, or one storage error being enough to declare a timeout. - [Monitoring] Fixed a nested executor publishing statuses to a queue nobody declares, and failed RabbitMQ publishes being dropped with nothing in the log. - [Redis] Fixed `put_object()` rejecting file-like objects, which made `upload_file()` always fail, and `head_object()` reporting every key as missing on Redis 7 and up, where the `DEBUG OBJECT` command it relied on is disabled. - [Redis] Fixed `list_objects()` returning the object bodies instead of their keys and sizes and not skipping keys whose value is gone, `head_bucket()` returning a bool instead of the bucket metadata, and `delete_objects()` raising on an empty list. @@ -59,7 +60,7 @@ - [Multiprocessing] Fixed shared list writes being dropped or misplaced through slices, `remove()`, `index()`, `pop()` and `del`, and a slice of a shared array returning one element too many. - [Multiprocessing] Fixed manager proxies not raising the `KeyError`, `ValueError` and `IndexError` the standard library raises. - [Multiprocessing] Fixed concurrent updates to a shared object overwriting each other, and a shared object being deleted while on its way to a worker. -- [Multiprocessing] Fixed `Condition.wait()` never reporting a notify, so every wait looked like a timeout. +- [Multiprocessing] Fixed `Condition.wait()` never reporting a notify, keeping a recursively acquired lock so nobody could notify it, and not giving the lock back after a Redis error. - [Multiprocessing] Fixed over-releasing a lock or bounded semaphore passing silently, and a re-entrant `RLock` giving back a token it never took. - [Multiprocessing] Fixed `Value()` and `Array()` ignoring their `lock` argument. - [Multiprocessing] Fixed `lithops.multiprocessing.context` being shadowed by a context instance, which broke every `mp.context.`. @@ -84,6 +85,7 @@ - [Storage] Fixed `CloudFileProxy.listdir()` returning nothing for its default argument. - [Worker] Fixed the remote invoker returning before its invocations in flight were done. - [Worker] Fixed the function process being aborted on macOS from the second call on, by setting Apple's fork-safety flag, and one killed by the OOM killer, or by any signal, being reported as a missing result. +- [Worker] Fixed the non-Unix worker pool sharing one task object across its calls, mixing up their ids, data and logs, by spawning a process per worker. - [Azure] Fixed the `az` CLI calls deadlocking when a command filled the stderr pipe. - [Azure Containers] Fixed a deploy racing a provisioning operation already in progress, and a container app left in `Failed` state never being recreated. - [CLI] Fixed `job list` and `worker list` crashing when there was nothing to list. diff --git a/lithops/concurrent/futures.py b/lithops/concurrent/futures.py index e24e9436f..2d139af79 100644 --- a/lithops/concurrent/futures.py +++ b/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 diff --git a/lithops/constants.py b/lithops/constants.py index 3de6b2421..c172e186a 100644 --- a/lithops/constants.py +++ b/lithops/constants.py @@ -145,7 +145,6 @@ def _in_temp(*parts): 'aws_batch', 'gcp_cloudrun', 'gcp_functions', - 'cloudrun', 'azure_functions', 'azure_containers', 'aliyun_fc', diff --git a/lithops/executors.py b/lithops/executors.py index 67b5e2770..b0d33d932 100644 --- a/lithops/executors.py +++ b/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): """ diff --git a/lithops/job/job.py b/lithops/job/job.py index fef84454b..f78ad1218 100644 --- a/lithops/job/job.py +++ b/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 @@ -47,7 +46,6 @@ logger = logging.getLogger(__name__) FUNCTION_CACHE = set() -_FUNC_SERIALIZE_CACHE = weakref.WeakKeyDictionary() MAX_DATA_IN_PAYLOAD = 8 * 1024 # Per invocation. 8KB @@ -62,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, @@ -115,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) @@ -127,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 diff --git a/lithops/monitoring/backends/storage/storage.py b/lithops/monitoring/backends/storage/storage.py index e3fb7f8a7..90dfec162 100644 --- a/lithops/monitoring/backends/storage/storage.py +++ b/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() diff --git a/lithops/monitoring/monitor.py b/lithops/monitoring/monitor.py index c7b319897..0195810f0 100644 --- a/lithops/monitoring/monitor.py +++ b/lithops/monitoring/monitor.py @@ -36,6 +36,8 @@ LOG_INTERVAL = 30 # Print monitor debug every LOG_INTERVAL seconds +MAX_TIMEOUT_STATUS_QUERIES = 3 + # Package every monitoring backend lives under. A Monitor subclass defined # there is a backend, and the contract is checked against it BACKENDS_PACKAGE = 'lithops.monitoring.backends' @@ -173,6 +175,7 @@ def __init__(self, executor_id, self._futures_lock = threading.RLock() self.futures = set() self._futures_by_id = {} + self._timeout_query_failures = {} self.present_jobs = set() # vars for _generate_tokens @@ -391,6 +394,14 @@ def _check_new_futures(self, call_status, f): return True + def _apply_recovered_status(self, future, call_status): + """Apply a stored completion unless another reader already did.""" + if _is_finished(future): + return False + if not self._check_new_futures(call_status, future): + self._mark_ready(future, call_status) + return True + def _future_timeout_checker(self, futures=None): """ Checks if running futures exceeded the timeout @@ -400,15 +411,51 @@ def _future_timeout_checker(self, futures=None): futures = self.tracked_futures() futures_running = [f for f in futures if f.running and f._call_status] for fut in futures_running: - try: - start_tstamp = fut._call_status['worker_start_tstamp'] - fut_timeout = start_tstamp + fut.execution_timeout + 5 - if current_time > fut_timeout: - msg = ( - 'The function exceeded the execution timeout ' - f'of {fut.execution_timeout} seconds.' + start_tstamp = fut._call_status['worker_start_tstamp'] + fut_timeout = start_tstamp + fut.execution_timeout + 5 + if current_time <= fut_timeout: + continue + + # A lost message or delayed poll is not proof the worker timed + # out. Check its terminal status directly, regardless of the + # message channel's sweep interval or storage listing latency. + fut_id = _future_id(fut) + if self.internal_storage is not None: + try: + call_status = self.internal_storage.get_call_status(*fut_id) + fut._status_query_count += 1 + except Exception: + failures = self._timeout_query_failures.get(fut_id, 0) + 1 + self._timeout_query_failures[fut_id] = failures + if failures < MAX_TIMEOUT_STATUS_QUERIES: + logger.debug( + f'{log_prefix(self.executor_id)} - Could not ' + f'verify the timeout of call {fut.call_id}; ' + f'retrying ({failures}/' + f'{MAX_TIMEOUT_STATUS_QUERIES})', + exc_info=True, + ) + continue + logger.warning( + f'{log_prefix(self.executor_id)} - Could not verify ' + f'the timeout of call {fut.call_id} in ' + f'{failures} attempts; reporting the timeout', + exc_info=True, ) - raise TimeoutError('HANDLER', msg) + call_status = None + self._timeout_query_failures.pop(fut_id, None) + if _is_finished(fut): + continue + if call_status: + self._apply_recovered_status(fut, call_status) + continue + + try: + msg = ( + 'The function exceeded the execution timeout ' + f'of {fut.execution_timeout} seconds.' + ) + raise TimeoutError('HANDLER', msg) except TimeoutError: # Raising and catching the error right away is what fills # sys.exc_info(), so that the client re-raises a real @@ -636,6 +683,13 @@ def _apply_status_message(self, call_status): else: self._hold_status(call_status) + def _apply_recovered_status(self, future, call_status): + if not super()._apply_recovered_status(future, call_status): + return False + if self.generate_tokens and 'activation_id' in call_status: + self._generate_tokens(call_status) + return True + def _storage_sweep(self): """ Picks up from the Object Storage the statuses whose message never @@ -670,11 +724,7 @@ def _storage_sweep(self): future._status_query_count += 1 if not call_status: continue - if not self._check_new_futures(call_status, future): - self._mark_ready(future, call_status) - recovered += 1 - if self.generate_tokens and 'activation_id' in call_status: - self._generate_tokens(call_status) + recovered += self._apply_recovered_status(future, call_status) if recovered: logger.debug( diff --git a/lithops/multiprocessing/synchronize.py b/lithops/multiprocessing/synchronize.py index 64f0e5273..73f80f885 100644 --- a/lithops/multiprocessing/synchronize.py +++ b/lithops/multiprocessing/synchronize.py @@ -206,6 +206,12 @@ def release(self): self.owned = False super().release() + def _release_save(self): + self.release() + + def _acquire_restore(self, state): + self.acquire() + # # Recursive lock @@ -248,6 +254,17 @@ def release(self): if self._count == 0: super().release() + def _release_save(self): + """Release all acquisitions while a condition waits.""" + count = self._count + self._count = 0 + super().release() + return count + + def _acquire_restore(self, count): + self.acquire() + self._count = count + # # Condition variable @@ -274,6 +291,25 @@ def acquire(self): def release(self): self._lock.release() + def _release_save(self): + """ + Hands the lock over while waiting. A lock that keeps no recursion + depth, such as a plain Semaphore, is simply released, which is what + threading.Condition does for the same case + """ + release_save = getattr(self._lock, '_release_save', None) + if release_save is None: + self._lock.release() + return None + return release_save() + + def _acquire_restore(self, state): + acquire_restore = getattr(self._lock, '_acquire_restore', None) + if acquire_restore is None: + self._lock.acquire() + else: + acquire_restore(state) + def __enter__(self): return self._lock.__enter__() @@ -290,16 +326,19 @@ def wait(self, timeout=None): if not res: raise Exception('Condition ({}) could not enqueue waiting key'.format(self._notify_handle)) - # Release lock, wait to get notified, acquire lock - self.release() - logger.debug('Waiting for token %s on condition %s', wait_handle, self._notify_handle) - if timeout is not None and timeout <= 0: - # BLPOP reads a zero timeout as "block for ever" - notified = self._client.lpop(wait_handle) is not None - else: - notified = _blpop(self._client, wait_handle, timeout) is not None - self._client.expire(wait_handle, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) - self.acquire() + # A notifier needs the lock even when the waiter acquired it more + # than once. Restore ownership on errors as well as on notification. + state = self._release_save() + try: + logger.debug('Waiting for token %s on condition %s', wait_handle, self._notify_handle) + if timeout is not None and timeout <= 0: + # BLPOP reads a zero timeout as "block for ever" + notified = self._client.lpop(wait_handle) is not None + else: + notified = _blpop(self._client, wait_handle, timeout) is not None + self._client.expire(wait_handle, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + finally: + self._acquire_restore(state) # Whether a notify arrived, rather than the timeout expiring, which # is what the standard library returns and callers branch on return notified diff --git a/lithops/tests/test_concurrent_futures.py b/lithops/tests/test_concurrent_futures.py index c1e3ca531..d9b9e05c6 100644 --- a/lithops/tests/test_concurrent_futures.py +++ b/lithops/tests/test_concurrent_futures.py @@ -743,6 +743,110 @@ def test_wait_mixes_in_stdlib_futures(self): class TestLifecycle: + @pytest.mark.parametrize('method', ['submit', 'map']) + @pytest.mark.parametrize('wait_for_results', [True, False]) + def test_shutdown_includes_a_submission_in_progress(self, method, wait_for_results): + entered = threading.Event() + release = threading.Event() + shutdown_done = threading.Event() + outcomes, errors = [], [] + lf = FakeLithopsFuture(value=42) + inner = FakeInnerExecutor(lf, map_results=[lf]) + ex = _adapter(inner) + ex._owns_executor = True + native_method = inner.call_async if method == 'submit' else inner.map + + def blocked(*args, **kwargs): + entered.set() + if not release.wait(timeout=5): + raise TimeoutError('test submission was not released') + return native_method(*args, **kwargs) + + setattr(inner, 'call_async' if method == 'submit' else 'map', blocked) + + def submit(): + try: + outcomes.append(ex.submit(abs, -42) if method == 'submit' else ex.map(abs, [-42])) + except BaseException as exc: + errors.append(exc) + + def shutdown(): + try: + ex.shutdown(wait=wait_for_results) + shutdown_done.set() + except BaseException as exc: + errors.append(exc) + + submitter = threading.Thread(target=submit) + closer = threading.Thread(target=shutdown) + submitter.start() + assert entered.wait(timeout=5) + closer.start() + try: + returned_early = shutdown_done.wait(timeout=0.1) + exited_early = inner.exited + finally: + release.set() + submitter.join(timeout=5) + closer.join(timeout=5) + ex.shutdown(wait=True) + # wait=False can already have handed teardown to the reaper. + if ex._reaper is not None: + ex._reaper.join(timeout=5) + assert not errors + assert not returned_early and not exited_early + assert inner.exited + if method == 'submit': + assert outcomes[0].result(timeout=5) == 42 + else: + assert list(outcomes[0]) == [42] + + def test_slow_submission_does_not_block_an_earlier_result(self): + entered, release, resolved = threading.Event(), threading.Event(), threading.Event() + lf = FakeLithopsFuture(finished=False) + inner = FakeInnerExecutor(lf) + ex = _adapter(inner) + first = ex.submit(abs, -42) + errors, results = [], [] + + def blocked(*args, **kwargs): + entered.set() + if not release.wait(timeout=5): + raise TimeoutError('test submission was not released') + return FakeLithopsFuture(value=1) + + inner.call_async = blocked + + def submit(): + try: + ex.submit(abs, -1) + except BaseException as exc: + errors.append(exc) + + def resolve(): + try: + results.append(first.result(timeout=1)) + resolved.set() + except BaseException as exc: + errors.append(exc) + + submitter = threading.Thread(target=submit) + reader = threading.Thread(target=resolve) + submitter.start() + assert entered.wait(timeout=5) + lf.finish(value=42) + reader.start() + try: + finished_while_submitting = resolved.wait(timeout=2) + finally: + release.set() + submitter.join(timeout=5) + reader.join(timeout=5) + ex.shutdown(wait=True) + assert not errors + assert finished_while_submitting + assert results == [42] + def test_submit_after_shutdown_raises(self): ex = _adapter() ex.shutdown(wait=False) diff --git a/lithops/tests/test_constants.py b/lithops/tests/test_constants.py index 1ba0d8e7a..717e86438 100644 --- a/lithops/tests/test_constants.py +++ b/lithops/tests/test_constants.py @@ -12,7 +12,11 @@ # limitations under the License. # +import importlib import os +import pathlib + +import pytest from lithops import constants from lithops.utils import get_default_backend, get_mode @@ -32,6 +36,27 @@ def test_default_backends_are_known(self): assert constants.LOCALHOST not in constants.SERVERLESS_BACKENDS assert constants.LOCALHOST not in constants.STANDALONE_BACKENDS + @pytest.mark.parametrize('backends, package', [ + ('SERVERLESS_BACKENDS', 'lithops.serverless.backends'), + ('STANDALONE_BACKENDS', 'lithops.standalone.backends'), + ]) + def test_the_known_backends_are_the_ones_that_exist(self, backends, package): + """ + The tuples name the backends, and the packages under + lithops/*/backends are what they resolve to. A name in the tuple + with no package behind it turns a clear "Unknown compute backend" + into a ModuleNotFoundError further down; a package nobody listed + is a backend get_mode() refuses + """ + module = importlib.import_module(package) + root = pathlib.Path(module.__file__).parent + on_disk = { + entry.name for entry in root.iterdir() + if entry.is_dir() and not entry.name.startswith(('_', '.')) + } + + assert set(getattr(constants, backends)) == on_disk + def test_backend_collections_are_immutable(self): assert isinstance(constants.SERVERLESS_BACKENDS, tuple) assert isinstance(constants.STANDALONE_BACKENDS, tuple) diff --git a/lithops/tests/test_job.py b/lithops/tests/test_job.py index 65161de3e..04cebb560 100644 --- a/lithops/tests/test_job.py +++ b/lithops/tests/test_job.py @@ -33,7 +33,6 @@ from lithops.job.job import ( FUNCTION_CACHE, MAX_DATA_IN_PAYLOAD, - _FUNC_SERIALIZE_CACHE, _store_func_and_modules, invalidate_function_cache, ) @@ -124,14 +123,10 @@ def _storage(): @pytest.fixture def fresh_function_cache(): saved = set(FUNCTION_CACHE) - saved_serialize = {func: dict(entries) for func, entries in _FUNC_SERIALIZE_CACHE.items()} FUNCTION_CACHE.clear() - _FUNC_SERIALIZE_CACHE.clear() yield FUNCTION_CACHE FUNCTION_CACHE.clear() FUNCTION_CACHE.update(saved) - _FUNC_SERIALIZE_CACHE.clear() - _FUNC_SERIALIZE_CACHE.update(saved_serialize) @pytest.fixture @@ -481,9 +476,55 @@ def __call__(self, objs, inc, exc): assert job.data_key is not None -class TestFuncSerializeCache: +class TestFunctionSerialization: - def test_second_map_skips_cloudpickle_of_same_function( + @pytest.mark.parametrize('callable_kind', ['closure', 'instance']) + def test_changed_callable_state_is_uploaded_again(self, fresh_function_cache, callable_kind): + state = {'multiplier': 2} + + def compute(x): + return x * state['multiplier'] + + class Multiply: + multiplier = 2 + + def __call__(self, x): + return x * self.multiplier + + func = compute if callable_kind == 'closure' else Multiply() + storage = _storage() + first = _make_map_job(func=func, internal_storage=storage, include_modules=None) + first_payload = pickle.loads(storage.put_func.call_args.args[1]) + assert pickle.loads(first_payload['func'])(10) == 20 + + state['multiplier'] = 3 + if callable_kind == 'instance': + func.multiplier = 3 + second = _make_map_job(func=func, internal_storage=storage, include_modules=None) + second_payload = pickle.loads(storage.put_func.call_args.args[1]) + assert pickle.loads(second_payload['func'])(10) == 30 + assert second.func_key != first.func_key + + def test_dependency_payload_matches_the_target_runtime(self, fresh_function_cache): + import yaml + + def compute(value): + return yaml.safe_load(value) + + storage = _storage() + _make_map_job( + func=compute, iterdata=['a: 1'], internal_storage=storage, + include_modules=['yaml'], runtime_meta=_runtime_meta(preinstalls=[['yaml', True]]), + ) + assert not pickle.loads(storage.put_func.call_args.args[1])['module_data'] + _make_map_job( + func=compute, iterdata=['a: 1'], internal_storage=storage, + include_modules=['yaml'], runtime_meta=_runtime_meta(preinstalls=[]), + ) + payload = pickle.loads(storage.put_func.call_args.args[1]) + assert 'yaml/__init__.py' in payload['module_data'] + + def test_second_map_refreshes_serialization_but_deduplicates_upload( self, monkeypatch, fresh_function_cache ): dumped = [] @@ -508,9 +549,10 @@ def tracking_dumps(obj): _make_map_job( internal_storage=storage, func=_echo, iterdata=[3, 4] ) - assert dumped.count(_echo) == 1 + assert dumped.count(_echo) == 2 data_dumps = [obj for obj in dumped if obj is not _echo] assert len(data_dumps) == 4 + storage.put_func.assert_called_once() class TestInvalidateFunctionCache: diff --git a/lithops/tests/test_map.py b/lithops/tests/test_map.py index 3ef3a0de4..a5949ff60 100644 --- a/lithops/tests/test_map.py +++ b/lithops/tests/test_map.py @@ -185,3 +185,16 @@ def mul_two(x): fexec = lithops.FunctionExecutor(config=pytest.lithops_config) result = fexec.map(add_one, [1, 2, 3]).map(mul_two).get_result() assert result == [4, 6, 8] + + @pytest.mark.parametrize('container', [list, tuple, lambda fs: fs[:]]) + def test_chaining_plain_futures_hides_consumed_outputs(self, container): + with lithops.FunctionExecutor(config=pytest.lithops_config) as fexec: + first = fexec.map(lambda x: x * x, [1, 2]) + fexec.map(lambda x: x * 2, container(first)) + assert fexec.get_result() == [2, 8] + + def test_chaining_a_slice_keeps_unconsumed_outputs(self): + with lithops.FunctionExecutor(config=pytest.lithops_config) as fexec: + first = fexec.map(lambda x: x * x, [1, 2]) + fexec.map(lambda x: x * 2, first[:1]) + assert fexec.get_result() == [4, 2] diff --git a/lithops/tests/test_monitor.py b/lithops/tests/test_monitor.py index 9309db4aa..78df518c2 100644 --- a/lithops/tests/test_monitor.py +++ b/lithops/tests/test_monitor.py @@ -29,6 +29,7 @@ from lithops.monitoring.backends import resolve_backend from lithops.monitoring.monitor import ( LOG_INTERVAL, + MAX_TIMEOUT_STATUS_QUERIES, Monitor, PollingMessageMonitor, _is_finished, @@ -572,6 +573,85 @@ def test_is_finished_and_is_started(self): class TestTimeoutAndStatusLog: + @pytest.mark.parametrize('channel', ['storage', 'message']) + @pytest.mark.parametrize('failed', [False, True]) + def test_stored_completion_wins_over_an_expired_timeout(self, monkeypatch, channel, failed): + storage = MagicMock() + cls = StorageMonitor if channel == 'storage' else PollingMessageMonitor + monitor = cls('sess-0', storage, queue.Queue(), {'M000': 1}, True, {'monitoring_interval': 1}) + future = FakeFuture('M000', running=True, execution_timeout=10, activation_id='act-1') + future._call_status = {'type': '__init__', 'worker_start_tstamp': 100} + monitor.add_futures([future]) + status = { + 'type': '__end__', 'executor_id': 'sess-0', 'job_id': 'M000', + 'call_id': '00000', 'activation_id': 'act-1', 'chunksize': 1, + 'worker_start_tstamp': 100, 'worker_end_tstamp': 102, + 'exception': failed, 'result': 'worker result', + } + storage.get_job_status.return_value = (set(), {('sess-0', 'M000', '00000')}) + storage.get_call_status.return_value = status + monkeypatch.setattr('lithops.monitoring.monitor.time.time', lambda: 116) + if channel == 'storage': + monitor._poll_and_process_job_status() + else: + monitor._last_message_tstamp = 100 + monkeypatch.setattr(monitor, '_poll_once', monitor.stop) + monitor.run() + assert monitor.token_bucket_q.get_nowait() == '#' + assert monitor.token_bucket_q.empty() + assert future._call_status == status + + def test_timeout_waits_for_storage_to_recover(self): + monitor = _monitor() + monitor.internal_storage = MagicMock() + monitor.internal_storage.get_call_status.side_effect = OSError('storage unavailable') + future = FakeFuture('M000', running=True, activation_id='act-1') + future._call_status = {'worker_start_tstamp': time.time() - 100} + monitor._future_timeout_checker([future]) + assert not future.ready + + monitor.internal_storage.get_call_status.side_effect = None + monitor.internal_storage.get_call_status.return_value = None + monitor._future_timeout_checker([future]) + assert future.ready + assert future._call_status['exception'] is True + + def test_timeout_is_reported_when_storage_stays_unreachable(self): + monitor = _monitor() + monitor.internal_storage = MagicMock() + monitor.internal_storage.get_call_status.side_effect = OSError('storage unavailable') + future = FakeFuture('M000', running=True, activation_id='act-1') + future._call_status = {'worker_start_tstamp': time.time() - 100} + for _ in range(MAX_TIMEOUT_STATUS_QUERIES - 1): + monitor._future_timeout_checker([future]) + assert not future.ready + monitor._future_timeout_checker([future]) + assert future.ready + assert future._call_status['exception'] is True + + def test_the_failure_budget_is_counted_per_call(self): + monitor = _monitor() + monitor.internal_storage = MagicMock() + monitor.internal_storage.get_call_status.side_effect = OSError('storage unavailable') + futures = [ + FakeFuture('M000', call_id=f'{i:05}', running=True, activation_id='act-1') + for i in range(MAX_TIMEOUT_STATUS_QUERIES) + ] + for future in futures: + future._call_status = {'worker_start_tstamp': time.time() - 100} + # One round over several calls is one failure each, not a shared + # budget that the last call in the round exhausts + monitor._future_timeout_checker(futures) + assert not any(future.ready for future in futures) + + def test_unexpired_calls_do_not_query_storage(self): + monitor = _monitor() + monitor.internal_storage = MagicMock() + future = FakeFuture('M000', running=True, activation_id='act-1') + future._call_status = {'worker_start_tstamp': time.time()} + monitor._future_timeout_checker([future]) + monitor.internal_storage.get_call_status.assert_not_called() + @pytest.fixture(autouse=True) def _propagate_monitor_logs(self): """ diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index 84a97c1d2..767181b56 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -1585,6 +1585,61 @@ def spy(client, name, timeout): class TestConditionContract: + def test_wait_fully_releases_a_recursive_lock(self, redis): + import copy + from lithops.multiprocessing import Condition, RLock + + lock = RLock() + cond = Condition(lock) + # A remote worker has its own ownership state for the same Redis key. + peer_lock = copy.copy(lock) + peer_cond = copy.copy(cond) + peer_cond._lock = peer_lock + acquired = [] + + def notify(): + acquired.append(peer_lock.acquire(timeout=1)) + if acquired[-1]: + try: + peer_cond.notify() + finally: + peer_lock.release() + + lock.acquire() + lock.acquire() + thread = threading.Thread(target=notify) + thread.start() + try: + notified = cond.wait(timeout=2) + assert lock.owned and lock._count == 2 + finally: + lock.release() + lock.release() + thread.join(timeout=3) + assert acquired == [True] + assert notified is True + + @pytest.mark.parametrize('recursive', [False, True]) + def test_wait_restores_the_lock_after_a_redis_error(self, redis, recursive): + from lithops.multiprocessing import Condition, Lock, RLock + + lock = RLock() if recursive else Lock() + cond = Condition(lock) + depth = 2 if recursive else 1 + for _ in range(depth): + lock.acquire() + try: + with patch('lithops.multiprocessing.synchronize._blpop', side_effect=OSError('connection lost')): + with pytest.raises(OSError, match='connection lost'): + cond.wait(timeout=1) + assert lock.owned + if recursive: + assert lock._count == depth + finally: + if lock.owned: + for _ in range(depth): + lock.release() + def test_wait_says_whether_it_was_notified(self, redis): """ The standard library returns False on a timeout and True on a diff --git a/lithops/tests/test_worker.py b/lithops/tests/test_worker.py index 4fe36a5a5..f428c4bca 100644 --- a/lithops/tests/test_worker.py +++ b/lithops/tests/test_worker.py @@ -131,6 +131,11 @@ def _record_task(task): fid.write(str(os.getpid())) +def _record_spawned_call(value, out_dir, id): + with open(os.path.join(out_dir, f'{id:05}-{value}'), 'w') as fid: + fid.write(str(os.getpid())) + + def _task(**kwargs): values = dict( extra_env={}, @@ -189,6 +194,25 @@ def test_loads_func_and_data(self): class TestTaskConsumer: + def test_queued_calls_keep_distinct_state(self): + job = _task(call_ids=['00000', '00001'], data=[b'first', b'second']) + work_queue = handler_module._fill_queue(job, 2) + rendezvous = threading.Barrier(2) + seen = [] + + def observe(task): + rendezvous.wait(timeout=5) + seen.append((task.call_id, task.data)) + + threads = [threading.Thread(target=task_consumer, args=(i, work_queue)) for i in range(2)] + with patch.object(handler_module, 'prepare_and_run_task', side_effect=observe): + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=5) + assert sorted(seen) == [('00000', b'first'), ('00001', b'second')] + assert job.data == [b'first', b'second'] + def test_runs_tasks_then_stops_on_sentinel(self): q = Queue() task = _task() @@ -302,37 +326,42 @@ def test_multi_worker_starts_processes_and_joins(self): ctx.Process.return_value = proc with patch('lithops.worker.handler.create_job', return_value=job): with patch('lithops.worker.handler.setup_lithops_logger'): - with patch('lithops.worker.handler._MP_CTX', ctx): + with patch('lithops.worker.handler._MP_CTX', ctx), \ + patch('lithops.worker.handler.is_unix_system', return_value=True): function_handler({}) ctx.Manager.assert_not_called() assert ctx.Process.call_count == 2 assert proc.start.call_count == 2 assert proc.join.call_count == 2 - def test_multi_worker_runs_every_task_once(self): + def test_non_unix_workers_use_process_isolation(self): + job = _task( + worker_processes=2, call_ids=['00000', '00001'], data=[b'a', b'b'] + ) + with patch.object(handler_module, 'create_job', return_value=job), \ + patch.object(handler_module, 'is_unix_system', return_value=False), \ + patch.object(handler_module, '_run_spawn_pool') as spawned: + function_handler({}) + spawned.assert_called_once_with(job, 2) + + def test_spawn_pool_runs_every_task_in_a_child(self, tmp_path): + from lithops.config import default_config + n_tasks = 6 + config = default_config(config_data={ + 'lithops': {'backend': 'localhost', 'storage': 'localhost'}, + 'localhost': {'storage_bucket': str(tmp_path / 'storage')}, + }) job = _task( - worker_processes=3, + config=config, worker_processes=2, call_ids=[f'{i:05}' for i in range(n_tasks)], - data=[f'd{i}'.encode() for i in range(n_tasks)], + data=[pickle.dumps({'value': f'd{i}', 'out_dir': str(tmp_path)}) for i in range(n_tasks)], + func=pickle.dumps(_record_spawned_call), ) - seen = [] - with patch('lithops.worker.handler.create_job', return_value=job): - with patch('lithops.worker.handler.setup_lithops_logger'): - with patch( - 'lithops.worker.handler.prepare_and_run_task', - side_effect=lambda task: seen.append( - (task.call_id, task.data) - ), - ): - with patch( - 'lithops.worker.handler._run_process_pool', - new=handler_module._run_thread_pool, - ): - function_handler({}) - assert sorted(seen) == [ - (f'{i:05}', f'd{i}'.encode()) for i in range(n_tasks) - ] + handler_module._run_spawn_pool(job, job.worker_processes) + done = sorted(p for p in tmp_path.iterdir() if p.is_file()) + assert [p.name for p in done] == [f'{i:05}-d{i}' for i in range(n_tasks)] + assert all(p.read_text() != str(os.getpid()) for p in done) @pytest.mark.skipif( not is_unix_system(), reason='the process pool needs fork' diff --git a/lithops/worker/handler.py b/lithops/worker/handler.py index fab3b8c7d..c3d66df03 100644 --- a/lithops/worker/handler.py +++ b/lithops/worker/handler.py @@ -159,15 +159,21 @@ def create_job(payload: Dict[str, Any]) -> SimpleNamespace: return job -def _fill_queue(job: SimpleNamespace, worker_processes: int) -> Queue: +def _fill_queue(job: SimpleNamespace, worker_processes: int, work_queue=None): """ Loads every task of the job in a queue, followed by one sentinel per worker. Every task is known upfront, so nothing is queued afterwards """ - work_queue = Queue() + if work_queue is None: + work_queue = Queue() for call_id, data in zip(job.call_ids, job.data): - work_queue.put((job, call_id, data)) + task = SimpleNamespace(**vars(job)) + # Keep only this call in the queued task. Besides isolating mutable + # state, this avoids pickling the whole input list once per call. + task.call_ids = [call_id] + task.data = data + work_queue.put((task, call_id, data)) for _ in range(worker_processes): work_queue.put(ShutdownSentinel()) @@ -203,21 +209,34 @@ def _run_process_pool(job: SimpleNamespace, worker_processes: int) -> None: worker.join() -def _run_thread_pool(job: SimpleNamespace, worker_processes: int) -> None: +def _run_spawn_pool(job: SimpleNamespace, worker_processes: int) -> None: """ - Runs the tasks of the job in threads. Used where there is no fork, so - tasks share this interpreter instead of getting a process each + Runs workers in separate interpreters where fork is unavailable. + + The multiprocessing queue gives each call its own deserialized task. + Threads would share the task namespace, cwd, environment and log streams. """ - work_queue = _fill_queue(job, worker_processes) + ctx = mp.get_context('spawn') + work_queue = ctx.Queue() workers = [] - - for pid in range(worker_processes): - worker = Thread(target=task_consumer, args=(pid, work_queue)) - workers.append(worker) - worker.start() - - for worker in workers: - worker.join() + try: + for pid in range(worker_processes): + worker = ctx.Process(target=task_consumer, args=(pid, work_queue)) + worker.start() + workers.append(worker) + _fill_queue(job, worker_processes, work_queue) + for worker in workers: + worker.join() + if worker.exitcode != 0: + raise RuntimeError(f'Worker process {worker.pid} exited with code {worker.exitcode}') + finally: + for worker in workers: + if worker.is_alive(): + worker.terminate() + worker.join() + # If consumers died, the feeder may still hold data nobody can read. + work_queue.cancel_join_thread() + work_queue.close() def function_handler(payload: Dict[str, Any]) -> None: @@ -238,7 +257,7 @@ def function_handler(payload: Dict[str, Any]) -> None: elif is_unix_system(): _run_process_pool(job, worker_processes) else: - _run_thread_pool(job, worker_processes) + _run_spawn_pool(job, worker_processes) module_path = os.path.join(MODULES_DIR, job.job_key) if module_path in sys.path: From e5426b4b510f924ae75c7e8f1a17f9a4811dfbf7 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Wed, 9 Sep 2026 15:01:15 +0200 Subject: [PATCH 04/10] Update CI --- .github/workflows/tests-all-os.yml | 24 ++++++++---------- lithops/storage/cloud_proxy.py | 40 +++++++++++++++++++----------- 2 files changed, 35 insertions(+), 29 deletions(-) diff --git a/.github/workflows/tests-all-os.yml b/.github/workflows/tests-all-os.yml index 2cb93c12c..e27a3bc3b 100644 --- a/.github/workflows/tests-all-os.yml +++ b/.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" diff --git a/lithops/storage/cloud_proxy.py b/lithops/storage/cloud_proxy.py index 817e8b6be..fec481c8c 100644 --- a/lithops/storage/cloud_proxy.py +++ b/lithops/storage/cloud_proxy.py @@ -281,19 +281,29 @@ def cloud_open(filename, mode='r', cloud_storage=None): raise ValueError(f"Unsupported mode '{mode}': only 'r' and 'w' are") -if not is_lithops_worker(): - try: - _storage = CloudStorage() - except FileNotFoundError: - # should never happen unless we are using - # this module classes for other purposes - os = None - open = None +def __getattr__(name): + """ + Builds the ``os`` and ``open`` stand-ins the first time one is read. + + Building them while this module was imported reached the default storage + backend from the import itself, so a backend that is not configured, not + installed or not reachable broke ``import`` for everything defined here. + Whatever the backend has to say is now raised where the stand-in is + asked for, rather than turning into a ``None`` that fails later on + """ + if name not in ('os', 'open'): + raise AttributeError(f'module {__name__!r} has no attribute {name!r}') + + if is_lithops_worker(): + # should never be used unless we explicitly import + # inside a function, which is not a good practice + value = None else: - os = CloudFileProxy(_storage) - open = partial(cloud_open, cloud_storage=_storage) -else: - # should never be used unless we explicitly import - # inside a function, which is not a good practice - os = None - open = None + storage = CloudStorage() + value = ( + CloudFileProxy(storage) if name == 'os' + else partial(cloud_open, cloud_storage=storage) + ) + + globals()[name] = value + return value From 3f0d9bf471b0c9e7fbb3019e0e73ed93e2888713 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Wed, 9 Sep 2026 15:30:21 +0200 Subject: [PATCH 05/10] Update tests --- lithops/job/partitioner.py | 11 ++- lithops/tests/test_config.py | 2 + lithops/tests/test_job.py | 31 +++++++++ lithops/tests/test_localhost.py | 16 +++-- lithops/tests/test_multiprocessing.py | 98 ++++++++++++++++++++++----- 5 files changed, 133 insertions(+), 25 deletions(-) diff --git a/lithops/job/partitioner.py b/lithops/job/partitioner.py index 3b8f37d50..9c07693f3 100644 --- a/lithops/job/partitioner.py +++ b/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) diff --git a/lithops/tests/test_config.py b/lithops/tests/test_config.py index 3f6390ea1..b98a06f9d 100644 --- a/lithops/tests/test_config.py +++ b/lithops/tests/test_config.py @@ -194,6 +194,8 @@ def test_explicit_file_expands_user_home(self, monkeypatch, tmp_path): cfg = tmp_path / 'home.yml' cfg.write_text(yaml.dump({'lithops': {'mode': 'localhost'}})) monkeypatch.setenv('HOME', str(tmp_path)) + # expanduser() reads USERPROFILE on Windows and ignores HOME + monkeypatch.setenv('USERPROFILE', str(tmp_path)) loaded = load_config('~/home.yml', log=False) assert loaded['lithops']['mode'] == 'localhost' diff --git a/lithops/tests/test_job.py b/lithops/tests/test_job.py index 04cebb560..b309cb3ef 100644 --- a/lithops/tests/test_job.py +++ b/lithops/tests/test_job.py @@ -812,6 +812,37 @@ def test_empty_iterdata_returns_empty_partitions(self): [], ) + @pytest.mark.parametrize('path', [ + r'C:\\data\\a.txt', 'C:/data/a.txt', r'\\\\server\\share\\a.txt', + ]) + def test_a_windows_path_is_read_as_a_local_file(self, path): + """ + Nothing starts with a separator on Windows, so a drive letter or a + UNC share used to fall through to the Object Storage branch and the + local file was looked for in a bucket + """ + with patch( + 'lithops.job.partitioner._split_objects_from_paths', + return_value=([{'obj': path}], [1]), + ) as split_paths: + parts, ppo = create_partitions( + {}, _storage(), [{'obj': path}], None, None, None + ) + split_paths.assert_called_once() + assert [entry['obj'] for entry in split_paths.call_args.args[0]] == [path] + assert (parts, ppo) == ([{'obj': path}], [1]) + + @pytest.mark.parametrize('key', [ + 'bucket/key.csv', 'C:relative', 'bucket/a:b', 's3://bucket/key', + ]) + def test_a_storage_key_is_not_read_as_a_windows_path(self, key): + with patch( + 'lithops.job.partitioner._split_objects_from_object_storage', + return_value=([], []), + ) as split_objects: + create_partitions({}, _storage(), [{'obj': key}], None, None, None) + split_objects.assert_called_once() + def test_http_takes_precedence_over_paths(self, tmp_path): f = tmp_path / 'f.txt' f.write_bytes(b'hello world!!') diff --git a/lithops/tests/test_localhost.py b/lithops/tests/test_localhost.py index ad9eb5559..c76ce5f4c 100644 --- a/lithops/tests/test_localhost.py +++ b/lithops/tests/test_localhost.py @@ -39,7 +39,7 @@ from lithops.storage.backends.localhost.localhost import LocalhostStorageBackend from lithops.storage.utils import StorageNoSuchKeyError from lithops.tests.functions import simple_map_function, sleep_seconds -from lithops.utils import BackendType, CountDownLatch +from lithops.utils import BackendType, CountDownLatch, is_unix_system from lithops.version import __version__ @@ -469,8 +469,8 @@ def test_default_stop_kills_matching_process_group(self): env.task_processes['sess-0-M000-00000'] = proc env.jobs = {'sess-0-M000': CountDownLatch(0)} env.is_unix_system = True - with patch('lithops.localhost.utils.os.getpgid', return_value=9), \ - patch('lithops.localhost.utils.os.killpg') as killpg, \ + with patch('lithops.localhost.utils.os.getpgid', return_value=9, create=True), \ + patch('lithops.localhost.utils.os.killpg', create=True) as killpg, \ patch.object(v2.ExecutionEnvironment, '_teardown'): env.stop(['sess-0-M000']) killpg.assert_called_once_with(9, signal.SIGKILL) @@ -525,7 +525,7 @@ def test_finish_tears_down_without_killing(self): env.task_processes['sess-0-M000-00000'] = proc env.jobs = {'sess-0-M000': CountDownLatch(0)} env.consumer_threads = [MagicMock()] - with patch('lithops.localhost.utils.os.killpg') as killpg: + with patch('lithops.localhost.utils.os.killpg', create=True) as killpg: env.finish(['sess-0-M000']) killpg.assert_not_called() assert 'sess-0-M000-00000' in env.task_processes @@ -774,8 +774,8 @@ def test_stop_kills_process_group(self): proc.pid = 77 env.jobs['sess-0-M000'] = proc env.is_unix_system = True - with patch('lithops.localhost.utils.os.getpgid', return_value=5), \ - patch('lithops.localhost.utils.os.killpg') as killpg: + with patch('lithops.localhost.utils.os.getpgid', return_value=5, create=True), \ + patch('lithops.localhost.utils.os.killpg', create=True) as killpg: env.stop(['sess-0-M000']) killpg.assert_called_once_with(5, signal.SIGKILL) assert 'sess-0-M000' not in env.jobs @@ -1214,6 +1214,10 @@ def _ensure_localhost_python_image(): not _docker_daemon_available(), reason='docker/podman is not installed or the daemon is not running', ) +@pytest.mark.skipif( + not is_unix_system(), + reason='pulling the Linux image does not fit the timeout on Windows', +) class TestLocalhostContainerLive: """Live localhost jobs inside a Docker Hub python:X.Y container.""" diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index 767181b56..e83648ad3 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -22,9 +22,11 @@ """ import ctypes +import functools import gc import pickle import queue +import socket import sys import threading import time @@ -67,6 +69,31 @@ def redis(): return server +REDIS_HOST = 'localhost' +REDIS_PORT = 6379 + + +@functools.cache +def _redis_reachable(): + """ + Whether a server answers, asked once for the whole session. + + Every test that wants one used to open its own connection to find out, + and there are dozens of them. The socket is tried before the client + because a refused connection comes back at once, while the client + spends seconds on its own retries before it gives up + """ + try: + socket.create_connection((REDIS_HOST, REDIS_PORT), timeout=1).close() + except OSError: + return False + try: + mp_util.PicklableRedis(host=REDIS_HOST, port=REDIS_PORT).ping() + return True + except Exception: + return False + + @pytest.fixture def real_redis(): """ @@ -77,16 +104,14 @@ def real_redis(): meant to check in the same pass proves nothing. Skipped where no server is reachable; every key the test makes is dropped afterwards """ + if not _redis_reachable(): + pytest.skip('no Redis reachable on localhost') + # The picklable wrapper, which is what get_redis_client() builds. A plain # redis.Redis carries a connection pool with a lock in it, so a proxy # holding one cannot be sent to a worker -- a difference the tests must # not paper over - client = mp_util.PicklableRedis(host='localhost') - try: - client.ping() - except Exception: - pytest.skip('no Redis reachable on localhost') - + client = mp_util.PicklableRedis(host=REDIS_HOST, port=REDIS_PORT) before = set(client.keys('*')) saved_client, saved_config = mp_util.REDIS_CLIENT, mp_util.LITHOPS_CONFIG mp_util.REDIS_CLIENT = client @@ -1285,21 +1310,60 @@ def test_poll_zero_looks_at_the_buffer(self): assert self._connection(queue.Queue())._poll(0.0) is False def test_poll_returns_as_soon_as_a_message_lands(self): + """ + A message that lands mid-wait is picked up on the very next check, + and the waits leading up to it are the first small steps of the + backoff rather than the flat 0.1s this used to sleep. + + The waits are counted instead of timed: a wall clock here also + measures how late the runner got round to landing the message, + which is not what the poll is being held to + """ + from lithops.multiprocessing import connection as conn_mod + buff = queue.Queue() - timer = threading.Timer(0.01, lambda: buff.put(b'x')) - timer.start() - try: - started = time.monotonic() + waits = [] + + def wait_and_land(duration): + waits.append(duration) + if len(waits) == 3: + buff.put(b'x') + + with patch.object(conn_mod.time, 'sleep', side_effect=wait_and_land): assert self._connection(buff)._poll(5.0) is True - assert time.monotonic() - started < 0.09 - finally: - timer.cancel() + + assert len(waits) == 3 + assert waits == [ + conn_mod.POLL_MIN_SLEEP, + conn_mod.POLL_MIN_SLEEP * 2, + conn_mod.POLL_MIN_SLEEP * 4, + ] + assert sum(waits) < conn_mod.POLL_MAX_SLEEP def test_poll_does_not_overshoot_its_timeout(self): - started = time.monotonic() - assert self._connection(queue.Queue())._poll(0.05) is False - # The flat 0.1s step used to sleep past the deadline it was given - assert 0.05 <= time.monotonic() - started < 0.12 + """ + The waits add up to exactly the timeout, the last one being clipped + to what is left of it. The flat 0.1s step used to sleep past the + deadline it was given. + + Driven off a fake clock: a real one also measures how far the OS + overshot each sleep, which the poll cannot be held to + """ + from lithops.multiprocessing import connection as conn_mod + + waits = [] + clock = [0.0] + + def sleep(duration): + waits.append(duration) + clock[0] += duration + + with patch.object(conn_mod.time, 'sleep', side_effect=sleep), \ + patch.object(conn_mod.time, 'monotonic', side_effect=lambda: clock[0]): + assert self._connection(queue.Queue())._poll(0.05) is False + + assert sum(waits) == pytest.approx(0.05) + assert waits[-1] < conn_mod.POLL_MAX_SLEEP def test_poll_until_always_checks_once(self): from lithops.multiprocessing.connection import _poll_until From 3c673cfc91af24a4a9e428f0977456045986b21a Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Wed, 9 Sep 2026 15:51:34 +0200 Subject: [PATCH 06/10] Update tests --- lithops/localhost/config.py | 2 +- lithops/tests/test_executors.py | 5 ++++- lithops/tests/test_localhost.py | 13 +++++++++++++ lithops/tests/test_multiprocessing.py | 7 +++++++ lithops/tests/test_plots.py | 2 ++ lithops/tests/test_standalone.py | 6 +++++- lithops/tests/test_worker.py | 21 +++++++++++++++++++-- 7 files changed, 51 insertions(+), 5 deletions(-) diff --git a/lithops/localhost/config.py b/lithops/localhost/config.py index 882a47b21..22422a08d 100644 --- a/lithops/localhost/config.py +++ b/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( diff --git a/lithops/tests/test_executors.py b/lithops/tests/test_executors.py index b540529e5..5968b87f7 100644 --- a/lithops/tests/test_executors.py +++ b/lithops/tests/test_executors.py @@ -603,7 +603,10 @@ def test_map_reduce_spawn_reducer_always_and_percentage(self): def test_map_local_file_partitions(self, tmp_path): path = tmp_path / 'data.txt' text = 'alpha beta gamma delta\n' * 20 - path.write_text(text) + # newline='' keeps the bytes on disk as written: the default + # translates every \n to \r\n on Windows, and the worker hands + # back what it read + path.write_text(text, newline='') fexec = lithops.FunctionExecutor(config=pytest.lithops_config) futures = fexec.map( echo_object, str(path), obj_chunk_number=2, obj_newline=None diff --git a/lithops/tests/test_localhost.py b/lithops/tests/test_localhost.py index c76ce5f4c..b7c1dd8a6 100644 --- a/lithops/tests/test_localhost.py +++ b/lithops/tests/test_localhost.py @@ -78,6 +78,13 @@ def test_environment_default_for_python_and_paths(self): 'python3.12', '/usr/bin/python3', r'C:\Python\python.exe', + # A forward-slash drive path and a UNC share are local too, and + # a basename that does not look like an interpreter used to send + # both of them off to be pulled as a container image + 'C:/Python/python.exe', + r'C:\tools\my-interpreter.exe', + 'C:/tools/my-interpreter.exe', + r'\\server\share\my-interpreter.exe', ): assert localhost_config.get_environment(runtime) is ( localhost_config.LocalhostEnvironment.DEFAULT @@ -461,6 +468,9 @@ def test_default_run_task_does_not_dump_logs_when_signalled(self): log_fail.assert_not_called() assert 'sess-0-M000-00000' not in env.task_processes + @pytest.mark.skipif( + not is_unix_system(), reason='there is no SIGKILL on Windows' + ) def test_default_stop_kills_matching_process_group(self): env = v2.DefaultEnvironment(_config(worker_processes=1)) proc = MagicMock() @@ -767,6 +777,9 @@ def test_copy_lithops_uses_v1_runner(self, tmp_path, monkeypatch): runner_src = copy_pkg.call_args[0][1] assert runner_src.endswith(os.path.join('localhost', 'v1', 'runner.py')) + @pytest.mark.skipif( + not is_unix_system(), reason='there is no SIGKILL on Windows' + ) def test_stop_kills_process_group(self): env = v1.DefaultEnvironment(_config()) proc = MagicMock() diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index e83648ad3..3ad9f1d8a 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -36,6 +36,7 @@ import pytest from unittest.mock import MagicMock, patch +from lithops.utils import is_unix_system from lithops.multiprocessing import config as mp_config from lithops.multiprocessing import util as mp_util from lithops.tests.mp_fakeredis import FakeRedis @@ -1746,6 +1747,9 @@ class TestWaitAlarm: seconds """ + @pytest.mark.skipif( + not is_unix_system(), reason='wait() only arms an alarm on POSIX' + ) def test_a_fractional_timeout_is_rounded_up(self): lw = sys.modules['lithops.wait'] @@ -1764,6 +1768,9 @@ def test_a_timeout_that_has_passed_raises_at_once(self): with pytest.raises(TimeoutError, match='Timeout of 0 seconds'): lw._set_wait_alarm(0) + @pytest.mark.skipif( + not is_unix_system(), reason='wait() only arms an alarm on POSIX' + ) def test_pool_get_with_a_fractional_timeout_raises_timeout_error(self): """ Pool.AsyncResult.get(0.2) reached signal.alarm() through wait() and diff --git a/lithops/tests/test_plots.py b/lithops/tests/test_plots.py index 297bf393a..dd262be6f 100644 --- a/lithops/tests/test_plots.py +++ b/lithops/tests/test_plots.py @@ -47,6 +47,8 @@ def test_explicit_path_appends_suffix(self, tmp_path): def test_expands_user_home_when_tilde_present(self, tmp_path, monkeypatch): monkeypatch.setenv('HOME', str(tmp_path)) + # expanduser() reads USERPROFILE on Windows and ignores HOME + monkeypatch.setenv('USERPROFILE', str(tmp_path)) path = _plot_destination(os.path.join('~', 'out'), 'timeline.png') expected = '{}_{}'.format( os.path.realpath(os.path.join(str(tmp_path), 'out')), diff --git a/lithops/tests/test_standalone.py b/lithops/tests/test_standalone.py index 78bc0207e..586db67a4 100644 --- a/lithops/tests/test_standalone.py +++ b/lithops/tests/test_standalone.py @@ -26,7 +26,7 @@ is_container_runtime, lithops_pip_spec_from_config, ) -from lithops.utils import BackendType +from lithops.utils import BackendType, is_unix_system from lithops.version import __version__ @@ -145,6 +145,10 @@ def test_worker_setup_script_uses_native_python(self): assert '/usr/bin/python3' in script assert 'docker run --rm --name lithops_worker' not in script + @pytest.mark.skipif( + not is_unix_system(), + reason='the docker worker script is built from the client uid/gid', + ) def test_worker_setup_script_uses_docker_for_tagged_python(self): script = get_worker_setup_script( { diff --git a/lithops/tests/test_worker.py b/lithops/tests/test_worker.py index f428c4bca..b57a723dd 100644 --- a/lithops/tests/test_worker.py +++ b/lithops/tests/test_worker.py @@ -284,6 +284,10 @@ def test_get_returns_the_job_and_survives_data_rebinding(self): assert jar.get()[1:] == ('00001', b'd1') jar.close_reader() + @pytest.mark.skipif( + not is_unix_system(), + reason='only a POSIX pipe reports a dead reader as a broken pipe', + ) def test_dispatch_survives_workers_that_died(self): jar = self._jar(2) jar.close_reader() @@ -317,6 +321,9 @@ def test_single_worker_uses_threading_queue(self): consumer.assert_called_once() assert consumer.call_args[0][0] == 0 + @pytest.mark.skipif( + not is_unix_system(), reason='the process pool needs fork' + ) def test_multi_worker_starts_processes_and_joins(self): job = _task( worker_processes=2, call_ids=['00000', '00001'], data=[b'a', b'b'] @@ -539,6 +546,9 @@ def _exc_of(self, status): }['exc_info'] return pickle.loads(ast.literal_eval(pickled))[0] + @pytest.mark.skipif( + not is_unix_system(), reason='there is no SIGKILL on Windows' + ) def test_a_sigkilled_jobrunner_is_reported_as_memory(self, tmp_path): task = _task() task.log_stream = MagicMock() @@ -639,6 +649,9 @@ class TestJobRunnerDeathReason: error looking in the wrong place """ + @pytest.mark.skipif( + not is_unix_system(), reason='there is no SIGKILL on Windows' + ) def test_sigkill_reads_as_memory(self): from lithops.worker.handler import _jobrunner_death_reason reason = _jobrunner_death_reason(-signal.SIGKILL) @@ -911,10 +924,14 @@ def test_get_runtime_metadata(self): def test_peak_memory_and_disk(self, tmp_path): mem = peak_memory() assert mem is None or mem >= 0 - assert free_disk_space(str(tmp_path)) > 0 + if is_unix_system(): + # free_disk_space() reads statvfs, which Windows has no equivalent of + assert free_disk_space(str(tmp_path)) > 0 def test_get_memory_usage_non_root_returns_none(self): - if os.geteuid() != 0: + # ps_mem needs both a Unix system and root, and it is the system + # that is checked first: os.geteuid() does not exist on Windows + if not is_unix_system() or os.geteuid() != 0: assert get_memory_usage() is None def test_memory_monitor_sends_peak_on_poll(self): From 447593e2e6ff3858d22de7c88fc070aea39b7330 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Wed, 9 Sep 2026 16:19:56 +0200 Subject: [PATCH 07/10] Update tests --- lithops/storage/storage.py | 4 +++- lithops/tests/test_storage_layer.py | 10 ++++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/lithops/storage/storage.py b/lithops/storage/storage.py index d55855bdb..0151161d3 100644 --- a/lithops/storage/storage.py +++ b/lithops/storage/storage.py @@ -424,9 +424,11 @@ def get_call_status(self, executor_id, job_id, call_id): status_key = utils.create_status_key(executor_id, job_id, call_id) try: data = self.storage.get_object(self.bucket, status_key) - return json.loads(data.decode('ascii')) except utils.StorageNoSuchKeyError: return None + if not data: + return None + return json.loads(data.decode('ascii')) def get_call_output(self, executor_id, job_id, call_id): """ diff --git a/lithops/tests/test_storage_layer.py b/lithops/tests/test_storage_layer.py index 2c22b13a8..1002e9cee 100644 --- a/lithops/tests/test_storage_layer.py +++ b/lithops/tests/test_storage_layer.py @@ -275,6 +275,16 @@ def test_get_call_status_decodes_ascii_json(self): internal.storage.get_object.return_value = b'{"ok": true}' assert internal.get_call_status('e', 'M000', '00000') == {'ok': True} + def test_get_call_status_reads_a_half_written_status_as_missing(self): + """ + The localhost backend fills an object after truncating it, so a poll + can read an empty one. That is a status not written yet, and used to + come back as a JSONDecodeError that failed the call + """ + internal = _bare_internal() + internal.storage.get_object.return_value = b'' + assert internal.get_call_status('e', 'M000', '00000') is None + def test_runtime_meta_memory_cache(self): RUNTIME_META_CACHE.clear() internal = _bare_internal() From 9f81814b2bf89f173fabe4c516e5803fd16aed17 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Wed, 9 Sep 2026 16:39:42 +0200 Subject: [PATCH 08/10] Update test --- lithops/tests/test_multiprocessing.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index 3ad9f1d8a..b9a77fe9e 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -85,7 +85,7 @@ def _redis_reachable(): spends seconds on its own retries before it gives up """ try: - socket.create_connection((REDIS_HOST, REDIS_PORT), timeout=1).close() + socket.create_connection((REDIS_HOST, REDIS_PORT), timeout=0.25).close() except OSError: return False try: From 7d947c81442000c111c7fcdbc1eeb38a0739bb07 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Thu, 10 Sep 2026 17:46:07 +0200 Subject: [PATCH 09/10] Update monitor --- lithops/monitoring/backends/__init__.py | 34 ++++----------------- lithops/monitoring/monitor.py | 40 +++++-------------------- lithops/tests/test_monitor.py | 32 ++------------------ 3 files changed, 15 insertions(+), 91 deletions(-) diff --git a/lithops/monitoring/backends/__init__.py b/lithops/monitoring/backends/__init__.py index a04352ef3..f307114a5 100644 --- a/lithops/monitoring/backends/__init__.py +++ b/lithops/monitoring/backends/__init__.py @@ -33,8 +33,8 @@ lithops.monitoring.backends. 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,15 +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, backend: Optional[str] = None, @@ -95,28 +86,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.``, 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.``, 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 diff --git a/lithops/monitoring/monitor.py b/lithops/monitoring/monitor.py index 0195810f0..195370285 100644 --- a/lithops/monitoring/monitor.py +++ b/lithops/monitoring/monitor.py @@ -112,40 +112,16 @@ class Monitor(threading.Thread): #: not turn into an AttributeError on the first status telemetry = NOOP_TELEMETRY - def __init_subclass__(cls, abstract: bool = False, **kwargs): + def __init_subclass__(cls, **kwargs): """ - Checks the backend contract as soon as the class is defined, so that - a backend that does not hold up its end fails at import rather than - halfway through a job. - - ``abstract=True`` opts a helper class out of the check + Names the backend after the package it is defined in, so that a + backend does not have to repeat the name it already has and the two + cannot drift apart. Classes outside the backends package, such as + the abstract helpers below, keep a backend_name of None """ super().__init_subclass__(**kwargs) - - package = _backend_package_of(cls) - if abstract or package is None: - return - if cls.backend_name is None: - cls.backend_name = package - elif cls.backend_name != package: - raise TypeError( - f"{cls.__name__}.backend_name is '{cls.backend_name}' but the " - f"backend package is '{package}'. The two name the same " - f"thing: the config section, and the value of 'monitoring:'" - ) - - polling = globals().get('PollingMessageMonitor') - implements_run = cls.run is not threading.Thread.run - implements_receive = polling is not None and issubclass( - cls, polling - ) and cls._receive_messages is not polling._receive_messages - if not implements_run and not implements_receive: - raise TypeError( - f'{cls.__name__} implements neither run() nor ' - f'_receive_messages(). A monitoring backend has to consume ' - f'statuses one way or the other' - ) + cls.backend_name = _backend_package_of(cls) def __init__(self, executor_id, internal_storage, @@ -519,7 +495,7 @@ def _print_status_log(self, force=False): ) -class MessageMonitor(Monitor, abstract=True): +class MessageMonitor(Monitor): """ Monitor for backends that receive one call-status message at a time (RabbitMQ, SQS, Pub/Sub, ...). @@ -735,7 +711,7 @@ def _storage_sweep(self): return recovered -class PollingMessageMonitor(MessageMonitor, abstract=True): +class PollingMessageMonitor(MessageMonitor): """ Pulls status messages with a timeout so the same loop can expire futures and notice stop(). Redis, SQS, Pub/Sub and Azure Queue use diff --git a/lithops/tests/test_monitor.py b/lithops/tests/test_monitor.py index 78df518c2..028bc8981 100644 --- a/lithops/tests/test_monitor.py +++ b/lithops/tests/test_monitor.py @@ -1673,8 +1673,8 @@ def test_a_failing_sweep_does_not_take_the_monitor_down(self): class TestBackendContract: """ - "Adding a backend is adding a package" only holds if a package that - does not hold up its end says so, instead of quietly monitoring nothing + "Adding a backend is adding a package" only holds if the package name + is what names the backend, and every built-in one is reachable that way """ def _in_backends(self, name='fake'): @@ -1687,38 +1687,10 @@ def test_backend_name_comes_from_the_package(self): )) assert cls.backend_name == 'fake' - def test_a_backend_name_that_disagrees_is_rejected(self): - with pytest.raises(TypeError, match='backend package'): - type('Drifted', (Monitor,), dict( - self._in_backends(), - backend_name='something-else', - run=lambda self: None, - )) - - def test_a_backend_that_consumes_nothing_is_rejected(self): - with pytest.raises(TypeError, match='neither run'): - type('Empty', (Monitor,), dict(self._in_backends())) - - def test_a_helper_class_can_opt_out(self): - cls = type('Helper', (Monitor,), dict(self._in_backends()), - abstract=True) - assert cls.backend_name is None - def test_classes_outside_the_backends_package_are_left_alone(self): cls = type('Local', (Monitor,), {'__module__': 'somewhere.else'}) assert cls.backend_name is None - def test_exports_are_checked_against_the_base_classes(self): - from lithops.monitoring.backends import load_backend_attr - module = MagicMock() - module.MonitoringBackend = 'not a class' - with patch( - 'lithops.monitoring.backends.importlib.import_module', - return_value=module, - ): - with pytest.raises(ValueError, match='not a Monitor subclass'): - load_backend_attr('fake', 'MonitoringBackend') - def test_a_missing_export_is_reported(self): from lithops.monitoring.backends import load_backend_attr module = SimpleNamespace() From 882d850ec9512845b2dec14bf935bc5e02a0f15b Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sat, 12 Sep 2026 11:44:49 +0200 Subject: [PATCH 10/10] Fix flake8 --- lithops/monitoring/backends/__init__.py | 1 + 1 file changed, 1 insertion(+) diff --git a/lithops/monitoring/backends/__init__.py b/lithops/monitoring/backends/__init__.py index f307114a5..4c53135fa 100644 --- a/lithops/monitoring/backends/__init__.py +++ b/lithops/monitoring/backends/__init__.py @@ -44,6 +44,7 @@ #: backend can act as a monitoring channel, so this always works DEFAULT_BACKEND = 'storage' + def resolve_backend( config: Optional[Dict[str, Any]] = None, backend: Optional[str] = None,