From 7d754f8f8e427b7b5c02b4ab4167369d12723533 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 00:04:04 +0200 Subject: [PATCH 1/9] Update Docs --- CHANGELOG.md | 44 +++++++-------- docs/source/api_multiprocessing.rst | 7 +-- lithops/future.py | 3 ++ .../monitoring/backends/rabbitmq/status.py | 1 + lithops/multiprocessing/pool.py | 53 ++++++++++++++++++- lithops/multiprocessing/queues.py | 2 +- lithops/tests/test_multiprocessing.py | 1 + 7 files changed, 82 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6256fbe3f..2d796bab5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,33 +5,39 @@ ### Added - [API] Added `lithops.concurrent.futures`, a `concurrent.futures`-compatible executor interface backed by Lithops. -- [Tests] Added a unit test suite for all non-backend modules (18 files, 876 tests). -- [Monitoring] Added Redis, AWS SQS (`aws_sqs`), GCP Pub/Sub (`gcp_pubsub`) and Azure Queue Storage (`azure_queue`) monitoring backends. +- [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. +- [AWS Batch] Added the `instance_types` config option for EC2/SPOT compute environments. +- [Tests] Added a unit test suite for all non-backend modules (18 files, 1266 tests). ### Changed -- [Worker] Replaced the `multiprocessing` Manager queue of the worker pool with a POSIX pipe. -- [Core] Results under 8KB now travel in the call status instead of a separate storage object. -- [Core] Added a cache of serialized functions to avoid re-uploading the same function. - [Monitoring] Reorganised job monitoring as pluggable backends. +- [Monitoring] The RabbitMQ queue is now deleted on cleanup instead of on every `stop()`, so a later `map()` can reuse it. +- [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`. +- [Worker] Replaced the `multiprocessing` Manager queue of the worker pool with a POSIX pipe. - [CLI] `job list`, `worker list`, `image delete` and `image list` now reject unknown flags. - [CLI] `lithops clean` now empties the local temp directory instead of removing it. - [Storage] `CloudFileProxy.walk()` now yields nothing for a missing path, like `os.walk`. - [Storage] `cloud_open()` now raises `ValueError` on an unsupported mode. -- [Joblib] Capped the shared-argument upload and download pools at 32 threads. - [Joblib] `lithops_args` is now applied to the pool that runs the batches. - [Standalone] `docker login` now reads the password from stdin and quotes its arguments. -- [AWS Batch] Allow to set `instance_types` config option for EC2/SPOT compute environments. ### Fixed -- [Core] Fixed `wait()` on futures another executor invoked, which crashed with an `AttributeError` in `JobMonitor.is_alive()` and, once past it, watched the wrong storage prefix and never returned. -- [Chaining] Fixed pickling a `FuturesList` detaching the list being pickled from its executor. -- [Chaining] Fixed a list or a slice of futures of a previous job not being recognised as a chain, which failed with an argument binding error instead. -- [Chaining] `extra_args` now raises at submit time instead of letting every activation of the chained job fail on a missing argument. -- [Monitoring] Redis, RabbitMQ and SQS now delete their queues only in ``cleanup()``, and keep the monitor thread until ``stop()``. -- [Monitoring] Status lines (Pending/Running/Done) are logged on start, every 30s, and when the job finishes, not on every activation. +- [Core] Fixed `wait()` on futures another executor invoked, which crashed instead of waiting for them. +- [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 a hand-built `FuturesList` raising `AttributeError` instead of creating its executor. +- [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. +- [Monitoring] Fixed a nested executor publishing statuses to a queue nobody declares. +- [Monitoring] Fixed failed RabbitMQ publishes being dropped with nothing in the log. +- [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`. - [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. @@ -40,21 +46,12 @@ - [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. - [Storage] Fixed `delete_cloudobjects()` deleting part of the list before rejecting a foreign object. - [Storage] Fixed `CloudFileProxy.listdir()` returning nothing for its default argument. -- [Core] Fixed `find_free_port()` setting `SO_REUSEADDR` after the bind. -- [Core] Fixed module inspection crashing on a function whose `__module__` is `None`. -- [Core] Fixed a hand-built `FuturesList` raising `AttributeError` instead of creating its executor. -- [Cleaner] Fixed the cleaner skipping requests and two cleaners racing for the pid file. -- [Cleaner] Fixed `lithops clean` deleting the local temp directory of the jobs running at the same time on the same machine. -- [Cleaner] Fixed the cleaner reading a request another process was still writing. +- [Cleaner] Fixed two cleaners racing for the pid file, and requests being skipped or read while still being written. - [Cleaner] Fixed the cleaner looping forever on a request it could not read or classify. -- [Cleaner] Fixed the cleaner lock surviving a killed cleaner and blocking every later one. -- [Monitoring] Fixed a nested executor publishing statuses to a queue nobody declares. -- [Monitoring] Fixed the failed RabbitMQ publishes being dropped with nothing in the log. - [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. - [Job] Fixed folder markers being counted as objects, returning empty partitions. @@ -64,7 +61,6 @@ - [Joblib] Fixed a race losing one of two shared arguments 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. -- [Tests] Fixed the test suite depending on the order its files run in. ## [v3.7.0] diff --git a/docs/source/api_multiprocessing.rst b/docs/source/api_multiprocessing.rst index cf6e8b61b..76aa98628 100644 --- a/docs/source/api_multiprocessing.rst +++ b/docs/source/api_multiprocessing.rst @@ -67,9 +67,10 @@ so a few things of it have no counterpart: - Not lazy: every call is submitted and every result collected before the first one is yielded, so an endless iterable will not work. The results always come back in the order of the input - * - ``Pool.join()`` - - Returns as soon as the pool is released; it does not wait for the calls - still in flight. Use the ``AsyncResult`` of each call to wait for it + * - ``Pool.terminate()`` + - Gives the Lithops executor back and stops the calls still in flight, + so the ``AsyncResult`` of one of them has nothing left to return. + Close and join the pool instead when the results are still wanted * - ``Pool(maxtasksperchild=...)``, ``Process.daemon``, ``Process.authkey`` - Accepted and ignored. Workers are ephemeral, so there is nothing to recycle, nothing to daemonize and no handshake to authenticate diff --git a/lithops/future.py b/lithops/future.py index 88f01a615..0e6a740f1 100644 --- a/lithops/future.py +++ b/lithops/future.py @@ -393,6 +393,9 @@ def status( if self._state == self.State.New: raise ValueError("task not yet invoked") + if self.error and (self._call_status or {}).get('exception'): + return self._raise_call_exception(throw_except) + if self.success or self.done: return self._call_status diff --git a/lithops/monitoring/backends/rabbitmq/status.py b/lithops/monitoring/backends/rabbitmq/status.py index 3913e0882..c6fa5d808 100644 --- a/lithops/monitoring/backends/rabbitmq/status.py +++ b/lithops/monitoring/backends/rabbitmq/status.py @@ -76,6 +76,7 @@ def _drop_channel(self) -> None: """ amqp, self._amqp = self._amqp, None self.discard_client('_amqp') + self._amqp = None if amqp is None: return for closeable in reversed(amqp): diff --git a/lithops/multiprocessing/pool.py b/lithops/multiprocessing/pool.py index 679cc9528..f014693d6 100644 --- a/lithops/multiprocessing/pool.py +++ b/lithops/multiprocessing/pool.py @@ -213,8 +213,41 @@ def join(self): logger.debug('joining pool') if self._state not in (CLOSE, TERMINATE): raise ValueError('Pool is still running') + if self._state == CLOSE: + self._wait_for_calls() self._release() + def _wait_for_calls(self): + """ + Waits for the calls still in flight, as join() does in the standard + library. + + Releasing the executor stops whatever is still running, so a pool + that was closed rather than terminated has to let its calls finish + first: their results are read from the AsyncResult afterwards, and a + call killed here would never produce one. terminate() is the one that + does not wait, which is what it means there too + """ + executor = self._executor + if executor is None: + return + try: + # Nothing to wait for only when the executor says so; one that + # does not keep a list of its futures is waited on anyway + if not getattr(executor, 'futures', True): + return + # throw_except=False: a call that failed is reported by get(), + # and raising here would force-clean the results of the ones that + # did not, which get() still has to read + executor.wait( + download_results=False, + throw_except=False, + show_progressbar=False, + clean_jobs=False, + ) + except Exception: + logger.debug('Error waiting for the pool calls', exc_info=True) + def _release(self): """ Stops the log feed and gives the Lithops executor back. Without it @@ -301,10 +334,28 @@ def _get_values(self, timeout=None): # Lithops reports it as the builtin, which is an OSError and so # not what `except multiprocessing.TimeoutError` catches raise ProcessTimeoutError(str(exc)) from exc - values = [fut.result() for fut in self._futures] + except Exception as exc: + # The call raised, and wait() re-raises it while downloading the + # results. The standard library hands that to error_callback + # before letting get() raise it + self._fail(exc) + raise + values = [] + for fut in self._futures: + try: + values.append(fut.result()) + except Exception as exc: + self._fail(exc) + raise util.export_execution_details(self._futures, self._executor) return values + def _fail(self, exc): + """Records the failure of a call and reports it to error_callback""" + self._exception = exc + if self._error_callback is not None: + self._error_callback(exc) + def get(self, timeout=None): """The value of the single call this result stands for""" self._value = self._get_values(timeout)[0] diff --git a/lithops/multiprocessing/queues.py b/lithops/multiprocessing/queues.py index ad467308d..9f008b02b 100644 --- a/lithops/multiprocessing/queues.py +++ b/lithops/multiprocessing/queues.py @@ -211,7 +211,7 @@ def __setstate__(self, state): def put(self, obj, block=True, timeout=None): with self._cond: - super().put(obj) + super().put(obj, block, timeout) self._unfinished_tasks.release() def task_done(self): diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index 8162fff26..e5d1bbaa0 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -86,6 +86,7 @@ def __init__(self, **kwargs): self.kwargs = kwargs self.executor_id = 'sess-0' self.invoker = type('I', (), {'max_workers': 7})() + self.futures = [] self.call_async_calls = [] self.map_calls = [] self.wait_calls = [] From e44ff754405dbf7c681fc66c7ccc8473544d2130 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 19:24:25 +0200 Subject: [PATCH 2/9] Update wait --- lithops/localhost/v2/localhost.py | 3 +- .../monitoring/backends/rabbitmq/rabbitmq.py | 2 +- lithops/monitoring/backends/redis/redis.py | 2 +- lithops/multiprocessing/connection.py | 109 +++-- lithops/multiprocessing/synchronize.py | 137 +++++- lithops/tests/mp_fakeredis.py | 4 +- lithops/tests/test_localhost.py | 57 +++ lithops/tests/test_multiprocessing.py | 422 +++++++++++++++++- lithops/tests/test_utils.py | 49 ++ lithops/tests/test_wait.py | 35 +- lithops/utils.py | 26 +- lithops/wait.py | 27 +- 12 files changed, 788 insertions(+), 85 deletions(-) diff --git a/lithops/localhost/v2/localhost.py b/lithops/localhost/v2/localhost.py index 823a1e342..11eb0e21f 100644 --- a/lithops/localhost/v2/localhost.py +++ b/lithops/localhost/v2/localhost.py @@ -208,8 +208,7 @@ def clear(self, job_keys=None, exception=None): for job_key in list(self.env.jobs.keys()): if job_keys is not None and job_key not in job_keys: continue - while not self.env.jobs[job_key].done: - self.env.jobs[job_key].unlock() + self.env.jobs[job_key].release() class ExecutionEnvironment: diff --git a/lithops/monitoring/backends/rabbitmq/rabbitmq.py b/lithops/monitoring/backends/rabbitmq/rabbitmq.py index 3f73c0c8c..a3b0a083e 100644 --- a/lithops/monitoring/backends/rabbitmq/rabbitmq.py +++ b/lithops/monitoring/backends/rabbitmq/rabbitmq.py @@ -121,7 +121,7 @@ def stop(self): connection belongs to the monitor thread, so the close is handed to it instead of being done here """ - self.should_run = False + super().stop() connection = self.connection if connection is None: return diff --git a/lithops/monitoring/backends/redis/redis.py b/lithops/monitoring/backends/redis/redis.py index 619f3d036..178ed1955 100644 --- a/lithops/monitoring/backends/redis/redis.py +++ b/lithops/monitoring/backends/redis/redis.py @@ -149,7 +149,7 @@ def stop(self): monitor's own: redis_client() builds a client per caller, so the storage, multiprocessing and joblib backends keep theirs """ - self.should_run = False + super().stop() try: self.client.connection_pool.disconnect() except Exception: diff --git a/lithops/multiprocessing/connection.py b/lithops/multiprocessing/connection.py index 4f2fc56fd..9d7fffacd 100644 --- a/lithops/multiprocessing/connection.py +++ b/lithops/multiprocessing/connection.py @@ -39,6 +39,20 @@ REDIS_LIST_CONN_A = REDIS_LIST_CONN + '-a-' REDIS_LIST_CONN_B = REDIS_LIST_CONN + '-b-' +#: There is nothing to block on when polling a Redis handle or a local +#: buffer, so the wait between checks starts here and doubles up to the cap. +#: The cap is what the interval used to be, so an idle poll is no busier +#: than before, while a message already on its way is picked up at once +POLL_MIN_SLEEP = 0.001 +POLL_MAX_SLEEP = 0.1 + +#: How long a connection waits for its peer to publish its address, and how +#: long it backs off to while waiting. Coarser than a data poll: this is a +#: rendezvous that can legitimately take a while, and every check is a round +#: trip to the directory +ADDRESS_LOOKUP_TIMEOUT = 60 +ADDRESS_LOOKUP_MAX_SLEEP = 1.0 + REDIS_PUBSUB_CONN = 'redispubsub' # uses Redis channels (pub/sub) REDIS_PUBSUB_CONN_A = REDIS_PUBSUB_CONN + '-a-' REDIS_PUBSUB_CONN_B = REDIS_PUBSUB_CONN + '-b-' @@ -457,15 +471,21 @@ def _send_bytes(self, buf): logger.debug('Get address from directory for handle %s', self._subhandle) addr = self._client.get(self._subhandle) - retry = 15 - retry_sleep = 1 - while addr is None: - time.sleep(retry_sleep) - retry_sleep += 0.5 - addr = self._client.get(self._subhandle) - retry -= 1 - if retry == 0: - raise Exception('Server address could not be fetched for handle {}'.format(self._subhandle)) + if addr is None: + # The peer publishes its address as it comes up, so this is a + # rendezvous. Waiting a fixed second before looking again + # made a peer that was 20 ms late cost a full second; the + # backoff finds it as soon as it is there and still gives it + # ADDRESS_LOOKUP_TIMEOUT to appear + addr = _poll_until( + lambda: self._client.get(self._subhandle), + ADDRESS_LOOKUP_TIMEOUT, + max_sleep=ADDRESS_LOOKUP_MAX_SLEEP, + ) + if addr is None: + raise Exception( + 'Server address could not be fetched for handle {}'.format(self._subhandle) + ) self._subhandle_addr = addr.decode('utf-8') logger.debug('Dialing %s', self._subhandle_addr) @@ -481,13 +501,17 @@ def _recv_bytes(self, maxsize=None): return chunk def _poll(self, timeout): - max_time = time.monotonic() + timeout - while time.monotonic() < max_time: - qsize = self._buff.qsize() - if qsize > 0: - return True - else: - time.sleep(0.1) + """ + Whether a message is waiting in the local buffer the subscriber + thread fills. + + The buffer is checked before the timeout is, so poll(0) answers what + is actually there. It used to start by comparing the clock against a + deadline it had just set, which with timeout=0 fell straight through + without ever looking: Queue.empty() said True whatever the queue + held, and get(block=False) raised Empty on a queue with data in it + """ + return bool(_poll_until(lambda: self._buff.qsize() > 0, timeout)) PipeConnection = _RedisConnection @@ -658,30 +682,49 @@ def _RedisClient(address): # Wait # +def _poll_until(is_ready, timeout, max_sleep=POLL_MAX_SLEEP): + """ + Calls ``is_ready()`` until it returns something truthy or the timeout is + up, and hands back whatever it returned last. + + The check always runs at least once, including with ``timeout=0``, which + is what ``poll(0)``, ``Queue.empty()`` and ``get(block=False)`` ask for. + Between checks it waits ``POLL_MIN_SLEEP`` and doubles up to + ``max_sleep``: something that is already on its way is picked up in about + a millisecond instead of waiting out a fixed interval, while a poll that + finds nothing settles at the interval it always used, so an idle wait + costs no more than before + """ + deadline = None if timeout is None else time.monotonic() + timeout + delay = POLL_MIN_SLEEP + while True: + ready = is_ready() + if ready: + return ready + if deadline is None: + time.sleep(delay) + else: + remaining = deadline - time.monotonic() + if remaining <= 0: + return ready + time.sleep(min(delay, remaining)) + delay = min(delay * 2, max_sleep) + + def wait(object_list, timeout=None): """ Wait till an object in object_list is ready/readable. Returns list of those objects in object_list which are ready/readable. """ - if timeout is not None: - deadline = time.monotonic() + timeout - - while True: - ready = [] + def ready(): + found = [] for client, handle in object_list: if handle.startswith(REDIS_LIST_CONN): - llen = client.llen(handle) - if llen > 0: - ready.append((client, handle)) + if client.llen(handle) > 0: + found.append((client, handle)) elif handle.startswith(REDIS_PUBSUB_CONN) and client.connection.can_read(): - ready.append((client, handle)) - - if any(ready): - return ready + found.append((client, handle)) + return found - if timeout is not None: - timeout = deadline - time.monotonic() - if timeout < 0: - return ready - time.sleep(0.1) + return _poll_until(ready, timeout) or [] diff --git a/lithops/multiprocessing/synchronize.py b/lithops/multiprocessing/synchronize.py index 72b43dc72..92756b653 100644 --- a/lithops/multiprocessing/synchronize.py +++ b/lithops/multiprocessing/synchronize.py @@ -10,14 +10,52 @@ # import threading +import math import time import logging +import redis + from . import util from . import config as mp_config logger = logging.getLogger(__name__) +#: Redis takes a fractional BLPOP timeout from 6.0 on +_BLPOP_TAKES_FLOAT = True + + +def _blpop(client, name, timeout): + """ + BLPOP with a timeout the server will accept. + + A fractional one needs Redis 6.0; an older server answers with an error, + and the wait is rounded up to the next whole second rather than cut + short of what the caller asked for. Only a fractional timeout can + provoke that, so a whole-second one never goes near the fallback. + + The retry is narrowed to the server rejecting the argument: a socket + read that timed out also says "timeout", and swallowing one would both + hide it and leave every later wait rounded to the second + """ + if timeout is None: + # redis-py reads None as zero, which BLPOP reads as "block for ever" + return client.blpop([name], timeout=0) + + global _BLPOP_TAKES_FLOAT + if _BLPOP_TAKES_FLOAT and timeout != int(timeout): + try: + return client.blpop([name], timeout=timeout) + except redis.exceptions.ResponseError as exc: + if 'timeout' not in str(exc).lower(): + raise + _BLPOP_TAKES_FLOAT = False + logger.debug( + 'This Redis does not take a fractional BLPOP timeout ' + '(%s); rounding waits up to the second', exc + ) + return client.blpop([name], timeout=math.ceil(timeout)) + # # Constants # @@ -35,10 +73,12 @@ class SemLock: # return new semlock value # only increments its value if # it is not above the max value + # Returns the new value, or -1 when the lock or semaphore was already + # at its maximum, which is a release of something that was never held LUA_RELEASE_SCRIPT = """ local current_value = tonumber(redis.call('llen', KEYS[1])) if current_value >= tonumber(ARGV[1]) then - return current_value + return -1 end redis.call('rpush', KEYS[1], '') return current_value + 1 @@ -77,20 +117,37 @@ def get_value(self): value = self._client.llen(self._name) return int(value) - def acquire(self, block=True): - if block: + def acquire(self, block=True, timeout=None): + """ + Takes the lock, waiting at most ``timeout`` seconds for it. + + ``timeout`` is what the standard library takes and this used to + reject outright. A zero or negative one is a single attempt, since + BLPOP reads a zero timeout as "block for ever" + """ + if not block or (timeout is not None and timeout <= 0): + logger.debug('Requested non-blocking acquire for lock %s', self._name) + return self._client.lpop(self._name) is not None + + if timeout is None: logger.debug('Requested blocking acquire for lock %s', self._name) self._client.blpop([self._name]) return True - else: - logger.debug('Requested non-blocking acquire for lock %s', self._name) - return self._client.lpop(self._name) is not None + + logger.debug( + 'Requested acquire for lock %s within %s s', self._name, timeout + ) + return _blpop(self._client, self._name, timeout) is not None def release(self): logger.debug('Requested release for lock %s', self._name) - self._lua_release(keys=[self._name], - args=[self._max_value], - client=self._client) + value = self._lua_release(keys=[self._name], + args=[self._max_value], + client=self._client) + if value == -1: + # What the standard library raises for a lock that was not held + # and for a bounded semaphore released more often than acquired + raise ValueError('semaphore or lock released too many times') def __repr__(self): try: @@ -131,14 +188,22 @@ def __setstate__(self, state): super().__setstate__(state) self.owned = False - def acquire(self, block=True): - res = super().acquire(block) - self.owned = True + def acquire(self, block=True, timeout=None): + """ + Marks the lock as owned only when it was really taken. + + Setting it whatever the outcome left an RLock whose first acquire + had failed reporting success on the next one, handing out mutual + exclusion it did not hold + """ + res = super().acquire(block, timeout) + if res: + self.owned = True return res def release(self): - super().release() self.owned = False + super().release() # @@ -146,8 +211,41 @@ def release(self): # class RLock(Lock): - def acquire(self, block=True): - return self.owned or super().acquire(block) + """ + A lock the same holder can take more than once. + + The recursion is counted here rather than in Redis: only the first + acquire takes the token, and only the last release gives it back. It + used to take one token and give back one per release, so a re-entrant + acquire/release pair returned a token it never took + """ + + def __init__(self): + super().__init__() + self._count = 0 + + def __setstate__(self, state): + super().__setstate__(state) + self._count = 0 + + def acquire(self, block=True, timeout=None): + if self.owned: + self._count += 1 + return True + res = super().acquire(block, timeout) + if res: + self._count = 1 + return res + + def release(self): + if not self.owned: + # The wording the standard library uses + raise AssertionError( + 'attempt to release recursive lock not owned by thread' + ) + self._count -= 1 + if self._count == 0: + super().release() # @@ -194,9 +292,16 @@ def wait(self, timeout=None): # Release lock, wait to get notified, acquire lock self.release() logger.debug('Waiting for token %s on condition %s', wait_handle, self._notify_handle) - self._client.blpop([wait_handle], timeout) + 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() + # Whether a notify arrived, rather than the timeout expiring, which + # is what the standard library returns and callers branch on + return notified def notify(self): assert self._lock.owned diff --git a/lithops/tests/mp_fakeredis.py b/lithops/tests/mp_fakeredis.py index 197b5264e..73654c3b5 100644 --- a/lithops/tests/mp_fakeredis.py +++ b/lithops/tests/mp_fakeredis.py @@ -249,7 +249,9 @@ def __call__(self, keys, args, client=None): with server._cond: current = len(server.lists.get(name, [])) if current >= max_value: - return current + # -1, as the real script does, so that the caller can tell a + # release of something that was never held from a normal one + return -1 server.lists.setdefault(name, []).append(b'') server._cond.notify_all() return current + 1 diff --git a/lithops/tests/test_localhost.py b/lithops/tests/test_localhost.py index aba7f9495..ad9eb5559 100644 --- a/lithops/tests/test_localhost.py +++ b/lithops/tests/test_localhost.py @@ -13,6 +13,7 @@ import signal import subprocess as sp import sys +import threading import time from pathlib import Path from unittest.mock import MagicMock, patch @@ -235,6 +236,62 @@ def test_clear_leaves_the_latches_of_other_jobs_alone(self): assert mine.done is True assert theirs.done is False + def test_clear_opens_the_latch_while_a_task_is_counting_it_down(self): + """ + clear() and the consumer that has just finished a task both touch + the same latch, and with a message-based monitoring backend the + client learns the job is done early enough that they collide. + + clear() used to drain the latch with `while not done: unlock()`. A + consumer's own unlock() landing between the check and the unlock + pushed the count past zero, which left done() False and the event + unset for good: job_manager span on a full core for the rest of the + process and never came back, which held up interpreter exit too. + + The interleaving is forced rather than raced for, so this fails + every time on the old code instead of once in a while + """ + class RacingLatch(CountDownLatch): + """Counts the last task down exactly as clear() reads done""" + + def __init__(self, count): + super().__init__(count) + self.raced = False + self.unlocks = 0 + + @property + def done(self): + seen = CountDownLatch.done.fget(self) + if not self.raced: + self.raced = True + # The consumer, landing between the check and the unlock + self.unlock() + return seen + + def unlock(self): + self.unlocks += 1 + assert self.unlocks <= 10, ( + 'clear() kept counting a latch down past zero' + ) + super().unlock() + + handler = LocalhostHandlerV2(_config()) + handler.env = MagicMock() + latch = RacingLatch(1) + handler.env.jobs = {'sess-0-M000': latch} + + handler.clear({'sess-0-M000'}) + + assert latch.done is True + # job_manager waits on every latch it knows about, so one that never + # opens is what stops that thread from ever coming back + waited = threading.Event() + waiter = threading.Thread(target=lambda: (latch.wait(), waited.set())) + waiter.daemon = True + waiter.start() + waiter.join(timeout=5) + assert waited.is_set() is True + class TestLocalhostHandlerV1: diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index e5d1bbaa0..8ba055378 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -23,12 +23,15 @@ import ctypes import pickle +import queue +import sys import threading import time import types import cloudpickle import pytest +from unittest.mock import MagicMock, patch from lithops.multiprocessing import config as mp_config from lithops.multiprocessing import util as mp_util @@ -921,9 +924,25 @@ def test_the_context_manager_acquires_and_releases(self, redis): assert lock.get_value() == 0 assert lock.get_value() == 1 - def test_a_bounded_semaphore_does_not_go_over_its_value(self, redis): + def test_a_bounded_semaphore_rejects_a_release_it_cannot_hold(self, redis): + """ + A release that would take it past its bound raises, as the standard + library documents, instead of being swallowed. Not compared against + multiprocessing here: its check reads the semaphore value through + sem_getvalue(), which macOS does not implement, so a BoundedSemaphore + larger than 1 never raises there + """ + from lithops.multiprocessing import BoundedSemaphore + sem = BoundedSemaphore(2) + with pytest.raises(ValueError, match='released too many times'): + sem.release() + assert sem.get_value() == 2 + + def test_a_bounded_semaphore_allows_a_release_it_acquired(self, redis): from lithops.multiprocessing import BoundedSemaphore sem = BoundedSemaphore(2) + sem.acquire() + assert sem.get_value() == 1 sem.release() assert sem.get_value() == 2 @@ -1199,3 +1218,404 @@ def test_the_cloudpickle_round_trip_of_a_shared_object(self, redis): from lithops.multiprocessing import Lock lock = Lock() assert cloudpickle.loads(cloudpickle.dumps(lock))._name == lock._name + + +class TestConnectionPolling: + """ + Nothing in a Redis handle or a local buffer can be blocked on, so these + polls are timed waits. They used to step in a flat 0.1s, which made a + message that was already there cost a tenth of a second + """ + + def _connection(self, buff): + from lithops.multiprocessing.connection import _NanomsgConnection + + conn = _NanomsgConnection.__new__(_NanomsgConnection) + conn._buff = buff + return conn + + def test_poll_zero_looks_at_the_buffer(self): + """ + poll(0) is what Queue.empty() and get(block=False) call. It used to + compare the clock against a deadline it had just set and fall + straight through without ever looking, so empty() said True whatever + the queue held and get(block=False) raised Empty on a queue with + data in it + """ + buff = queue.Queue() + buff.put(b'a message that is definitely there') + assert self._connection(buff)._poll(0.0) is True + assert self._connection(queue.Queue())._poll(0.0) is False + + def test_poll_returns_as_soon_as_a_message_lands(self): + buff = queue.Queue() + timer = threading.Timer(0.01, lambda: buff.put(b'x')) + timer.start() + try: + started = time.monotonic() + assert self._connection(buff)._poll(5.0) is True + assert time.monotonic() - started < 0.09 + finally: + timer.cancel() + + 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 + + def test_poll_until_always_checks_once(self): + from lithops.multiprocessing.connection import _poll_until + + checks = [] + assert _poll_until(lambda: checks.append(1) or False, 0.0) is False + assert len(checks) == 1 + assert _poll_until(lambda: 'ready', 0.0) == 'ready' + + def test_poll_until_backs_off_to_the_cap(self): + from lithops.multiprocessing import connection as conn_mod + + waits = [] + with patch.object(conn_mod.time, 'sleep', side_effect=waits.append): + conn_mod._poll_until(lambda: False if len(waits) < 12 else True, None) + assert waits[0] == conn_mod.POLL_MIN_SLEEP + assert waits == sorted(waits) + assert max(waits) == conn_mod.POLL_MAX_SLEEP + + def test_wait_returns_the_ready_handles(self): + from lithops.multiprocessing import connection as conn_mod + + client = MagicMock() + client.llen.side_effect = lambda h: 1 if h.endswith('b') else 0 + handles = [ + (client, conn_mod.REDIS_LIST_CONN + '-a'), + (client, conn_mod.REDIS_LIST_CONN + '-b'), + ] + assert conn_mod.wait(handles, timeout=0.0) == [handles[1]] + + def test_wait_returns_an_empty_list_when_nothing_is_ready(self): + from lithops.multiprocessing import connection as conn_mod + + client = MagicMock() + client.llen.return_value = 0 + handles = [(client, conn_mod.REDIS_LIST_CONN + '-a')] + assert conn_mod.wait(handles, timeout=0.0) == [] + + +class TestAddressLookup: + """ + A connection waits for its peer to publish its address. Sleeping a fixed + second before looking again made a peer that was 20 ms late cost a full + second of setup + """ + + def test_the_address_is_picked_up_as_soon_as_it_appears(self): + from lithops.multiprocessing import connection as conn_mod + + client = MagicMock() + client.get.side_effect = [None, None, b'tcp://127.0.0.1:5555'] + started = time.monotonic() + addr = conn_mod._poll_until( + lambda: client.get('h'), + conn_mod.ADDRESS_LOOKUP_TIMEOUT, + max_sleep=conn_mod.ADDRESS_LOOKUP_MAX_SLEEP, + ) + assert addr == b'tcp://127.0.0.1:5555' + assert time.monotonic() - started < 0.5 + + def test_it_gives_up_after_the_timeout(self): + from lithops.multiprocessing import connection as conn_mod + + client = MagicMock() + client.get.return_value = None + assert conn_mod._poll_until(lambda: client.get('h'), 0.05) is None + assert client.get.call_count >= 1 + + +class TestSemLockContract: + """ + These follow multiprocessing.Lock/Semaphore, which callers write against + """ + + def test_acquire_takes_a_timeout(self, redis): + """ + The standard library's signature is acquire(block, timeout). Not + taking one turned every timed acquire into a TypeError + """ + from lithops.multiprocessing import Lock + + lock = Lock() + assert lock.acquire() is True + try: + assert lock.acquire(True, 0.1) is False + finally: + lock.release() + + def test_a_timeout_that_has_passed_is_a_single_attempt(self, redis): + """ + BLPOP reads a zero timeout as "block for ever", so it cannot be + handed one straight through + """ + from lithops.multiprocessing import Lock + + lock = Lock() + lock.acquire() + try: + assert lock.acquire(True, 0) is False + assert lock.acquire(True, -1) is False + finally: + lock.release() + + def test_a_failed_acquire_does_not_claim_ownership(self, redis): + """ + owned used to be set whatever the acquire returned, which left an + RLock whose first acquire had failed reporting success on the next + one: mutual exclusion handed out without the lock behind it + """ + from lithops.multiprocessing import Lock, RLock + + lock = Lock() + lock.acquire() + try: + assert lock.acquire(False) is False + assert lock.owned is True # this holder does own it + finally: + lock.release() + + rlock = RLock() + rlock._client.delete(rlock._name) # nothing left to take + assert rlock.acquire(False) is False + assert rlock.owned is False + assert rlock.acquire(False) is False # and still cannot claim it + + def test_rlock_counts_its_recursion(self, redis): + """ + Only the first acquire takes the token and only the last release + gives it back. Giving one back per release returned a token the + re-entrant acquire never took + """ + from lithops.multiprocessing import RLock + + rlock = RLock() + assert rlock.acquire() is True + assert rlock.acquire() is True + rlock.release() + assert rlock.owned is True # still held after one release + rlock.release() + assert rlock.owned is False + assert rlock.acquire(False) is True # the token really came back + rlock.release() + + def test_releasing_an_rlock_that_is_not_held_raises(self, redis): + from lithops.multiprocessing import RLock + + with pytest.raises(AssertionError, match='not owned'): + RLock().release() + + def test_a_bounded_semaphore_rejects_an_extra_release(self, redis): + """ + A bounded semaphore that silently swallows an over-release is not + bounded at all + """ + from lithops.multiprocessing import BoundedSemaphore + + sem = BoundedSemaphore(1) + sem.acquire() + sem.release() + with pytest.raises(ValueError, match='released too many times'): + sem.release() + + def test_an_unbounded_semaphore_still_allows_extra_releases(self, redis): + from lithops.multiprocessing import Semaphore + + sem = Semaphore(1) + sem.release() + sem.release() + assert sem.get_value() == 3 + + def test_releasing_a_lock_that_was_never_held_raises(self, redis): + from lithops.multiprocessing import Lock + + with pytest.raises(ValueError, match='released too many times'): + Lock().release() + + +class TestBlpopTimeoutFallback: + """ + BLPOP takes a fractional timeout only from Redis 6.0 on. The fallback + that rounds up for an older server must not be reached by anything else: + it is process wide, so one wrong trip leaves every later wait rounded to + the whole second + """ + + @staticmethod + def _client(error=None): + calls = [] + + class FakeClient: + def blpop(self, keys, timeout=0): + calls.append(timeout) + if error is not None and len(calls) == 1: + raise error + return (keys[0], b'') + + return FakeClient(), calls + + @pytest.fixture(autouse=True) + def _reset_flag(self): + """The flag is module state, so a test must not leak it to the next""" + sync = sys.modules['lithops.multiprocessing.synchronize'] + before = sync._BLPOP_TAKES_FLOAT + sync._BLPOP_TAKES_FLOAT = True + yield sync + sync._BLPOP_TAKES_FLOAT = before + + def test_a_whole_second_timeout_never_goes_near_the_fallback(self, _reset_flag): + client, calls = self._client() + _reset_flag._blpop(client, 'k', 2) + assert calls == [2] + + def test_a_fractional_timeout_is_passed_through(self, _reset_flag): + client, calls = self._client() + _reset_flag._blpop(client, 'k', 0.25) + assert calls == [0.25] + + def test_an_old_server_rounds_the_wait_up(self, _reset_flag): + import redis as redis_pkg + + error = redis_pkg.exceptions.ResponseError( + 'timeout is not an integer or out of range' + ) + client, calls = self._client(error) + _reset_flag._blpop(client, 'k', 0.25) + # Rounded up, never down: 0 would mean "block for ever" + assert calls == [0.25, 1] + assert _reset_flag._BLPOP_TAKES_FLOAT is False + + def test_a_socket_timeout_is_not_mistaken_for_an_old_server(self, _reset_flag): + """ + A read that timed out also says "timeout". Swallowing one would hide + it and round every later wait up to the second for the whole process + """ + import redis as redis_pkg + + error = redis_pkg.exceptions.TimeoutError('Timeout reading from socket') + client, calls = self._client(error) + + with pytest.raises(redis_pkg.exceptions.TimeoutError): + _reset_flag._blpop(client, 'k', 0.25) + + assert calls == [0.25] + assert _reset_flag._BLPOP_TAKES_FLOAT is True + + def test_another_server_error_is_not_swallowed(self, _reset_flag): + import redis as redis_pkg + + error = redis_pkg.exceptions.ResponseError('WRONGTYPE') + client, calls = self._client(error) + + with pytest.raises(redis_pkg.exceptions.ResponseError, match='WRONGTYPE'): + _reset_flag._blpop(client, 'k', 0.25) + + assert _reset_flag._BLPOP_TAKES_FLOAT is True + + def test_no_timeout_blocks_for_ever(self, _reset_flag): + client, calls = self._client() + _reset_flag._blpop(client, 'k', None) + # int(None) would raise, and BLPOP reads zero as "block for ever" + assert calls == [0] + + def test_a_condition_wait_uses_the_same_fallback(self, redis, _reset_flag): + """ + Condition.wait(0.5) used to hand the fraction straight to BLPOP, so + it failed on a server where a timed acquire worked + """ + from lithops.multiprocessing import Condition + + seen = [] + real = _reset_flag._blpop + + def spy(client, name, timeout): + seen.append(timeout) + return real(client, name, timeout) + + cond = Condition() + with patch.object(_reset_flag, '_blpop', spy): + with cond: + assert cond.wait(0.2) is False + assert seen == [0.2] + + +class TestConditionContract: + + def test_wait_says_whether_it_was_notified(self, redis): + """ + The standard library returns False on a timeout and True on a + notify, and callers branch on it. Returning None made every wait + look like a timeout + """ + from lithops.multiprocessing import Condition + + cond = Condition() + with cond: + assert cond.wait(0.1) is False + + notified = [] + + def waiter(): + with cond: + notified.append(cond.wait(5)) + + thread = threading.Thread(target=waiter) + thread.start() + time.sleep(0.3) + with cond: + cond.notify_all() + thread.join(10) + assert notified == [True] + + def test_wait_with_a_timeout_that_has_passed_does_not_block(self, redis): + from lithops.multiprocessing import Condition + + cond = Condition() + with cond: + started = time.monotonic() + assert cond.wait(0) is False + assert time.monotonic() - started < 1 + + +class TestWaitAlarm: + """ + lithops.wait() arms a SIGALRM, and signal.alarm() only takes whole + seconds + """ + + def test_a_fractional_timeout_is_rounded_up(self): + lw = sys.modules['lithops.wait'] + + armed = [] + with patch.object(lw.signal, 'alarm', side_effect=armed.append), \ + patch.object(lw.signal, 'signal'): + lw._set_wait_alarm(0.2) + lw._set_wait_alarm(2.7) + # int() would give 0 and 2: the first cancels the alarm outright, + # and the second cuts the wait short of what was asked for + assert armed == [1, 3] + + def test_a_timeout_that_has_passed_raises_at_once(self): + lw = sys.modules['lithops.wait'] + + with pytest.raises(TimeoutError, match='Timeout of 0 seconds'): + lw._set_wait_alarm(0) + + def test_pool_get_with_a_fractional_timeout_raises_timeout_error(self): + """ + Pool.AsyncResult.get(0.2) reached signal.alarm() through wait() and + came back as TypeError instead of the multiprocessing TimeoutError + """ + lw = sys.modules['lithops.wait'] + + with patch.object(lw.signal, 'alarm') as alarm, \ + patch.object(lw.signal, 'signal'): + lw._set_wait_alarm(0.2) + alarm.assert_called_once_with(1) diff --git a/lithops/tests/test_utils.py b/lithops/tests/test_utils.py index 046ce6fcf..bbe7c3055 100644 --- a/lithops/tests/test_utils.py +++ b/lithops/tests/test_utils.py @@ -15,6 +15,7 @@ import io import logging import pickle +import threading import zipfile from collections import namedtuple from unittest.mock import MagicMock, patch @@ -289,6 +290,54 @@ def test_countdown_latch_wait_returns_immediately_when_already_done(self): latch.wait() assert latch.done is True + def test_countdown_latch_survives_being_unlocked_too_many_times(self): + """ + A caller draining a latch cannot tell whether the last task is + counting it down right now. An unlock past zero used to leave the + count negative, which never opened the latch: done() stayed False + for ever and wait() stopped blocking, which spun the localhost job + manager on a full core for the rest of the session + """ + latch = CountDownLatch(1) + latch.unlock() + latch.unlock() + latch.unlock() + + assert latch.done is True + latch.wait() + + def test_countdown_latch_release_opens_it_with_the_count_pending(self): + latch = CountDownLatch(3) + latch.unlock() + assert latch.done is False + + latch.release() + + assert latch.done is True + latch.wait() + # The tasks that never arrive must not close it again + latch.unlock() + assert latch.done is True + + def test_countdown_latch_is_not_closed_by_a_concurrent_drain(self): + """ + The consumer counting down its finished task and clear() opening + the latch, racing the way they do on a job that ends while its last + task is still reporting + """ + for _ in range(200): + latch = CountDownLatch(2) + latch.unlock() + threads = [ + threading.Thread(target=latch.unlock), + threading.Thread(target=latch.release), + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert latch.done is True + def test_log_prefix_builds_executor_job_and_call_identity(self): assert log_prefix('sess-0') == 'ExecutorID sess-0' assert log_prefix('sess-0', 'M000') == 'ExecutorID sess-0 | JobID M000' diff --git a/lithops/tests/test_wait.py b/lithops/tests/test_wait.py index 73ba84e4a..651cb0466 100644 --- a/lithops/tests/test_wait.py +++ b/lithops/tests/test_wait.py @@ -14,7 +14,6 @@ import importlib import signal -import threading from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -331,8 +330,7 @@ def get_data(fs, exec_data, **kwargs): return 1 with patch.object(wait_mod, 'JobMonitor', return_value=monitor) as cls, \ - patch.object(wait_mod, '_get_executor_data', side_effect=get_data), \ - patch.object(wait_mod.time, 'sleep'): + patch.object(wait_mod, '_get_executor_data', side_effect=get_data): wait( [future], show_progressbar=False, @@ -363,8 +361,7 @@ def get_data(fs, exec_data, **kwargs): with patch.object(wait_mod.signal, 'signal', side_effect=fake_signal), \ patch.object(wait_mod.signal, 'alarm') as alarm, \ - patch.object(wait_mod, '_get_executor_data', side_effect=get_data), \ - patch.object(wait_mod.time, 'sleep'): + patch.object(wait_mod, '_get_executor_data', side_effect=get_data): wait( [future], timeout=17, @@ -377,7 +374,7 @@ def get_data(fs, exec_data, **kwargs): alarm.assert_called_with(0) assert 'Timeout of 17 seconds exceeded' in handlers[signal.SIGALRM].args[0] - def test_all_completed_restarts_dead_monitor_and_sleeps_on_empty_poll(self): + def test_all_completed_restarts_dead_monitor_and_waits_on_empty_poll(self): future = FakeFuture() monitor = MagicMock() monitor.type = 'storage' @@ -395,15 +392,10 @@ def get_data(fs, exec_data, **kwargs): future.done = True return 3 - sleeps = [] - test_thread = threading.current_thread() - - def sleep(seconds): - if threading.current_thread() is test_thread: - sleeps.append(seconds) + waits = [] with patch.object(wait_mod, '_get_executor_data', side_effect=get_data), \ - patch.object(wait_mod.time, 'sleep', side_effect=sleep): + patch.object(wait_mod.time, 'sleep', waits.append): wait( [future], return_when=ALL_COMPLETED, @@ -413,7 +405,9 @@ def sleep(seconds): ) monitor.start.assert_called_once_with(fs=[future]) - assert sleeps == [0.1, 0] + # An empty poll sleeps the poll interval; a poll that fetched + # something goes straight round again without waiting at all + assert waits == [0.1] def test_wait_tracks_nested_futures_until_they_complete(self): child = FakeFuture(call_id='00001') @@ -433,13 +427,12 @@ def parent_status(**kwargs): internal = MagicMock() internal.backend = 'localhost' - with patch.object(wait_mod.time, 'sleep'): - done, not_done = wait( - [parent], - show_progressbar=False, - job_monitor=monitor, - internal_storage=internal, - ) + done, not_done = wait( + [parent], + show_progressbar=False, + job_monitor=monitor, + internal_storage=internal, + ) assert parent.success is True assert child.success is True diff --git a/lithops/utils.py b/lithops/utils.py index 77221de4b..bbdc2ea5b 100644 --- a/lithops/utils.py +++ b/lithops/utils.py @@ -1059,20 +1059,40 @@ def __init__(self, count): self.count = count self.event = threading.Event() self.lock = threading.Lock() + if count <= 0: + self.event.set() def unlock(self): + """ + Counts one down, and opens the latch once nothing is left. + + Clamped at zero: a caller draining a latch cannot tell whether the + last task is counting it down right now, and an unlock past zero + used to leave the count negative, which never opened the latch and + never let done() be true again + """ with self.lock: + if self.count == 0: + return self.count -= 1 if self.count == 0: self.event.set() + def release(self): + """ + Opens the latch at once, whatever is left of the count, for the + tasks that are never going to arrive because their job was stopped + """ + with self.lock: + self.count = 0 + self.event.set() + def wait(self): - if self.count > 0: - self.event.wait() + self.event.wait() @property def done(self): - return self.count == 0 + return self.event.is_set() CURRENT_PY_VERSION = version_str(sys.version_info) diff --git a/lithops/wait.py b/lithops/wait.py index c62712600..d5d22a9da 100644 --- a/lithops/wait.py +++ b/lithops/wait.py @@ -16,8 +16,8 @@ import signal import logging -import math import time +import math import concurrent.futures as cf from functools import partial from types import SimpleNamespace @@ -94,17 +94,30 @@ def _log_wait_start(prefix: str, return_when: Any, pending: int) -> None: ) -def _set_wait_alarm(timeout: int) -> None: +def _set_wait_alarm(timeout: float) -> None: """ - Arms a SIGALRM that aborts the wait once the timeout is exceeded + Arms a SIGALRM that aborts the wait once the timeout is exceeded. + + signal.alarm() takes whole seconds and rounds nothing, so a fractional + timeout used to raise TypeError before the wait even started, which is + what a caller asking for 0.5 s got. Rounded up rather than truncated: + int(0.5) is 0, and alarm(0) cancels the alarm instead of setting one, so + the wait would have run for ever with no timeout at all. + + A timeout that has already passed raises here, since there is no shorter + alarm than one second to arm """ - logger.debug(f'Setting waiting timeout to {timeout} seconds') error_msg = ( f'Timeout of {timeout} seconds exceeded waiting for ' 'function activations to finish' ) + if timeout <= 0: + raise TimeoutError(error_msg) + + seconds = math.ceil(timeout) + logger.debug(f'Setting waiting timeout to {timeout} seconds') signal.signal(signal.SIGALRM, partial(timeout_handler, error_msg)) - signal.alarm(timeout) + signal.alarm(seconds) def _create_progressbar(total: int, initial: int): @@ -168,7 +181,9 @@ def _poll_until_done( if _get_executor_data(fs, executor_data, **poll_kwargs): new_data = True - time.sleep(0 if new_data else sleep_sec) + if new_data: + continue + time.sleep(sleep_sec) def wait( From 4fdb9eacc5cb67f432473ba92e918470787fb623 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 20:31:48 +0200 Subject: [PATCH 3/9] Fix manager --- CHANGELOG.md | 16 +- lithops/multiprocessing/__init__.py | 2 +- lithops/multiprocessing/managers.py | 602 +++++++++++++++++++------- lithops/multiprocessing/util.py | 33 +- lithops/tests/test_multiprocessing.py | 597 ++++++++++++++++++++++++- 5 files changed, 1082 insertions(+), 168 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d796bab5..8f958a9ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,8 @@ - [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. - [AWS Batch] Added the `instance_types` config option for EC2/SPOT compute environments. -- [Tests] Added a unit test suite for all non-backend modules (18 files, 1266 tests). +- [Tests] Added a unit test suite for all non-backend modules. +- [Multiprocessing] Added `timeout` to `acquire()`, and `_getvalue()`, `_callmethod()` and `copy_proxy()` to the manager proxies. ### Changed @@ -24,12 +25,16 @@ - [Storage] `cloud_open()` now raises `ValueError` on an unsupported mode. - [Joblib] `lithops_args` is now applied to the pool that runs the batches. - [Standalone] `docker login` now reads the password from stdin and quotes its arguments. +- [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. ### Fixed - [Core] Fixed `wait()` on futures another executor invoked, which crashed instead of waiting for them. - [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 `wait()` with a fractional timeout raising `TypeError` from `signal.alarm()` instead of waiting. - [Core] Fixed a hand-built `FuturesList` raising `AttributeError` instead of creating its executor. - [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. @@ -38,6 +43,15 @@ - [Monitoring] Fixed failed RabbitMQ publishes being dropped with nothing in the log. - [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 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 `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. +- [Localhost] Fixed a job cleared mid-task leaving a latch closed, spinning the v2 job manager on a full core. - [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. diff --git a/lithops/multiprocessing/__init__.py b/lithops/multiprocessing/__init__.py index 0d75b5b85..5fcda1c41 100644 --- a/lithops/multiprocessing/__init__.py +++ b/lithops/multiprocessing/__init__.py @@ -25,7 +25,7 @@ ) from .context import CloudContext as DefaultContext from .connection import Pipe -from .managers import SyncManager as Manager +from .managers import Manager, SyncManager # noqa: F401 from .pool import Pool from .process import CloudProcess as Process from .queues import Queue, SimpleQueue, JoinableQueue diff --git a/lithops/multiprocessing/managers.py b/lithops/multiprocessing/managers.py index 0aef7927e..50e38892e 100644 --- a/lithops/multiprocessing/managers.py +++ b/lithops/multiprocessing/managers.py @@ -16,6 +16,7 @@ import redis import inspect +import types import cloudpickle import logging @@ -83,6 +84,15 @@ def __init__(self, address=None, authkey=None, serializer='pickle', ctx=None): self._managing = False self._mrefs = [] + @property + def address(self): + """ + Where the shared objects live. There is no manager process here, so + this is the Redis the proxies talk to rather than a socket + """ + config = (util.LITHOPS_CONFIG or {}).get('redis') or {} + return (config.get('host'), config.get('port')) + def get_server(self): pass @@ -122,19 +132,28 @@ def shutdown(self): self._managing = False @classmethod - def register(cls, typeid, proxytype=None, callable=None, exposed=None, + def register(cls, typeid, callable=None, proxytype=None, exposed=None, method_to_typeid=None, create_method=True, can_manage=True): """ - Register a typeid with the manager type + Register a typeid with the manager type. + + The standard library's signature is + ``register(typeid, callable=None, proxytype=None, ...)``, and its + documented idiom passes the class as ``callable``. The two were the + other way round here and ``callable`` was accepted and then ignored, + so the documented form built a proxy of None and blew up on first + use. Both names now mean the same thing -- the class to stand in + for -- and ``proxytype`` wins when both are given """ + klass = proxytype if proxytype is not None else callable def temp(self, *args, **kwargs): logger.debug('requesting creation of a shared %r object', typeid) if typeid in _builtin_types: - proxy = proxytype(*args, **kwargs) + proxy = klass(*args, **kwargs) else: - proxy = GenericProxy(typeid, proxytype, *args, **kwargs) + proxy = GenericProxy(typeid, klass, *args, **kwargs) if self._managing and can_manage and hasattr(proxy, '_ref'): proxy._ref.managed = True @@ -154,23 +173,91 @@ class BaseProxy: A base for proxies of shared objects """ + #: A class attribute, not an instance one. Held on the instance it went + #: into __dict__, and a module cannot be pickled: every proxy needed + #: cloudpickle to travel, where the standard library's pickle plainly + _pickler = cloudpickle + def __init__(self, typeid, serializer=None): self._typeid = typeid # object id self._oid = '{}-{}'.format(typeid, util.get_uuid()) - self._pickler = cloudpickle self._client = util.get_redis_client() self._ref = util.RemoteReference(self._oid, client=self._client) + def _getvalue(self): + """ + A copy of the referent, which is what the standard library's + BaseProxy._getvalue() hands back + """ + referent = getattr(self, '_referent', None) + if referent is None: + raise NotImplementedError( + '{} has no referent to copy'.format(type(self).__name__) + ) + return referent() + + def _callmethod(self, methodname, args=(), kwds=None): + """ + Calls a method of the referent by name. + + The standard library's proxies reach the manager process through + this; here the proxy already implements the methods, so it forwards + to itself. Provided because code written against the standard + library calls it directly + """ + method = getattr(self, methodname, None) + if method is None: + raise AttributeError( + '{!r} object has no method {!r}'.format( + type(self).__name__, methodname) + ) + return method(*args, **(kwds or {})) + + def __deepcopy__(self, memo): + """ + A plain copy of the referent. Without it, copy.deepcopy() walks the + proxy's own attributes and duplicates the Redis client + """ + import copy as _copy + + selfcopy = _copy.deepcopy(self._getvalue(), memo) + memo[id(self)] = selfcopy + return selfcopy + + def _expiry(self): + return mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME) + + def _field(self, k): + """ + The hash field a key is stored under. + + Keys used to be handed to redis-py as they were, which accepts only + bytes, str, int and float: a tuple key raised DataError, and an int + or float key came back out of keys() as a str, so d[1] = x then + d.items() gave ('1', x). Pickling the key keeps whatever the caller + put in. Lives here rather than on DictProxy because NamespaceProxy + borrows those methods unbound + """ + return self._pickler.dumps(k) + def __repr__(self): return '<{} object, typeid={}, key={}>'.format(type(self).__name__, self._typeid, self._oid) def __str__(self): """ - Return representation of the referent (or a fall-back if that fails) + The referent's repr, which is what the standard library's proxies + print. Falling back to the proxy's own repr made print(shared_list) + show instead of the list """ - return repr(self) + referent = getattr(self, '_referent', None) + if referent is None: + return repr(self) + try: + return repr(referent()) + except Exception: + return repr(self) # @@ -215,7 +302,6 @@ def __getstate__(self): return { '_typeid': self._typeid, '_oid': self._oid, - '_pickler': self._pickler, '_client': self._client, '_ref': self._ref, '_klass': self._klass, @@ -225,7 +311,6 @@ def __getstate__(self): def __setstate__(self, state): self._typeid = state['_typeid'] self._oid = state['_oid'] - self._pickler = state['_pickler'] self._client = state['_client'] self._ref = state['_ref'] self._klass = state['_klass'] @@ -239,7 +324,25 @@ def __init__(self, proxy, attr_name, shared_object): self._shared_object = shared_object self._proxy = proxy + #: How long a method call may hold the object, and how long another + #: one waits for it. Bounded so a worker that dies mid-call cannot lock + #: the object up for the rest of the job + LOCK_TIMEOUT = 60 + def __call__(self, *args, **kwargs): + # Read state, run the method, write back what changed -- with + # nothing to stop two workers doing that at once, both read the same + # state and the second overwrote the first: counter.increment() + # called twice concurrently left the counter at one. The standard + # library runs every call in the manager's own process, one at a + # time, and this is what stands in for that + client = self._proxy._client + with client.lock(self._proxy._oid + '-call', + timeout=self.LOCK_TIMEOUT, + blocking_timeout=self.LOCK_TIMEOUT): + return self._call(*args, **kwargs) + + def _call(self, *args, **kwargs): attrs = self._proxy._client.hgetall(self._proxy._oid) hashes = {} @@ -265,7 +368,10 @@ def __call__(self, *args, **kwargs): for attr_name in shared: attr = getattr(self._shared_object, attr_name) attr_bin = self._proxy._pickler.dumps(attr) - if hash(attr_bin) != hashes[attr_name]: + # A method that sets an attribute for the first time leaves a + # name the pre-call HGETALL never saw, and looking it up used to + # raise KeyError from inside the call + if hash(attr_bin) != hashes.get(attr_name): pipeline.hset(self._proxy._oid, attr_name, attr_bin) pipeline.expire(self._proxy._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) pipeline.execute() @@ -315,6 +421,38 @@ def __init__(self, iterable=None): if iterable is not None: self.extend(iterable) + def _mutate(self, change): + """ + Reads the whole list, hands it to ``change`` and writes back what it + returns, as one atomic step. + + Redis has no slice assignment, sort or insert, so these have to be + done here. WATCH is what makes that safe: another client writing to + the key between the read and the write aborts the transaction and it + is retried, instead of that write being silently dropped. The old + code did DELETE followed by RPUSH with no guard at all, so a + concurrent append was lost and readers saw an empty list in between + + ``change`` returns ``(new_items, answer)`` and ``answer`` is handed + back to the caller + """ + answer = {} + + def apply(pipe): + raw = pipe.lrange(self._oid, 0, -1) + items = [self._pickler.loads(v) for v in raw] + new_items, answer['value'] = change(items) + pipe.multi() + pipe.delete(self._oid) + if new_items: + pipe.rpush( + self._oid, *[self._pickler.dumps(v) for v in new_items] + ) + pipe.expire(self._oid, self._expiry()) + + self._client.transaction(apply, self._oid) + return answer['value'] + def __setitem__(self, i, obj): if isinstance(i, int) or hasattr(i, '__index__'): idx = i.__index__() @@ -322,38 +460,23 @@ def __setitem__(self, i, obj): try: pipeline = self._client.pipeline() pipeline.lset(self._oid, idx, serialized) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + pipeline.expire(self._oid, self._expiry()) pipeline.execute() except redis.exceptions.ResponseError: # raised when index >= len(self) raise IndexError('list assignment index out of range') - elif isinstance(i, slice): # TODO: step - start, end, step = deslice(i) - if start is None: - return - - if end < 0: - end = len(self) + end - - pipeline = self._client.pipeline(transaction=False) - try: - iterable = iter(obj) - for j in range(start, end): - obj = next(iterable) - serialized = self._pickler.dumps(obj) - pipeline.lset(self._oid, j, serialized) - except StopIteration: - pass - except redis.exceptions.ResponseError: - # raised when index >= len(self) - pipeline.execute() - self.extend(iterable) - return - except TypeError: - raise TypeError('can only assign an iterable') - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) - pipeline.execute() + elif isinstance(i, slice): + # Done through the whole list rather than element by element. + # The old code walked range(start, end) over a bound LRANGE + # reports inclusively, so it wrote one element too few, dropped + # `l[:0] = x` and `l[len(l):] = x` entirely, ignored the step, + # and could not grow or shrink the list at all + def change(items): + items[i] = obj + return items, None + + self._mutate(change) else: raise TypeError('list indices must be integers ' 'or slices, not {}'.format(type(i))) @@ -369,13 +492,18 @@ def __getitem__(self, i): return self._pickler.loads(serialized) raise IndexError('list index out of range') - elif isinstance(i, slice): # TODO: step + elif isinstance(i, slice): start, end, step = deslice(i) + if step is not None and step != 1: + # LRANGE cannot step, and returning the contiguous range + # regardless meant l[::2] gave back the whole list and + # l[::-1] gave it back the right way round + return self.tolist()[i] if start is None: return [] pipeline = self._client.pipeline() pipeline.lrange(self._oid, start, end) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + pipeline.expire(self._oid, self._expiry()) serialized, _ = pipeline.execute() unserialized = [self._pickler.loads(obj) for obj in serialized] return unserialized @@ -387,18 +515,28 @@ def __getitem__(self, i): def extend(self, iterable): if isinstance(iterable, type(self)): self._extend_same_type(iterable, 1) - else: - if iterable != []: - values = map(self._pickler.dumps, iterable) - pipeline = self._client.pipeline() - pipeline.rpush(self._oid, *values) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) - pipeline.execute() + return + # Drawn off before it is measured. `iterable != []` was true for + # every empty thing that is not a list -- (), '', an empty set, a + # generator, the reversed() in reverse() -- and RPUSH with no values + # is an error from the server, not a no-op + values = [self._pickler.dumps(obj) for obj in iterable] + if not values: + return + pipeline = self._client.pipeline() + pipeline.rpush(self._oid, *values) + pipeline.expire(self._oid, self._expiry()) + pipeline.execute() def _extend_same_type(self, listproxy, repeat=1): self._lua_extend_list(keys=[self._oid, listproxy._oid], args=[repeat], client=self._client) + # The script only RPUSHes. Without this a list built from another + # proxy -- ListProxy(other), a deepcopy, an in-place multiply -- + # got a key that never expired, while one built from a plain list + # got one that did + self._client.expire(self._oid, self._expiry()) def append(self, obj): serialized = self._pickler.dumps(obj) @@ -411,26 +549,39 @@ def pop(self, index=None): if index is None: pipeline = self._client.pipeline() pipeline.rpop(self._oid) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + pipeline.expire(self._oid, self._expiry()) serialized, _ = pipeline.execute() + if serialized is None: + # RPOP on a missing key answers nil, and falling off the end + # here handed the caller None as if it were an element + raise IndexError('pop from empty list') + return self._pickler.loads(serialized) - if serialized is not None: - return self._pickler.loads(serialized) - else: - item = self[index] - sentinel = util.get_uuid() - self[index] = sentinel - self.remove(sentinel) - return item + # Read, remove and return in one step. Doing it as LINDEX, LSET of a + # sentinel and LREM let another writer shift the list in between, so + # the element that came back was not the one that was removed + def change(items): + item = items.pop(index) + return items, item - def __deepcopy__(self, memo): - selfcopy = type(self)() + return self._mutate(change) - # We should test the DUMP/RESTORE strategy - # although it has serialization costs - selfcopy._extend_same_type(self) + def _referent(self): + return self.tolist() - memo[id(self)] = selfcopy + def _new_empty(self): + return type(self)() + + def copy_proxy(self): + """ + A second shared list holding the same elements, copied server side. + + deepcopy() used to do this, but the standard library's proxies + deepcopy to a plain value, and code written against it expects a + list back rather than another handle on shared state + """ + selfcopy = self._new_empty() + selfcopy._extend_same_type(self) return selfcopy def __add__(self, x): @@ -448,7 +599,7 @@ def __iadd__(self, x): def __mul__(self, n): if not isinstance(n, int): - raise TypeError("TypeError: can't multiply sequence" + raise TypeError("can't multiply sequence" " by non-int of type {}".format(type(n))) if n < 1: # return type(self)() @@ -464,60 +615,88 @@ def __rmul__(self, n): def __imul__(self, n): if not isinstance(n, int): - raise TypeError("TypeError: can't multiply sequence" + raise TypeError("can't multiply sequence" " by non-int of type {}".format(type(n))) if n > 1: self._extend_same_type(self, repeat=n - 1) + elif n <= 0: + # list *= 0 empties the list; the guard used to leave it alone + self._client.delete(self._oid) return self def __len__(self): - return self._client.llen(self._oid) + pipeline = self._client.pipeline() + pipeline.llen(self._oid) + pipeline.expire(self._oid, self._expiry()) + length, _ = pipeline.execute() + return length + + def __contains__(self, obj): + # One round trip. Without it `in` falls back to the old __getitem__ + # sequence protocol, which is a LINDEX per element + return obj in self.tolist() def remove(self, obj): - serialized = self._pickler.dumps(obj) - pipeline = self._client.pipeline() - pipeline.lrem(self._oid, 1, serialized) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) - pipeline.execute() - return self + """ + Removes the first element equal to ``obj``, like list.remove. + + LREM matches the stored pickle byte for byte, which is not what == + means: remove(1) left a 1.0 in place, and so did removing a dict + whose keys were built in a different order. A value that is not + there raised nothing at all + """ + def change(items): + items.remove(obj) + return items, None + + self._mutate(change) def __delitem__(self, i): - sentinel = util.get_uuid() - self[i] = sentinel - self.remove(sentinel) + def change(items): + del items[i] + return items, None + + self._mutate(change) def tolist(self): - serialized = self._client.lrange(self._oid, 0, -1) - unserialized = [self._pickler.loads(obj) for obj in serialized] - return unserialized + pipeline = self._client.pipeline() + pipeline.lrange(self._oid, 0, -1) + pipeline.expire(self._oid, self._expiry()) + serialized, _ = pipeline.execute() + return [self._pickler.loads(obj) for obj in serialized] # The following methods can't be (properly) implemented on Redis # To still provide the functionality, the list is fetched # entirely, operated in-memory and then put back to Redis def reverse(self): - rev = reversed(self[:]) - self._client.delete(self._oid) - self.extend(rev) - return self + self._mutate(lambda items: (items[::-1], None)) def sort(self, key=None, reverse=False): - sortd = sorted(self[:], key=key, reverse=reverse) - self._client.delete(self._oid) - self.extend(sortd) - return self + self._mutate( + lambda items: (sorted(items, key=key, reverse=reverse), None) + ) - def index(self, obj, start=0, end=-1): - return self[:].index(obj, start, end) + def index(self, obj, start=0, end=None): + """ + The standard library looks to the end of the list by default. This + took end=-1, a real stop that leaves the last element out, so + looking for the last element raised ValueError + """ + items = self.tolist() + if end is None: + return items.index(obj, start) + return items.index(obj, start, end) def count(self, obj): - return self[:].count(obj) + return self.tolist().count(obj) def insert(self, index, obj): - new_list = self[:] - new_list.insert(index, obj) - self._client.delete(self._oid) - self.extend(new_list) + def change(items): + items.insert(index, obj) + return items, None + + self._mutate(change) class DictProxy(BaseProxy): @@ -526,17 +705,20 @@ def __init__(self, *args, **kwargs): super().__init__('dict') self.update(*args, **kwargs) + def _referent(self): + return self.todict() + def __setitem__(self, k, v): serialized = self._pickler.dumps(v) pipeline = self._client.pipeline() - pipeline.hset(self._oid, k, serialized) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + pipeline.hset(self._oid, self._field(k), serialized) + pipeline.expire(self._oid, self._expiry()) pipeline.execute() def __getitem__(self, k): pipeline = self._client.pipeline() - pipeline.hget(self._oid, k) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + pipeline.hget(self._oid, self._field(k)) + pipeline.expire(self._oid, self._expiry()) serialized, _ = pipeline.execute() if serialized is None: raise KeyError(k) @@ -545,18 +727,26 @@ def __getitem__(self, k): def __delitem__(self, k): pipeline = self._client.pipeline() - pipeline.hdel(self._oid, k) - pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + pipeline.hdel(self._oid, self._field(k)) + pipeline.expire(self._oid, self._expiry()) res, _ = pipeline.execute() if res == 0: raise KeyError(k) def __contains__(self, k): - return self._client.hexists(self._oid, k) + pipeline = self._client.pipeline() + pipeline.hexists(self._oid, self._field(k)) + pipeline.expire(self._oid, self._expiry()) + exists, _ = pipeline.execute() + return bool(exists) def __len__(self): - return self._client.hlen(self._oid) + pipeline = self._client.pipeline() + pipeline.hlen(self._oid) + pipeline.expire(self._oid, self._expiry()) + length, _ = pipeline.execute() + return length def __iter__(self): return iter(self.keys()) @@ -569,82 +759,131 @@ def get(self, k, default=None): else: return v - def pop(self, k, default=None): - try: - v = self.__getitem__(k) - except KeyError: - return default - else: - self.__delitem__(k) - return v + def pop(self, k, *args): + """ + dict.pop(k) raises KeyError when the key is not there; only + pop(k, default) answers with the default. The default used to be + baked into the signature, so a missing key quietly gave back None + """ + if len(args) > 1: + raise TypeError( + 'pop expected at most 2 arguments, got {}'.format(1 + len(args)) + ) + field = self._field(k) + pipeline = self._client.pipeline() + pipeline.hget(self._oid, field) + pipeline.hdel(self._oid, field) + pipeline.expire(self._oid, self._expiry()) + serialized, _removed, _ = pipeline.execute() + if serialized is None: + if args: + return args[0] + raise KeyError(k) + return self._pickler.loads(serialized) def popitem(self): - try: - key = self.keys()[0] - item = (key, self.__getitem__(key)) - self.__delitem__(key) - return item - except IndexError: - raise KeyError('popitem(): dictionary is empty') + """ + Taken in one step. Reading a key, then fetching it, then deleting it + let another caller take the same pair, or delete it in between and + turn this into a KeyError naming that key rather than a + 'dictionary is empty' + """ + answer = {} + + def apply(pipe): + fields = pipe.hkeys(self._oid) + if not fields: + raise KeyError('popitem(): dictionary is empty') + field = fields[0] + serialized = pipe.hget(self._oid, field) + if serialized is None: + raise redis.WatchError() + answer['value'] = ( + self._pickler.loads(field), self._pickler.loads(serialized) + ) + pipe.multi() + pipe.hdel(self._oid, field) + pipe.expire(self._oid, self._expiry()) + + self._client.transaction(apply, self._oid) + return answer['value'] def setdefault(self, k, default=None): serialized = self._pickler.dumps(default) - res = self._client.hsetnx(self._oid, k, serialized) + pipeline = self._client.pipeline() + pipeline.hsetnx(self._oid, self._field(k), serialized) + # Every other writer refreshes the expiry. A dict only ever written + # through setdefault used to get a key that outlived the job + pipeline.expire(self._oid, self._expiry()) + res, _ = pipeline.execute() if res == 1: return default - else: - return self.__getitem__(k) + return self.__getitem__(k) def update(self, *args, **kwargs): - items = [] + items = {} if args != (): if len(args) > 1: raise TypeError('update expected at most' ' 1 arguments, got {}'.format(len(args))) try: for k in args[0].keys(): - items.extend((k, self._pickler.dumps(args[0][k]))) - except Exception: + items[self._field(k)] = self._pickler.dumps(args[0][k]) + except AttributeError: try: - items = [] # just in case + items = {} # just in case for k, v in args[0]: - items.extend((k, self._pickler.dumps(v))) + items[self._field(k)] = self._pickler.dumps(v) except Exception: raise TypeError(type(args[0])) for k in kwargs.keys(): - items.extend((k, self._pickler.dumps(kwargs[k]))) + items[self._field(k)] = self._pickler.dumps(kwargs[k]) - if len(items) > 0: - self._client.execute_command('HMSET', self._oid, *items) - self._client.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + if items: + # One pipelined HSET rather than the deprecated HMSET and a + # separate round trip for the expiry + pipeline = self._client.pipeline() + pipeline.hset(self._oid, mapping=items) + pipeline.expire(self._oid, self._expiry()) + pipeline.execute() def keys(self): - return [k.decode() for k in self._client.hkeys(self._oid)] + pipeline = self._client.pipeline() + pipeline.hkeys(self._oid) + pipeline.expire(self._oid, self._expiry()) + fields, _ = pipeline.execute() + return [self._pickler.loads(k) for k in fields] def values(self): - return [self._pickler.loads(v) for v in self._client.hvals(self._oid)] + pipeline = self._client.pipeline() + pipeline.hvals(self._oid) + pipeline.expire(self._oid, self._expiry()) + values, _ = pipeline.execute() + return [self._pickler.loads(v) for v in values] def items(self): - raw_dict = self._client.hgetall(self._oid) - items = [] - for k, v in raw_dict.items(): - items.append((k.decode(), self._pickler.loads(v))) - return items + return list(self.todict().items()) def clear(self): self._client.delete(self._oid) def copy(self): - # TODO: use lua script - return type(self)(self.items()) + """ + A plain dict, like the standard library. Handing back another proxy + made what looks like a local copy allocate a second Redis key + """ + return self.todict() def todict(self): - raw_dict = self._client.hgetall(self._oid) - py_dict = {} - for k, v in raw_dict.items(): - py_dict[k.decode()] = self._pickler.loads(v) - return py_dict + pipeline = self._client.pipeline() + pipeline.hgetall(self._oid) + pipeline.expire(self._oid, self._expiry()) + raw_dict, _ = pipeline.execute() + return { + self._pickler.loads(k): self._pickler.loads(v) + for k, v in raw_dict.items() + } class NamespaceProxy(BaseProxy): @@ -673,6 +912,9 @@ def __delattr__(self, k): except KeyError: raise AttributeError(k) + def _referent(self): + return types.SimpleNamespace(**DictProxy.todict(self)) + def _todict(self): return DictProxy.todict(self) @@ -680,11 +922,24 @@ def _todict(self): class ValueProxy(BaseProxy): def __init__(self, typecode='Any', value=None, lock=True): super().__init__('Value({})'.format(typecode)) - if value is not None: - self.set(value) + self._typecode = typecode + # Written whatever it is. Skipping a None left the key missing, and + # get() then handed None to loads() and raised TypeError instead of + # answering None + self.set(value) + + def _referent(self): + return self.get() def get(self): - serialized = self._client.get(self._oid) + pipeline = self._client.pipeline() + pipeline.get(self._oid) + # Read without refreshing, a value that is polled and never written + # disappears once REDIS_EXPIRY_TIME is up, mid-job + pipeline.expire(self._oid, mp_config.get_parameter(mp_config.REDIS_EXPIRY_TIME)) + serialized, _ = pipeline.execute() + if serialized is None: + return None return self._pickler.loads(serialized) def set(self, value): @@ -695,9 +950,23 @@ def set(self, value): class ArrayProxy(ListProxy): - def __init__(self, typecode, sequence, lock=True): + def __init__(self, typecode='Any', sequence=None, lock=True): + """ + Takes a size as well as a sequence, like multiprocessing.Array. + + Array('i', 10) allocates ten zeroed slots there; here it reached + extend(10) and raised TypeError: 'int' object is not iterable + """ + self._typecode = typecode + if isinstance(sequence, int): + sequence = [0] * sequence super().__init__(sequence) + def _new_empty(self): + # ListProxy.__deepcopy__ builds an empty one of the same type, and + # this one needs its typecode + return type(self)(self._typecode) + # # Definition of SyncManager @@ -715,19 +984,32 @@ class SyncManager(BaseManager): """ -SyncManager.register('list', ListProxy) -SyncManager.register('dict', DictProxy) -SyncManager.register('Namespace', NamespaceProxy) -SyncManager.register('Lock', synchronize.Lock) -SyncManager.register('RLock', synchronize.RLock) -SyncManager.register('Semaphore', synchronize.Semaphore) -SyncManager.register('BoundedSemaphore', synchronize.BoundedSemaphore) -SyncManager.register('Condition', synchronize.Condition) -SyncManager.register('Event', synchronize.Event) -SyncManager.register('Barrier', synchronize.Barrier) -SyncManager.register('Queue', queues.Queue) -SyncManager.register('SimpleQueue', queues.SimpleQueue) -SyncManager.register('JoinableQueue', queues.JoinableQueue) -SyncManager.register('Value', ValueProxy) -SyncManager.register('Array', ArrayProxy) -SyncManager.register('Pool', pool.Pool, can_manage=False) +def Manager(): + """ + A started SyncManager, which is what multiprocessing.Manager() returns. + + The class itself used to be exported under this name, so every manager + came back unstarted: shutdown() was a no-op, _number_of_objects() always + said zero, and nothing handed out was ever collected + """ + manager = SyncManager() + manager.start() + return manager + + +SyncManager.register('list', proxytype=ListProxy) +SyncManager.register('dict', proxytype=DictProxy) +SyncManager.register('Namespace', proxytype=NamespaceProxy) +SyncManager.register('Lock', proxytype=synchronize.Lock) +SyncManager.register('RLock', proxytype=synchronize.RLock) +SyncManager.register('Semaphore', proxytype=synchronize.Semaphore) +SyncManager.register('BoundedSemaphore', proxytype=synchronize.BoundedSemaphore) +SyncManager.register('Condition', proxytype=synchronize.Condition) +SyncManager.register('Event', proxytype=synchronize.Event) +SyncManager.register('Barrier', proxytype=synchronize.Barrier) +SyncManager.register('Queue', proxytype=queues.Queue) +SyncManager.register('SimpleQueue', proxytype=queues.SimpleQueue) +SyncManager.register('JoinableQueue', proxytype=queues.JoinableQueue) +SyncManager.register('Value', proxytype=ValueProxy) +SyncManager.register('Array', proxytype=ArrayProxy) +SyncManager.register('Pool', proxytype=pool.Pool, can_manage=False) diff --git a/lithops/multiprocessing/util.py b/lithops/multiprocessing/util.py index 1b7347ad8..aaf8d850c 100644 --- a/lithops/multiprocessing/util.py +++ b/lithops/multiprocessing/util.py @@ -133,6 +133,10 @@ def __init__(self, referenced, managed=False, client=None): self._callback = None self.managed = managed + # The object that just built this holds a reference. Without it the + # count started a whole owner short, so the first owner to go away + # took the shared object with it + self.incref() @property def managed(self): @@ -140,6 +144,9 @@ def managed(self): @managed.setter def managed(self, value): + self._set_managed(value) + + def _set_managed(self, value): managed = value if self._callback is not None: @@ -153,6 +160,16 @@ def managed(self, value): self._client, self._rck, self._referenced) def __getstate__(self): + """ + Takes a reference on behalf of the copy that comes back out. + + Nothing owns a proxy while it is bytes on its way to a worker. The + count used to be raised only once it was unpickled, so a proxy + pickled from a temporary -- passed straight into a call, or copied -- + was collected while in flight and deleted the shared object before + the copy ever existed + """ + self.incref() return (self._rck, self._referenced, self._client, self.managed) @@ -160,8 +177,9 @@ def __setstate__(self, state): (self._rck, self._referenced, self._client) = state[:-1] self._callback = None - self.managed = state[-1] - self.incref() + # Adopts the reference __getstate__ took. Raising it again here + # would leave one that nothing ever gives back + self._set_managed(state[-1]) def incref(self): if not self.managed: @@ -181,7 +199,7 @@ def decref(self): def refcount(self): count = self._client.get(self._rck) - return 1 if count is None else int(count) + 1 + return 0 if count is None else int(count) def collect(self): if len(self._referenced) > 0: @@ -190,8 +208,15 @@ def collect(self): @staticmethod def _finalize(client, rck, referenced): + """ + Deletes the shared object once the last owner is gone. + + The creator now takes a reference of its own, so the count reaching + zero is what says nobody is left; it used to have to go negative, + which is one owner too many + """ count = int(client.decr(rck, 1)) - if count < 0 and len(referenced) > 0: + if count <= 0 and len(referenced) > 0: client.delete(*referenced) diff --git a/lithops/tests/test_multiprocessing.py b/lithops/tests/test_multiprocessing.py index 8ba055378..84a97c1d2 100644 --- a/lithops/tests/test_multiprocessing.py +++ b/lithops/tests/test_multiprocessing.py @@ -22,6 +22,7 @@ """ import ctypes +import gc import pickle import queue import sys @@ -66,6 +67,39 @@ def redis(): return server +@pytest.fixture +def real_redis(): + """ + A real server, for the manager proxies. + + They lean on hashes, WATCH/MULTI and pickled hash fields, none of which + the in-memory double has, and writing the double and the code it is + meant to check in the same pass proves nothing. Skipped where no server + is reachable; every key the test makes is dropped afterwards + """ + # 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') + + before = set(client.keys('*')) + saved_client, saved_config = mp_util.REDIS_CLIENT, mp_util.LITHOPS_CONFIG + mp_util.REDIS_CLIENT = client + mp_util.LITHOPS_CONFIG = {'redis': {'host': 'localhost'}} + try: + yield client + finally: + mp_util.REDIS_CLIENT, mp_util.LITHOPS_CONFIG = saved_client, saved_config + made = set(client.keys('*')) - before + if made: + client.delete(*made) + + class FakeFuture: def __init__(self, value=None, error=False): self.executor_id = 'sess-0' @@ -213,10 +247,13 @@ def test_managed_reference_does_not_count(self, redis): assert ref.decref() is None def test_unmanaged_reference_counts_up_and_down(self, redis): + # Building one takes a reference of its own, so the count starts at + # one rather than zero: whatever built it is an owner ref = mp_util.RemoteReference('key-1', client=redis) - assert ref.incref() == 1 + assert ref.refcount() == 1 assert ref.incref() == 2 - assert ref.decref() == 1 + assert ref.incref() == 3 + assert ref.decref() == 2 def test_the_counter_key_is_collected_with_the_referenced_ones(self, redis): ref = mp_util.RemoteReference(['key-1', 'key-2'], client=redis) @@ -1619,3 +1656,559 @@ def test_pool_get_with_a_fractional_timeout_raises_timeout_error(self): patch.object(lw.signal, 'signal'): lw._set_wait_alarm(0.2) alarm.assert_called_once_with(1) + + +class TestListProxy: + """ + lithops.multiprocessing.Manager().list(), against a real Redis. Every + case here is one the proxy used to get wrong + """ + + @staticmethod + def _list(*args): + from lithops.multiprocessing import managers + return managers.ListProxy(*args) + + def test_it_matches_a_plain_list(self, real_redis): + assert self._list([1, 2, 3]).tolist() == [1, 2, 3] + assert len(self._list([1, 2, 3])) == 3 + assert self._list([]).tolist() == [] + + # -- slice reads + + def test_a_step_is_honoured(self, real_redis): + plain = [0, 1, 2, 3, 4, 5] + proxy = self._list(plain) + # LRANGE cannot step, so these used to come back as the whole list, + # and l[::-1] came back the right way round + assert proxy[::2] == plain[::2] + assert proxy[::-1] == plain[::-1] + assert proxy[1:5:2] == plain[1:5:2] + + def test_slice_reads_match_a_plain_list(self, real_redis): + plain = [0, 1, 2, 3, 4] + proxy = self._list(plain) + for s in [slice(0, 2), slice(None, None), slice(1, -1), slice(2, 2), + slice(None, 0), slice(3, None), slice(1, 0), slice(0, -9)]: + assert proxy[s] == plain[s], s + + # -- slice assignment + + def test_slice_assignment_writes_every_element(self, real_redis): + """ + deslice() hands back the inclusive end LRANGE wants, and the caller + walked it with range(), which is exclusive: the last element of + every slice assignment was left as it was + """ + proxy, plain = self._list([1, 2, 3]), [1, 2, 3] + proxy[0:2] = [9, 8] + plain[0:2] = [9, 8] + assert proxy.tolist() == plain == [9, 8, 3] + + def test_assigning_the_whole_list(self, real_redis): + proxy = self._list([1, 2, 3]) + proxy[:] = [9, 8, 7] + assert proxy.tolist() == [9, 8, 7] + + def test_slice_assignment_can_prepend(self, real_redis): + """l[:0] = x hit the start-is-None guard and did nothing at all""" + proxy, plain = self._list([1, 2, 3]), [1, 2, 3] + proxy[:0] = [0] + plain[:0] = [0] + assert proxy.tolist() == plain == [0, 1, 2, 3] + + def test_slice_assignment_can_append(self, real_redis): + """l[len(l):] = x walked an empty range and did nothing""" + proxy, plain = self._list([1, 2, 3]), [1, 2, 3] + proxy[3:] = [4, 5] + plain[3:] = [4, 5] + assert proxy.tolist() == plain == [1, 2, 3, 4, 5] + + def test_slice_assignment_can_grow_and_shrink(self, real_redis): + proxy, plain = self._list([1, 2]), [1, 2] + proxy[0:5] = [9, 8, 7, 6, 5] + plain[0:5] = [9, 8, 7, 6, 5] + assert proxy.tolist() == plain + + proxy, plain = self._list([1, 2, 3, 4]), [1, 2, 3, 4] + proxy[1:3] = [] + plain[1:3] = [] + assert proxy.tolist() == plain == [1, 4] + + def test_extended_slice_assignment(self, real_redis): + proxy, plain = self._list([0, 1, 2, 3]), [0, 1, 2, 3] + proxy[::2] = ['a', 'b'] + plain[::2] = ['a', 'b'] + assert proxy.tolist() == plain + + def test_an_extended_slice_of_the_wrong_length_raises(self, real_redis): + proxy = self._list([0, 1, 2, 3]) + with pytest.raises(ValueError): + proxy[::2] = [1, 2, 3] + + # -- deletion + + def test_deleting_a_slice(self, real_redis): + """ + __delitem__ assigned a uuid sentinel, which the slice branch then + iterated character by character, leaving ['a', '7', 3] + """ + proxy, plain = self._list([1, 2, 3]), [1, 2, 3] + del proxy[0:2] + del plain[0:2] + assert proxy.tolist() == plain == [3] + + def test_deleting_an_index(self, real_redis): + proxy = self._list([1, 2, 3]) + del proxy[1] + assert proxy.tolist() == [1, 3] + + # -- remove / index / count + + def test_remove_matches_by_equality_not_by_pickle(self, real_redis): + """ + LREM compares the stored pickle byte for byte. dumps(1) is not + dumps(1.0), so remove(1) walked past a 1.0 and did nothing + """ + proxy = self._list([1.0, 2]) + proxy.remove(1) + assert proxy.tolist() == [2] + + def test_remove_matches_an_equal_dict_built_in_another_order(self, real_redis): + proxy = self._list([{'a': 1, 'b': 2}]) + proxy.remove({'b': 2, 'a': 1}) + assert proxy.tolist() == [] + + def test_removing_something_that_is_not_there_raises(self, real_redis): + proxy = self._list([1, 2]) + with pytest.raises(ValueError): + proxy.remove(99) + assert proxy.tolist() == [1, 2] + + def test_index_looks_to_the_end_by_default(self, real_redis): + """The default end=-1 left the last element out of every search""" + proxy = self._list([1, 2, 3]) + assert proxy.index(3) == 2 + assert proxy.index(2, 1) == 1 + with pytest.raises(ValueError): + proxy.index(99) + + def test_count(self, real_redis): + assert self._list([1, 1, 2]).count(1) == 2 + + def test_contains(self, real_redis): + proxy = self._list([1, 2, 3]) + assert (2 in proxy) is True + assert (99 in proxy) is False + + # -- pop + + def test_popping_an_empty_list_raises(self, real_redis): + """RPOP answers nil, and that nil used to be handed back as a value""" + with pytest.raises(IndexError): + self._list([]).pop() + + def test_pop_returns_and_removes_the_same_element(self, real_redis): + proxy = self._list([1, 2, 3]) + assert proxy.pop() == 3 + assert proxy.pop(0) == 1 + assert proxy.tolist() == [2] + + def test_popping_a_bad_index_raises(self, real_redis): + with pytest.raises(IndexError): + self._list([1]).pop(5) + + # -- extend / multiply / reverse / sort / insert + + def test_extending_with_an_empty_iterable(self, real_redis): + """ + `iterable != []` was true for every empty thing that is not a list, + and RPUSH with no values is an error from the server + """ + for empty in [(), '', set(), iter([]), (x for x in [])]: + proxy = self._list([1]) + proxy.extend(empty) + assert proxy.tolist() == [1] + + def test_reversing_an_empty_list(self, real_redis): + """reversed([]) is an iterator, so it got past the `!= []` guard""" + proxy = self._list([]) + proxy.reverse() + assert proxy.tolist() == [] + + def test_reverse_and_sort(self, real_redis): + proxy = self._list([3, 1, 2]) + proxy.sort() + assert proxy.tolist() == [1, 2, 3] + proxy.reverse() + assert proxy.tolist() == [3, 2, 1] + proxy.sort(reverse=True) + assert proxy.tolist() == [3, 2, 1] + + def test_insert(self, real_redis): + proxy = self._list([1, 3]) + proxy.insert(1, 2) + assert proxy.tolist() == [1, 2, 3] + + def test_multiplying_in_place_by_zero_empties_the_list(self, real_redis): + """The `n > 1` guard made l *= 0 and l *= -1 no-ops""" + for n in (0, -1): + proxy = self._list([1, 2, 3]) + proxy *= n + assert proxy.tolist() == [] + + def test_multiplying_in_place(self, real_redis): + proxy = self._list([1, 2]) + proxy *= 2 + assert proxy.tolist() == [1, 2, 1, 2] + + def test_a_list_built_from_another_proxy_gets_an_expiry(self, real_redis): + """The Lua extend only RPUSHed, so the key never expired""" + source = self._list([1, 2]) + copy = self._list(source) + assert copy.tolist() == [1, 2] + assert real_redis.ttl(copy._oid) > 0 + + def test_str_shows_the_list(self, real_redis): + assert str(self._list([1, 2])) == '[1, 2]' + + +class TestDictProxy: + + @staticmethod + def _dict(*args, **kwargs): + from lithops.multiprocessing import managers + return managers.DictProxy(*args, **kwargs) + + def test_it_matches_a_plain_dict(self, real_redis): + d = self._dict({'a': 1}, b=2) + assert d.todict() == {'a': 1, 'b': 2} + assert d['a'] == 1 + assert len(d) == 2 + assert ('a' in d) is True + assert sorted(d.keys()) == ['a', 'b'] + assert sorted(d.values()) == [1, 2] + assert sorted(d.items()) == [('a', 1), ('b', 2)] + + def test_update_from_a_sequence_of_pairs(self, real_redis): + assert self._dict([('a', 1), ('b', 2)]).todict() == {'a': 1, 'b': 2} + + def test_keys_keep_their_type(self, real_redis): + """ + Keys went to redis-py as they were, so an int key came back a str: + d[1] = 'x' then d.keys() gave ['1'] + """ + d = self._dict() + d[1] = 'x' + assert d.keys() == [1] + assert d.items() == [(1, 'x')] + assert (1 in d) is True + + def test_any_hashable_key_works(self, real_redis): + """A tuple key raised DataError, and True and None raised too""" + d = self._dict() + for key in [(1, 2), True, None, 3.5, b'raw', frozenset({1})]: + d[key] = 'v' + assert d[key] == 'v' + assert key in d + del d[key] + + def test_a_bool_key_is_not_confused_with_an_int(self, real_redis): + d = self._dict() + d[1] = 'int' + d[True] = 'bool' + assert d[1] == 'int' + assert d[True] == 'bool' + + def test_pop_without_a_default_raises(self, real_redis): + """The default was baked in, so a missing key quietly gave None""" + d = self._dict({'a': 1}) + assert d.pop('a') == 1 + with pytest.raises(KeyError): + d.pop('missing') + assert d.pop('missing', 'fallback') == 'fallback' + + def test_pop_removes_the_key(self, real_redis): + d = self._dict({'a': 1}) + d.pop('a') + assert 'a' not in d + + def test_popitem(self, real_redis): + d = self._dict({'a': 1}) + assert d.popitem() == ('a', 1) + with pytest.raises(KeyError, match='dictionary is empty'): + d.popitem() + + def test_setdefault(self, real_redis): + d = self._dict() + assert d.setdefault('a', 1) == 1 + assert d.setdefault('a', 2) == 1 + # Every other writer refreshes the expiry; this one never did + assert real_redis.ttl(d._oid) > 0 + + def test_copy_is_a_plain_dict(self, real_redis): + """It used to hand back a proxy, allocating a second Redis key""" + copy = self._dict({'a': 1}).copy() + assert copy == {'a': 1} + assert isinstance(copy, dict) + + def test_missing_key_raises(self, real_redis): + d = self._dict() + with pytest.raises(KeyError): + d['nope'] + with pytest.raises(KeyError): + del d['nope'] + assert d.get('nope') is None + assert d.get('nope', 5) == 5 + + def test_reads_refresh_the_expiry(self, real_redis): + """ + A dict written once and then only read used to lose its key when + REDIS_EXPIRY_TIME was up, mid-job + """ + d = self._dict({'a': 1}) + real_redis.expire(d._oid, 5) + d.todict() + assert real_redis.ttl(d._oid) > 10 + + def test_str_shows_the_dict(self, real_redis): + assert str(self._dict({'a': 1})) == "{'a': 1}" + + +class TestNamespaceProxy: + + @staticmethod + def _ns(**kwargs): + from lithops.multiprocessing import managers + return managers.NamespaceProxy(**kwargs) + + def test_attributes_round_trip(self, real_redis): + ns = self._ns(x=1) + assert ns.x == 1 + ns.y = 2 + assert ns.y == 2 + del ns.y + with pytest.raises(AttributeError): + ns.y + + def test_a_missing_attribute_raises(self, real_redis): + with pytest.raises(AttributeError): + self._ns().nope + + +class TestValueProxy: + + @staticmethod + def _value(*args): + from lithops.multiprocessing import managers + return managers.ValueProxy(*args) + + def test_it_holds_a_value(self, real_redis): + v = self._value('i', 7) + assert v.get() == 7 + v.set(9) + assert v.value == 9 + + def test_it_can_hold_none(self, real_redis): + """ + A None was never written, so the key did not exist and get() handed + None to loads() and raised TypeError + """ + assert self._value('i', None).get() is None + assert self._value().get() is None + + def test_reads_refresh_the_expiry(self, real_redis): + v = self._value('i', 1) + real_redis.expire(v._oid, 5) + v.get() + assert real_redis.ttl(v._oid) > 10 + + +class TestArrayProxy: + + @staticmethod + def _array(*args): + from lithops.multiprocessing import managers + return managers.ArrayProxy(*args) + + def test_it_can_be_built_from_a_size(self, real_redis): + """Array('i', 10) reached extend(10) and raised TypeError""" + assert self._array('i', 10).tolist() == [0] * 10 + + def test_it_can_be_built_from_a_sequence(self, real_redis): + assert self._array('i', [1, 2, 3]).tolist() == [1, 2, 3] + + def test_it_can_be_deep_copied(self, real_redis): + """ + A plain list, like the standard library, which deepcopies a proxy to + its referent. copy_proxy() is what makes another shared one + """ + import copy + + arr = self._array('i', [1, 2]) + assert copy.deepcopy(arr) == [1, 2] + assert arr.copy_proxy().tolist() == [1, 2] + + +class TestManagerContract: + + def test_manager_returns_a_started_manager(self, real_redis): + """ + The class was exported under this name, so every manager came back + unstarted: shutdown() was a no-op and nothing was ever collected + """ + import lithops.multiprocessing as mp + + manager = mp.Manager() + try: + assert manager._managing is True + shared = manager.list([1, 2]) + assert manager._number_of_objects() == 1 + assert shared.tolist() == [1, 2] + finally: + manager.shutdown() + assert manager._managing is False + + def test_register_takes_the_class_as_callable(self, real_redis): + """ + The standard library's documented idiom is register(typeid, + callable=Cls). callable was accepted and then ignored, so the proxy + was built from None and blew up on first use + """ + import lithops.multiprocessing as mp + + class Maths: + def __init__(self): + self.total = 0 + + def add(self, n): + self.total += n + return self.total + + class MyManager(mp.SyncManager): + pass + + MyManager.register('Maths', callable=Maths) + with MyManager() as manager: + maths = manager.Maths() + assert maths.add(3) == 3 + assert maths.add(4) == 7 + + def test_a_method_may_create_a_new_attribute(self, real_redis): + """ + The hashes came from the pre-call HGETALL, so a name the method set + for the first time was missing from it and raised KeyError + """ + import lithops.multiprocessing as mp + + class Grower: + def __init__(self): + self.x = 0 + + def add_y(self): + self.y = 5 + return 'ok' + + class MyManager(mp.SyncManager): + pass + + MyManager.register('Grower', callable=Grower) + with MyManager() as manager: + assert manager.Grower().add_y() == 'ok' + + def test_concurrent_method_calls_do_not_lose_an_update(self, real_redis): + """ + Read state, run the method, write back: with nothing serialising it, + two workers both read total=0 and both wrote total=1, and one + increment vanished. Serialising calls is the whole point of a manager + """ + import lithops.multiprocessing as mp + + class Counter: + def __init__(self): + self.total = 0 + + def increment(self): + self.total += 1 + + class MyManager(mp.SyncManager): + pass + + MyManager.register('Counter', callable=Counter) + with MyManager() as manager: + counter = manager.Counter() + start = threading.Barrier(8) + + def bump(): + start.wait() + counter.increment() + + threads = [threading.Thread(target=bump) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + stored = real_redis.hget(counter._oid, 'total') + assert cloudpickle.loads(stored) == 8 + + +class TestRemoteReferenceOwnership: + """ + Proxies are reference counted in Redis, and the shared object is deleted + once the last owner is gone + """ + + def test_a_proxy_pickled_from_a_temporary_keeps_its_data(self, real_redis): + """ + Nothing owns a proxy while it is bytes on the way to a worker. The + count was raised only on unpickling, so a proxy built and pickled in + one expression was collected in flight and deleted the shared object + before the copy existed + """ + from lithops.multiprocessing import managers + + data = pickle.dumps(managers.ListProxy([1, 2])) + gc.collect() + assert pickle.loads(data).tolist() == [1, 2] + + def test_the_creator_holds_a_reference(self, real_redis): + from lithops.multiprocessing import managers + + proxy = managers.ListProxy([1]) + assert proxy._ref.refcount() == 1 + + def test_a_copy_keeps_the_object_alive(self, real_redis): + from lithops.multiprocessing import managers + + proxy = managers.ListProxy([1, 2]) + oid = proxy._oid + copy = pickle.loads(pickle.dumps(proxy)) + del proxy + gc.collect() + # The copy is still holding it + assert copy.tolist() == [1, 2] + assert real_redis.exists(oid) + + def test_the_last_owner_going_away_deletes_the_object(self, real_redis): + from lithops.multiprocessing import managers + + proxy = managers.ListProxy([1, 2]) + oid, rck = proxy._oid, proxy._ref._rck + del proxy + gc.collect() + assert not real_redis.exists(oid) + assert not real_redis.exists(rck) + + def test_a_managed_proxy_is_left_to_the_manager(self, real_redis): + """A manager collects what it handed out, on shutdown""" + import lithops.multiprocessing as mp + + manager = mp.Manager() + shared = manager.list([1, 2]) + oid = shared._oid + del shared + gc.collect() + assert real_redis.exists(oid) + manager.shutdown() + assert not real_redis.exists(oid) From a0afc8971047886cb3ba79add47dcd864f6d7528 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 20:55:54 +0200 Subject: [PATCH 4/9] Update changelog --- CHANGELOG.md | 69 ++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 62 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8f958a9ec..7ef563b88 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,39 +6,68 @@ - [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. +- [Monitoring] Added the `aws_sqs`, `gcp_pubsub` and `azure_queue` config sections, which fall back to the `aws`, `gcp` and `azure_storage` credentials. - [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. -- [Tests] Added a unit test suite for all non-backend modules. - [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()`). +- [Docs] Added architecture diagrams for the localhost, VM, AWS EC2, Azure VMs and IBM VPC backends. +- [Docs] Added the `lithops.concurrent.futures` reference, the localhost v1/v2 guide, and rewrote the monitoring docs. +- [Tests] Added a unit test suite for all non-backend modules. ### Changed - [Monitoring] Reorganised job monitoring as pluggable backends. -- [Monitoring] The RabbitMQ queue is now deleted on cleanup instead of on every `stop()`, so a later `map()` can reuse it. +- [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. - [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. -- [CLI] `lithops clean` now empties the local temp directory instead of removing it. - [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] `imap()` and `imap_unordered()` now default to the configured chunksize. +- [Docs] Removed the `website/` sources; the landing page now lives in the documentation site. ### Fixed - [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 `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 `wait()` with a fractional timeout raising `TypeError` from `signal.alarm()` instead of waiting. +- [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 `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`. +- [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. - [Multiprocessing] Fixed `error_callback` never being called by `apply_async()`, `map_async()` and `starmap_async()`. @@ -51,24 +80,50 @@ - [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 `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 `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. - [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 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 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 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 `CloudFileProxy.listdir()` returning nothing for its default argument. -- [Cleaner] Fixed two cleaners racing for the pid file, and requests being skipped or read while still being written. -- [Cleaner] Fixed the cleaner looping forever on a request it could not read or classify. - [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. -- [Job] Fixed folder markers being counted as objects, returning empty partitions. +- [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. +- [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. +- [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. - [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. From da9d998eeeaa9ddef22d9ce308360018efd0c5b4 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 21:59:14 +0200 Subject: [PATCH 5/9] Update backends --- CHANGELOG.md | 11 + config/config_template.yaml | 7 - docs/source/storage_config/infinispan.md | 38 +-- .../storage/backends/infinispan/infinispan.py | 96 +++++-- .../backends/infinispan_hotrod/__init__.py | 3 - .../backends/infinispan_hotrod/config.py | 29 --- .../infinispan_hotrod/infinispan_hotrod.py | 238 ------------------ lithops/storage/backends/redis/redis.py | 123 +++++---- 8 files changed, 158 insertions(+), 387 deletions(-) delete mode 100644 lithops/storage/backends/infinispan_hotrod/__init__.py delete mode 100644 lithops/storage/backends/infinispan_hotrod/config.py delete mode 100644 lithops/storage/backends/infinispan_hotrod/infinispan_hotrod.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ef563b88..d1f628379 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,17 @@ ### Fixed +- [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. diff --git a/config/config_template.yaml b/config/config_template.yaml index c2d553975..baf3d59f8 100644 --- a/config/config_template.yaml +++ b/config/config_template.yaml @@ -545,13 +545,6 @@ #- storage #cache_type: default # Cache type -#infinispan_hotrod: - #endpoint: - #username: - #password: - #cache_names: - #- storage - # ============================================================================= # RabbitMQ – required when `lithops.monitoring=rabbitmq`, by k8s with diff --git a/docs/source/storage_config/infinispan.md b/docs/source/storage_config/infinispan.md index bdae6465c..50f0490ed 100644 --- a/docs/source/storage_config/infinispan.md +++ b/docs/source/storage_config/infinispan.md @@ -1,7 +1,6 @@ # Infinispan -Lithops with Infinispan as storage backend. Infinispan provides two different endpoints: `infinispan` (REST) and -`infinispan_hotrod` (native binary). +Lithops with Infinispan as storage backend, over the Infinispan REST endpoint. ## Installation @@ -13,7 +12,6 @@ Lithops with Infinispan as storage backend. Infinispan provides two different en Edit your Lithops config file and add the following keys: -### REST endpoint ```yaml lithops: storage: infinispan @@ -30,43 +28,13 @@ Edit your Lithops config file and add the following keys: - ... ``` -#### Summary of configuration keys for Infinispan: +## Summary of configuration keys for Infinispan: |Group|Key|Default|Mandatory|Additional info| |---|---|---|---|---| |infinispan | endpoint | |yes | Endpoint to your Infinispan server | |infinispan | username | |yes | The username | |infinispan | password | |yes | The password | -|infinispan | mech | |no | Authentication mechanism | +|infinispan | mech | DIGEST |no | Authentication mechanism: DIGEST or BASIC | |infinispan | cache_names | | no | List of cache names. Each bucket will be mapped to a different cache with the same name. Defaults to `['storage']` | |infinispan | cache_type | | no | Type of the cache. Defaults to `default` | - - -### Hotrod endpoint: - -To run this endpoint you need to compile and install the Infinispan Python client ([home page](https://github.com/infinispan/python-client)). - -```yaml - lithops: - storage: infinispan_hotrod - data_limit: 8 # More space for data than the 4MB default - - infinispan_hotrod: - username : - password : - endpoint : - cache_names : - - cache_name_1 - - cache_name_2 - - ... -``` - -#### Summary of configuration keys for Infinispan_hotrod: - -|Group|Key|Default|Mandatory|Additional info| -|---|---|---|---|---| -|infinispan_hotrod | endpoint | |yes | Endpoint to your Infinispan server | -|infinispan_hotrod | username | |yes | The username | -|infinispan_hotrod | password | |yes | The password | -|infinispan_hotrod | cache_names | | no | List of cache names. Each bucket will be mapped to a different cache with the same name. Defaults to `['storage']` | - diff --git a/lithops/storage/backends/infinispan/infinispan.py b/lithops/storage/backends/infinispan/infinispan.py index 406f5a5cc..0a6058777 100644 --- a/lithops/storage/backends/infinispan/infinispan.py +++ b/lithops/storage/backends/infinispan/infinispan.py @@ -22,6 +22,7 @@ import json import base64 import io +from concurrent.futures import ThreadPoolExecutor from requests.auth import HTTPBasicAuth from requests.auth import HTTPDigestAuth from lithops.constants import STORAGE_CLI_MSG @@ -29,6 +30,21 @@ logger = logging.getLogger(__name__) +CONN_POOL_SIZE = 32 + + +def _parse_range(bytes_range): + """ + Translates an HTTP byte range into the slice of the value it selects. + Accepts 'L-H', 'L-' (from L to the end) and '-N' (the last N bytes) + """ + start, _, end = bytes_range.partition('-') + + if not start: # '-N': the last N bytes + return slice(-int(end), None) + + return slice(int(start), int(end) + 1 if end else None) + class InfinispanBackend: """ @@ -38,17 +54,31 @@ class InfinispanBackend: def __init__(self, infinispan_config): logger.debug("Creating Infinispan storage client") self.infinispan_config = infinispan_config - self.mech = infinispan_config.get('auth_mech', 'DIGEST') + # the documented key is 'mech'; 'auth_mech' is what the code used to + # read, and is kept working for configurations written against it + self.mech = infinispan_config.get('mech') or \ + infinispan_config.get('auth_mech', 'DIGEST') if self.mech == 'DIGEST': - self.auth = HTTPDigestAuth(infinispan_config.get('username'), - infinispan_config.get('password')) + auth_class = HTTPDigestAuth elif self.mech == 'BASIC': - self.auth = HTTPBasicAuth(infinispan_config.get('username'), - infinispan_config.get('password')) + auth_class = HTTPBasicAuth + else: + raise Exception( + f"Unsupported Infinispan authentication mechanism '{self.mech}'" + ", it must be one of DIGEST or BASIC" + ) + self.auth = auth_class(infinispan_config.get('username'), + infinispan_config.get('password')) self.endpoint = infinispan_config.get('endpoint') self.cache_names = infinispan_config.get('cache_names', ['storage']) self.cache_type = infinispan_config.get('cache_type', 'org.infinispan.DIST_SYNC') self.infinispan_client = requests.session() + # the default pool of 10 throws away connections as soon as a + # listing, or a handful of workers, run requests side by side + adapter = requests.adapters.HTTPAdapter(pool_connections=CONN_POOL_SIZE, + pool_maxsize=CONN_POOL_SIZE) + self.infinispan_client.mount('http://', adapter) + self.infinispan_client.mount('https://', adapter) self.__is_server_version_supported() self.caches = {} @@ -106,6 +136,7 @@ def put_object(self, bucket_name, key, data): auth=self.auth, headers=self.headers) logger.debug(resp) + resp.raise_for_status() def get_object(self, bucket_name, key, stream=False, extra_get_args={}): """ @@ -116,13 +147,12 @@ def get_object(self, bucket_name, key, stream=False, extra_get_args={}): """ url = self.__key_url(bucket_name, key) res = self.infinispan_client.get(url, headers=self.headers, auth=self.auth) - data = res.content - if data is None or len(data) == 0: + if res.status_code == 404: raise StorageNoSuchKeyError(bucket_name, key) + res.raise_for_status() + data = res.content if 'Range' in extra_get_args: - byte_range = extra_get_args['Range'].replace('bytes=', '') - first_byte, last_byte = map(int, byte_range.split('-')) - data = data[first_byte:last_byte + 1] + data = data[_parse_range(extra_get_args['Range'][6:])] if stream: return io.BytesIO(data) return data @@ -181,8 +211,6 @@ def head_object(self, bucket_name, key): :rtype: str/bytes """ obj = self.get_object(bucket_name, key) - if obj is None: - raise StorageNoSuchKeyError(bucket=bucket_name, key=key) return {'content-length': str(len(obj))} def delete_object(self, bucket_name, key): @@ -200,19 +228,21 @@ def delete_objects(self, bucket_name, key_list): :param bucket: bucket name :param key_list: list of keys """ - result = [] - for key in key_list: - self.delete_object(bucket_name, key) - return result + return [self.delete_object(bucket_name, key) for key in key_list] def head_bucket(self, bucket_name): """ Head bucket from COS with a name. Throws StorageNoSuchKeyError if the given bucket does not exist. :param bucket_name: name of the bucket :return: Metadata of the bucket - :rtype: str/bytes + :rtype: dict """ - raise NotImplementedError + url = self.endpoint + '/rest/v2/caches/' + bucket_name + res = self.infinispan_client.head(url, auth=self.auth) + if res.status_code == 404: + raise StorageNoSuchKeyError(bucket_name, '') + res.raise_for_status() + return {'ResponseMetadata': {'HTTPStatusCode': 200}} def list_objects(self, bucket_name, prefix=None, match_pattern=None): """ @@ -233,14 +263,28 @@ def list_objects(self, bucket_name, prefix=None, match_pattern=None): pref = "" else: pref = prefix - for k in j: - if len(k) > 0: - key = k - if key.startswith(pref): - h = self.get_object(bucket_name, key) - d = {'Key': key, 'Size': len(h)} - result.append(d) - return result + keys = [k for k in j if len(k) > 0 and k.startswith(pref)] + if not keys: + return result + + def size_of(key): + # a key listed a moment ago may be gone by now, and a listing + # should skip it rather than fail + try: + return len(self.get_object(bucket_name, key)) + except StorageNoSuchKeyError: + return None + + # The REST API exposes no size metadata, so every value has to be + # read back; overlapping the requests keeps a listing from costing + # one full round trip per key + with ThreadPoolExecutor(max_workers=min(CONN_POOL_SIZE, len(keys))) as pool: + sizes = pool.map(size_of, keys) + + return [ + {'Key': key, 'Size': size} + for key, size in zip(keys, sizes) if size is not None + ] def list_keys(self, bucket_name, prefix=None): """ diff --git a/lithops/storage/backends/infinispan_hotrod/__init__.py b/lithops/storage/backends/infinispan_hotrod/__init__.py deleted file mode 100644 index 996aa27cd..000000000 --- a/lithops/storage/backends/infinispan_hotrod/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .infinispan_hotrod import InfinispanHotrodBackend as StorageBackend - -__all__ = ['StorageBackend'] diff --git a/lithops/storage/backends/infinispan_hotrod/config.py b/lithops/storage/backends/infinispan_hotrod/config.py deleted file mode 100644 index 23df799ed..000000000 --- a/lithops/storage/backends/infinispan_hotrod/config.py +++ /dev/null @@ -1,29 +0,0 @@ -# -# (C) Copyright IBM Corp. 2020 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -REQ_PARAMS = ('endpoint', 'username', 'password') - - -def load_config(config_data): - if 'infinispan_hotrod' not in config_data: - raise Exception("infinispan section is mandatory in the configuration") - - for param in REQ_PARAMS: - if param not in config_data['infinispan_hotrod']: - msg = f"'{param}' is mandatory under 'infinispan' section of the configuration" - raise Exception(msg) - - config_data['infinispan_hotrod']['storage_bucket'] = 'storage' diff --git a/lithops/storage/backends/infinispan_hotrod/infinispan_hotrod.py b/lithops/storage/backends/infinispan_hotrod/infinispan_hotrod.py deleted file mode 100644 index 7629437aa..000000000 --- a/lithops/storage/backends/infinispan_hotrod/infinispan_hotrod.py +++ /dev/null @@ -1,238 +0,0 @@ -# -# (C) Copyright RedHat Inc. 2021 -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -# - -import os -import io -import shutil -import logging -import requests -from requests.auth import HTTPDigestAuth -from lithops.constants import STORAGE_CLI_MSG -from lithops.storage.utils import StorageNoSuchKeyError -from Infinispan import Infinispan -logger = logging.getLogger(__name__) - - -class InfinispanHotrodBackend: - """ - Infinispan Hotrod backend - """ - - def __init__(self, infinispan_config): - logger.debug("Creating Infinispan Hotrod storage client") - self.infinispan_config = infinispan_config - conf = Infinispan.Configuration() - connConf = infinispan_config.get('endpoint').split(":") - conf.addServer(connConf[0], int(connConf[1]) if len(connConf) > 1 else 11222) - conf.setProtocol("2.8") - conf.setSasl("DIGEST-MD5", "node0", infinispan_config.get('username'), infinispan_config.get('password')) - self.conf = conf - self.cacheManager = Infinispan.RemoteCacheManager(conf) - self.cacheManager.start() - self.cacheManagerAdmin = Infinispan.RemoteCacheManagerAdmin(self.cacheManager) - self.basicAuth = HTTPDigestAuth(infinispan_config.get('username'), - infinispan_config.get('password')) - self.cache_names = infinispan_config.get('cache_names', ['storage']) - self.cache_type = infinispan_config.get('cache_type', 'org.infinispan.DIST_SYNC') - self.infinispan_client = requests.session() - - self.caches = {} - for cache_name in self.cache_names: - self.__create_cache(cache_name, self.cache_type) - - msg = STORAGE_CLI_MSG.format('Infinispan_hotrod') - logger.info("{} - Endpoint: {}".format(msg, self.endpoint)) - - def __create_cache(self, cache_name, cache_type): - self.caches[cache_name] = self.cacheManagerAdmin.getOrCreateCache(cache_name, cache_type) - - def __key(self, key): - return key - - def put_object(self, bucket_name, key, data): - """ - Put an object in Infinispan. Override the object if the key already exists. - :param key: key of the object. - :param data: data of the object - :type data: str/bytes/io.BytesIO - :return: None - """ - keyEncoded = self.__key(key) - keyVect = Infinispan.Util.fromString(keyEncoded) - if isinstance(data, str): - dataVec = Infinispan.Util.fromString(data) - elif isinstance(data, io.BytesIO): - r = data.read() - dataVec = Infinispan.UCharVector(r) - elif isinstance(data, bytes): - dataVec = Infinispan.UCharVector(data) - resp = self.caches[bucket_name].put(keyVect, dataVec) - logger.debug(resp) - - def get_object(self, bucket_name, key, stream=False, extra_get_args={}): - """ - Get object from COS with a key. Throws StorageNoSuchKeyError if the given key does not exist. - :param key: key of the object - :return: Data of the object - :rtype: str/bytes - """ - keyEncoded = self.__key(key) - keyVect = Infinispan.Util.fromString(keyEncoded) - resp = self.caches[bucket_name].get(keyVect) - if resp is None: - raise StorageNoSuchKeyError(bucket=bucket_name, key=key) - r = Infinispan.pvuc_value(resp) - b = bytes(r) - if 'Range' in extra_get_args: - byte_range = extra_get_args['Range'].replace('bytes=', '') - first_byte, last_byte = map(int, byte_range.split('-')) - b = b[first_byte:last_byte + 1] - if stream: - return io.BytesIO(b) - return b - - def upload_file(self, file_name, bucket, key=None, extra_args={}, config=None): - """Upload a file - - :param file_name: File to upload - :param bucket: Bucket to upload to - :param key: S3 object name. If not specified then file_name is used - :return: True if file was uploaded, else False - """ - # If S3 key was not specified, use file_name - if key is None: - key = os.path.basename(file_name) - - # Upload the file - try: - with open(file_name, 'rb') as in_file: - self.put_object(bucket, key, in_file) - except Exception as e: - logging.error(e) - return False - return True - - def download_file(self, bucket, key, file_name=None, extra_args={}, config=None): - """Download a file - - :param bucket: Bucket to download from - :param key: S3 object name. If not specified then file_name is used - :param file_name: File to upload - :return: True if file was downloaded, else False - """ - # If file_name was not specified, use S3 key - if file_name is None: - file_name = key - - # Download the file - try: - dirname = os.path.dirname(file_name) - if dirname and not os.path.exists(dirname): - os.makedirs(dirname) - with open(file_name, 'wb') as out: - data_stream = self.get_object(bucket, key, stream=True) - shutil.copyfileobj(data_stream, out) - except Exception as e: - logging.error(e) - return False - return True - - def head_object(self, bucket_name, key): - """ - Head object from COS with a key. Throws StorageNoSuchKeyError if the given key does not exist. - :param key: key of the object - :return: Data of the object - :rtype: str/bytes - """ - fullKey = self.__key(key) - keyVect = Infinispan.Util.fromString(fullKey) - obj = self.caches[bucket_name].get(keyVect) - if obj is None: - raise StorageNoSuchKeyError(bucket=bucket_name, key=key) - return {'content-length': str(obj.size())} - - def delete_object(self, bucket_name, key): - """ - Delete an object from storage. - :param bucket: bucket name - :param key: data key - """ - fullKey = self.__key(key) - self.caches[bucket_name].remove(Infinispan.Util.fromString(fullKey)) - return None - - def delete_objects(self, bucket_name, key_list): - """ - Delete a list of objects from storage. - :param bucket: bucket name - :param key_list: list of keys - """ - result = [] - for key in key_list: - self.delete_object(bucket_name, key) - return result - - def head_bucket(self, bucket_name): - """ - Head bucket from COS with a name. Throws StorageNoSuchKeyError if the given bucket does not exist. - :param bucket_name: name of the bucket - :return: Metadata of the bucket - :rtype: str/bytes - """ - raise NotImplementedError - - def list_objects(self, bucket_name, prefix=None, match_pattern=None): - """ - Return a list of objects for the given bucket and prefix. - :param bucket_name: Name of the bucket. - :param prefix: Prefix to filter object names. - :return: List of objects in bucket that match the given prefix. - :rtype: list of str - """ - keyListAsVec = self.caches[bucket_name].keys() - keyList = [] - if prefix is None: - pref = "" - else: - pref = prefix - for k in keyListAsVec: - if len(k) > 0: - if Infinispan.Util.toString(k).startswith(pref): - o = self.caches[bucket_name].get(k) - if o is not None: - size = len(self.caches[bucket_name].get(k)) - keyList.append({'Key': Infinispan.Util.toString(k), 'Size': size}) - return keyList - - def list_keys(self, bucket_name, prefix=None): - """ - Return a list of keys for the given prefix. - :param bucket_name: Name of the bucket. - :param prefix: Prefix to filter object names. - :return: List of keys in bucket that match the given prefix. - :rtype: list of str - """ - keyListAsVec = self.caches[bucket_name].keys() - keyList = [] - if prefix is None: - pref = "" - else: - pref = prefix - for k in keyListAsVec: - if len(k) > 0: - if Infinispan.Util.toString(k).startswith(pref): - keyList.append(Infinispan.Util.toString(k)) - return keyList diff --git a/lithops/storage/backends/redis/redis.py b/lithops/storage/backends/redis/redis.py index d987b2c1d..f68332141 100644 --- a/lithops/storage/backends/redis/redis.py +++ b/lithops/storage/backends/redis/redis.py @@ -51,9 +51,11 @@ def put_object(self, bucket_name, key, data): :param bucket_name: bucket name :param key: key of the object. :param data: data of the object - :type data: str/bytes + :type data: str/bytes/file-like :return: None """ + if hasattr(data, 'read'): + data = data.read() if not isinstance(data, (str, bytes, bytearray)): raise TypeError(type(data), 'valid types: {}'.format((str, bytes, bytearray))) @@ -91,9 +93,15 @@ def get_object(self, bucket_name, key, stream=False, extra_get_args={}): redis_key = self._format_key(bucket_name, key) try: if 'Range' in extra_get_args: # expected format: Range='bytes=L-H' - bytes_range = extra_get_args.pop('Range')[6:] - start, end = self._parse_range(bytes_range) - data = self._client.getrange(redis_key, start, end) + start, end = self._parse_range(extra_get_args['Range'][6:]) + pipeline = self._client.pipeline(False) + pipeline.exists(redis_key) + pipeline.getrange(redis_key, start, end) + exists, data = pipeline.execute() + # GETRANGE answers the empty string for a key that is not + # there, which is also a legitimate answer for one that is + if not exists: + data = None else: data = self._client.get(redis_key) @@ -164,13 +172,16 @@ def head_object(self, bucket_name, key): :rtype: dict """ redis_key = self._format_key(bucket_name, key) - try: - meta = self._client.debug_object(redis_key) - except redis.exceptions.ResponseError: + + pipeline = self._client.pipeline(False) + pipeline.exists(redis_key) + pipeline.strlen(redis_key) + exists, length = pipeline.execute() + + if not exists: raise StorageNoSuchKeyError(bucket_name, key) - meta['content-length'] = meta['serializedlength'] - 1 - return meta + return {'content-length': str(length)} def delete_object(self, bucket_name, key): """ @@ -186,6 +197,9 @@ def delete_objects(self, bucket_name, key_list): :param bucket_name: bucket name :param key_list: list of keys """ + if not key_list: + return + redis_key_list = [self._format_key(bucket_name, k) for k in key_list] pipeline = self._client.pipeline(False) @@ -206,7 +220,10 @@ def head_bucket(self, bucket_name): :return: metadata of the bucket :rtype: dict """ - return bool(self._client.exists(self._format_key(bucket_name, ''))) + if not self._client.exists(self._format_key(bucket_name, '')): + raise StorageNoSuchKeyError(bucket_name, '') + + return {'ResponseMetadata': {'HTTPStatusCode': 200}} def list_objects(self, bucket_name, prefix=None, match_pattern=None): """ @@ -216,10 +233,24 @@ def list_objects(self, bucket_name, prefix=None, match_pattern=None): :return: List of objects in bucket that match the given prefix. :rtype: list of dict """ + keys = self.list_keys(bucket_name, prefix) + + # STRLEN keeps the listing off the wire: the sizes are all that is + # needed and the bodies can be arbitrarily large. EXISTS goes with + # it because STRLEN answers 0 for a key that is not there, and the + # directory sets outlive a value that was evicted or expired - a + # phantom has to be dropped, not reported as a zero-byte object pipeline = self._client.pipeline(False) - for key in self.list_keys(bucket_name, prefix): - pipeline.get(self._format_key(bucket_name, key)) - return pipeline.execute() + for key in keys: + redis_key = self._format_key(bucket_name, key) + pipeline.exists(redis_key) + pipeline.strlen(redis_key) + res = pipeline.execute() + + return [ + {'Key': key, 'Size': size} + for key, exists, size in zip(keys, res[::2], res[1::2]) if exists + ] def list_keys(self, bucket_name, prefix=None): """ @@ -233,52 +264,46 @@ def list_keys(self, bucket_name, prefix=None): redis_prefix = self._format_key(bucket_name, prefix) pdir = '/'.join(redis_prefix.split('/')[:-1]) + '/' - dir_keys = [key.decode() for key in self._client.smembers(pdir)] key_list = [] + pending = [] - for key in dir_keys: - full_key = pdir + key + for member in self._client.smembers(pdir): + full_key = pdir + member.decode() if full_key.startswith(redis_prefix): - if full_key.endswith('/'): - key_list.extend(self._walk(bucket_name, full_key)) - else: - key_list.append(full_key) + target = pending if full_key.endswith('/') else key_list + target.append(full_key) + + # Breadth-first, one pipelined round trip per level of the tree. + # Descending one directory at a time costs a round trip per + # directory, and a job holds one directory per activation + while pending: + pipeline = self._client.pipeline(False) + for dir_key in pending: + pipeline.smembers(dir_key) + + next_level = [] + for dir_key, members in zip(pending, pipeline.execute()): + for member in members: + full_key = dir_key + member.decode() + target = next_level if full_key.endswith('/') else key_list + target.append(full_key) + pending = next_level offset = len(bucket_name) + 1 return [key[offset:] for key in key_list] - def _walk(self, bucket_name, dir_key): - dir_keys = [key.decode() for key in self._client.smembers(dir_key)] - key_list = [] - - for key in dir_keys: - full_key = dir_key + key - if full_key.endswith('/'): - key_list.extend(self._walk(bucket_name, full_key)) - else: - key_list.append(full_key) - - return key_list - def _format_key(self, bucket, key): return '/'.join([bucket, key]) def _parse_range(self, bytes_range): - if '--' in bytes_range: - bytes_range = bytes_range.replace('--', '-') - sign = -1 - else: - sign = 1 + """ + Translates an HTTP byte range into the (start, end) pair GETRANGE + wants, where both ends are inclusive and -1 is the last byte. + Accepts 'L-H', 'L-' (from L to the end) and '-N' (the last N bytes) + """ + start, _, end = bytes_range.partition('-') - if '-' in bytes_range: - bytes_range = bytes_range.split('-') - if bytes_range[0] == '': - end = int(bytes_range[1]) * -1 - start = end - else: - start = int(bytes_range[0]) - end = int(bytes_range[1]) * sign - else: - start = end = int(bytes_range) + if not start: # '-N': the last N bytes + return -int(end), -1 - return start, end + return int(start), int(end) if end else -1 From 2e8ef580cbc0de262c1055919302b96ba200dd2b Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 22:02:57 +0200 Subject: [PATCH 6/9] Update docs --- CHANGELOG.md | 3 --- 1 file changed, 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d1f628379..a0934763b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,8 +12,6 @@ - [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()`). -- [Docs] Added architecture diagrams for the localhost, VM, AWS EC2, Azure VMs and IBM VPC backends. -- [Docs] Added the `lithops.concurrent.futures` reference, the localhost v1/v2 guide, and rewrote the monitoring docs. - [Tests] Added a unit test suite for all non-backend modules. ### Changed @@ -41,7 +39,6 @@ - [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] `imap()` and `imap_unordered()` now default to the configured chunksize. -- [Docs] Removed the `website/` sources; the landing page now lives in the documentation site. ### Fixed From 26f14bea2b9c71a16ecab1a8a0ca10f3eb9a0c21 Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 22:18:00 +0200 Subject: [PATCH 7/9] Update CI --- .github/workflows/tests-all-os.yml | 4 +++- .github/workflows/tests.yml | 4 +++- lithops/multiprocessing/synchronize.py | 1 + 3 files changed, 7 insertions(+), 2 deletions(-) diff --git a/.github/workflows/tests-all-os.yml b/.github/workflows/tests-all-os.yml index 7459380f8..da5d58e7d 100644 --- a/.github/workflows/tests-all-os.yml +++ b/.github/workflows/tests-all-os.yml @@ -6,7 +6,7 @@ on: jobs: localhost_tests: runs-on: ${{ matrix.os }} - timeout-minutes: 6 + timeout-minutes: 20 env: OBJC_DISABLE_INITIALIZE_FORK_SAFETY: YES @@ -60,6 +60,8 @@ jobs: uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} + cache: 'pip' + cache-dependency-path: setup.py - name: Install Lithops run: | diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index a8479f0ed..6836e1962 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -15,7 +15,7 @@ jobs: localhost_tests: runs-on: ubuntu-22.04 - timeout-minutes: 5 + timeout-minutes: 15 strategy: fail-fast: false @@ -30,6 +30,8 @@ jobs: uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} + cache: 'pip' + cache-dependency-path: setup.py - name: Install Lithops run: | diff --git a/lithops/multiprocessing/synchronize.py b/lithops/multiprocessing/synchronize.py index 92756b653..64f0e5273 100644 --- a/lithops/multiprocessing/synchronize.py +++ b/lithops/multiprocessing/synchronize.py @@ -56,6 +56,7 @@ def _blpop(client, name, timeout): ) return client.blpop([name], timeout=math.ceil(timeout)) + # # Constants # From 2eceb32a5a4273bcee529e8cedd230130197a66c Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 23:37:34 +0200 Subject: [PATCH 8/9] Update CI --- .github/workflows/tests-all-os.yml | 14 ++++++-------- .github/workflows/tests.yml | 13 ++++++++++++- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/.github/workflows/tests-all-os.yml b/.github/workflows/tests-all-os.yml index da5d58e7d..17ce1375a 100644 --- a/.github/workflows/tests-all-os.yml +++ b/.github/workflows/tests-all-os.yml @@ -23,10 +23,8 @@ jobs: python-version: "3.12" - os: ubuntu-latest python-version: "3.13" - - os: ubuntu-22.04 - python-version: "3.10" - - os: ubuntu-22.04 - python-version: "3.11" + - os: ubuntu-latest + python-version: "3.14" # macOS - os: macos-latest @@ -37,10 +35,8 @@ jobs: python-version: "3.12" - os: macos-latest python-version: "3.13" - - os: macos-15 - python-version: "3.10" - - os: macos-15 - python-version: "3.11" + - os: macos-latest + python-version: "3.14" # Windows - os: windows-latest @@ -51,6 +47,8 @@ jobs: python-version: "3.12" - os: windows-latest python-version: "3.13" + - os: windows-latest + python-version: "3.14" steps: - name: Clone Lithops repository diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 6836e1962..dcfe35315 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -14,9 +14,20 @@ on: jobs: localhost_tests: - runs-on: ubuntu-22.04 + runs-on: ubuntu-24.04 timeout-minutes: 15 + services: + redis: + image: redis:7 + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + strategy: fail-fast: false matrix: From 51c627f71fbb43784742c103c0bf9d73000fa8fa Mon Sep 17 00:00:00 2001 From: JosepSampe Date: Sun, 6 Sep 2026 23:45:30 +0200 Subject: [PATCH 9/9] Update CI --- .github/workflows/ibm-ce-tests.yml | 19 +++++----- .github/workflows/python-linting.yml | 16 +++++---- .github/workflows/tests-all-os.yml | 52 ++++++---------------------- .github/workflows/tests.yml | 12 +++++-- setup.cfg | 9 +++++ setup.py | 1 + 6 files changed, 50 insertions(+), 59 deletions(-) create mode 100644 setup.cfg diff --git a/.github/workflows/ibm-ce-tests.yml b/.github/workflows/ibm-ce-tests.yml index 038793f28..c593f531e 100644 --- a/.github/workflows/ibm-ce-tests.yml +++ b/.github/workflows/ibm-ce-tests.yml @@ -12,7 +12,7 @@ jobs: determine_runnable_test_jobs: runs-on: ubuntu-latest - if: github.event_name == 'pull_request' || github.event_name == 'workflow_dispatch' + timeout-minutes: 5 outputs: localhost: ${{ steps.script.outputs.localhost }} @@ -23,20 +23,22 @@ jobs: id: script run: | echo "localhost=true" >> $GITHUB_OUTPUT - echo "code_engine=false" >> $GITHUB_OUTPUT + # The Code Engine job needs the config secret to run at all + echo "code_engine=$HAVE_LITHOPS_CONFIG" >> $GITHUB_OUTPUT localhost_tests: runs-on: ubuntu-latest + timeout-minutes: 15 needs: determine_runnable_test_jobs if: needs.determine_runnable_test_jobs.outputs.localhost == 'true' steps: - name: Clone Lithops repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Install Python 3.10 - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.10' @@ -47,20 +49,21 @@ jobs: - name: Run Lithops tests run: | cd lithops/tests - pytest -v --backend localhost --storage localhost + pytest -v --timeout=120 --timeout-method=thread --backend localhost --storage localhost ibm_ce_cos_tests: runs-on: ubuntu-latest + timeout-minutes: 60 needs: determine_runnable_test_jobs if: needs.determine_runnable_test_jobs.outputs.code_engine == 'true' steps: - name: Clone Lithops repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Install Python 3.10 - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.10' @@ -75,7 +78,7 @@ jobs: - name: Build new runtime run: | - docker login -u ${{ secrets.DOCKER_USER }} -p ${{ secrets.DOCKER_TOKEN }} + echo '${{ secrets.DOCKER_TOKEN }}' | docker login -u '${{ secrets.DOCKER_USER }}' --password-stdin cd runtime/code_engine lithops runtime build -f Dockerfile.githubci ${{ secrets.DOCKER_USER }}/lithops-ce-gihub-ci:${{ github.run_id }} -b code_engine sed -i '/runtime: lithops-ce/c\ runtime: '${{ secrets.DOCKER_USER }}'/lithops-ce-gihub-ci:'${{ github.run_id }} $LITHOPS_CONFIG_FILE diff --git a/.github/workflows/python-linting.yml b/.github/workflows/python-linting.yml index 0e953752b..bbf84c444 100644 --- a/.github/workflows/python-linting.yml +++ b/.github/workflows/python-linting.yml @@ -7,21 +7,27 @@ on: paths: - 'setup.py' - 'lithops/**' + - '.github/workflows/python-linting.yml' workflow_dispatch: # this allows to run the workflow manually through the github dashboard +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: flake8: runs-on: ubuntu-latest + timeout-minutes: 10 steps: - name: Clone Lithops repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Install Python 3.10 - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: '3.10' @@ -30,10 +36,6 @@ jobs: python3 -m pip install --upgrade pip pip3 install -U flake8 - - name: Install Lithops - run: | - pip3 install -U . - - name: Lint with flake8 run: | - flake8 lithops --count --max-line-length=180 --statistics --ignore W605,W503 + flake8 lithops --count --statistics diff --git a/.github/workflows/tests-all-os.yml b/.github/workflows/tests-all-os.yml index 17ce1375a..2cb93c12c 100644 --- a/.github/workflows/tests-all-os.yml +++ b/.github/workflows/tests-all-os.yml @@ -11,56 +11,27 @@ jobs: OBJC_DISABLE_INITIALIZE_FORK_SAFETY: YES strategy: - fail-fast: False + fail-fast: false matrix: - include: - # Linux - - os: ubuntu-latest - python-version: "3.10" - - os: ubuntu-latest - python-version: "3.11" - - os: ubuntu-latest - python-version: "3.12" - - os: ubuntu-latest - python-version: "3.13" - - os: ubuntu-latest - python-version: "3.14" - - # macOS - - os: macos-latest - python-version: "3.10" - - os: macos-latest - python-version: "3.11" - - os: macos-latest - python-version: "3.12" - - os: macos-latest - python-version: "3.13" - - os: macos-latest - python-version: "3.14" - - # Windows - - os: windows-latest - python-version: "3.10" - - os: windows-latest - python-version: "3.11" - - os: windows-latest - python-version: "3.12" - - os: windows-latest - python-version: "3.13" - - os: windows-latest - python-version: "3.14" + os: [ubuntu-latest, macos-latest, windows-latest] + python-version: ['3.10', '3.11', '3.12', '3.13', '3.14'] steps: - name: Clone Lithops repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Install Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} cache: 'pip' cache-dependency-path: setup.py + - name: Install Redis + uses: shogo82148/actions-setup-redis@v1 + with: + redis-version: '7.x' + - name: Install Lithops run: | pip3 install -U .[tests] @@ -78,8 +49,7 @@ jobs: - name: Run Lithops tests run: | cd lithops/tests - # pytest -v --durations=0 --backend localhost --storage localhost - pytest -v --durations=0 -o log_cli=true --log-cli-level=DEBUG --backend localhost --storage localhost + pytest -v --durations=0 --timeout=120 --timeout-method=thread --backend localhost --storage localhost - name: Display last 500 lines of the Lithops log file if: cancelled() || failure() diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index dcfe35315..4d4ec9e37 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -7,10 +7,16 @@ on: paths: - 'setup.py' - 'lithops/**' + - '.github/workflows/tests.yml' workflow_dispatch: # this allows to run the workflow manually through the github dashboard +# A new push to a PR supersedes the run that is still going for it +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: localhost_tests: @@ -35,10 +41,10 @@ jobs: steps: - name: Clone Lithops repository - uses: actions/checkout@v4 + uses: actions/checkout@v5 - name: Install Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v6 with: python-version: ${{ matrix.python-version }} cache: 'pip' @@ -51,4 +57,4 @@ jobs: - name: Run Lithops tests run: | cd lithops/tests - pytest -v --backend localhost --storage localhost + pytest -v --timeout=120 --timeout-method=thread --backend localhost --storage localhost diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 000000000..02cc59ebe --- /dev/null +++ b/setup.cfg @@ -0,0 +1,9 @@ +[flake8] +max-line-length = 180 +extend-ignore = W605, W503 +exclude = + .git, + __pycache__, + build, + dist, + *.egg-info diff --git a/setup.py b/setup.py index 7a180fb46..bb1dee9fe 100644 --- a/setup.py +++ b/setup.py @@ -84,6 +84,7 @@ ], 'tests': [ 'pytest', + 'pytest-timeout', 'kubernetes', 'pika', 'ibm-cos-sdk',