Concurrent Futures API

Concurrent Futures API#

lithops.concurrent.futures is a drop-in for Python’s concurrent.futures Executor interface. Swap the import and the same client keeps working:

# from concurrent.futures import ProcessPoolExecutor, as_completed, wait
from lithops.concurrent.futures import ProcessPoolExecutor, as_completed, wait

def same_client(executor):
    with executor:
        future = executor.submit(pow, 323, 1235)
        print(future.result())
        print(list(executor.map(abs, [-1, 2, -3])))

same_client(ProcessPoolExecutor())

ThreadPoolExecutor is provided under the same name for import compatibility; both names run tasks on Lithops workers.

This is separate from the Core API (lithops.FunctionExecutor), whose map() returns futures, which has call_async() instead of submit(), and whose wait() is a method with Lithops-specific return_when values.

Implemented standard-library surface#

The module exports the names application code actually uses:

  • Executor — submit(), eager map() (results, not futures), shutdown(wait=True, *, cancel_futures=False), context manager

  • ProcessPoolExecutor / ThreadPoolExecutor — Lithops-backed pools

  • Future — subclass of concurrent.futures.Future: result(), exception(), done(), running(), cancel(), cancelled(), add_done_callback()

  • wait / as_completed — same contract and constants (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED)

  • Exceptions — CancelledError, TimeoutError, BrokenExecutor, InvalidStateError

InterpreterPoolExecutor (Python 3.14) is not implemented: Lithops has no isolated-interpreter workers.

map() submits one Lithops map() job for the whole iterable, not one submit() per item. Callables travel through a trampoline, so builtins such as abs and pow work. Its chunksize is how many items each Lithops worker takes, which is what it means for the standard ProcessPoolExecutor too; left unset, the chunksize of the Lithops configuration applies rather than the standard library default of one item per worker.

from lithops.concurrent.futures import FunctionExecutor, as_completed

def load(url):
    return url, len(url)

with FunctionExecutor() as executor:
    futures = {executor.submit(load, url): url for url in ('a', 'bb', 'ccc')}
    for future in as_completed(futures):
        url, size = future.result()
        print(url, size)

Mode-specific subclasses LocalhostExecutor, ServerlessExecutor and StandaloneExecutor pin the Lithops execution mode the same way the Core API executors do. You can wrap an existing Core API executor:

import lithops
from lithops.concurrent.futures import FunctionExecutor

fexec = lithops.FunctionExecutor()
with FunctionExecutor(executor=fexec) as executor:
    print(executor.submit(pow, 2, 8).result())

Runtime differences#

The API matches concurrent.futures. The runtime is Lithops:

  • Workers are Lithops activations, not local threads or multiprocessing processes. mp_context, max_tasks_per_child, and thread_name_prefix are ignored.

  • initializer / initargs are not supported (workers are ephemeral) and raise NotImplementedError if provided.

  • cancel() cannot stop a job Lithops has already dispatched. submit() marks the future as running immediately, so cancel() returns False, and shutdown(cancel_futures=True) therefore still waits the calls out.

  • Lithops job options (runtime_memory, extra_env, execution_timeout, include_modules, exclude_modules) are set on the executor. Keyword arguments to submit(fn, *args, **kwargs) are passed to fn.

  • Each Future also exposes lithops_future and stats.

  • A call Lithops loses track of raises RuntimeError rather than handing back a silent None. It fails that one future; the executor stays usable.

  • RetryingFunctionExecutor cannot be wrapped. Its retries are driven from its own wait(), which this adapter never calls, so wrapping it would quietly give you no retries at all. Wrap the FunctionExecutor it holds.

Completion is tracked by the Lithops job monitor, the same one the native wait() uses: one batched poll per round for the whole job rather than one status read per call. Results are downloaded off the tracking thread, so a slow object does not hold up the futures behind it, and done() never blocks on storage.

concurrent.futures-compatible executors backed by Lithops.

The native Lithops executors (lithops.FunctionExecutor and friends) are intentionally different from concurrent.futures: map() returns futures, there is no submit(), and wait() lives on the executor. This module is the drop-in interface for code that already talks to ThreadPoolExecutor / ProcessPoolExecutor:

from lithops.concurrent.futures import ProcessPoolExecutor

with ProcessPoolExecutor() as executor:
    future = executor.submit(pow, 2, 8)
    print(future.result())
    print(list(executor.map(abs, [-1, 2, -3])))

Future subclasses concurrent.futures.Future, so the standard library’s wait() and as_completed() work unchanged, including with futures from other executors.

class lithops.concurrent.futures.FunctionExecutor(max_workers=None, *, executor=None, initializer=None, initargs=(), runtime_memory=None, extra_env=None, execution_timeout=None, include_modules=None, exclude_modules=None, **kwargs)#

Bases: Executor

concurrent.futures.Executor that runs callables on Lithops workers.

submit(fn, *args, **kwargs) and map(fn, *iterables) follow the standard library: map is eager and yields results, not futures. Internally, map is a single Lithops map() job rather than one submit per item, so a large iterator still benefits from Lithops batching.

Parameters:

  • max_workers – Passed through to the Lithops compute backend

  • executor – An existing lithops.FunctionExecutor (or compatible object) to wrap. When omitted, one is created from **kwargs

  • initializer – Not supported; Lithops workers are ephemeral. Providing a callable raises NotImplementedError

  • initargs – Ignored unless initializer is set

  • runtime_memory – Memory (MB) for every submitted call

  • extra_env – Extra environment variables for every submitted call

  • execution_timeout – Max seconds each function activation may run

  • include_modules – Modules to pickle into the worker payload

  • exclude_modules – Modules to keep out of the worker payload

  • kwargs – Forwarded to the Lithops executor constructor (config, backend, storage, log_level, …)

property lithops_executor#

The wrapped Lithops FunctionExecutor.

map(fn, *iterables, timeout=None, chunksize=None, buffersize=None)#

Eager map, as in the standard library: it returns an iterator over the results, and every call is submitted before it does.

chunksize is how many items each Lithops worker takes, which is what it means for the standard ProcessPoolExecutor too. Left unset, the Lithops configuration decides, rather than the standard library default of one item per worker overriding it

shutdown(wait=True, *, cancel_futures=False)#

Clean-up the resources associated with the Executor.

It is safe to call this method several times. Otherwise, no other methods can be called after this one.

Args:
wait: If True then shutdown will not return until all running

futures have finished executing and the resources used by the executor have been reclaimed.

cancel_futures: If True then shutdown will cancel all pending

futures. Futures that are completed or running will not be cancelled.

submit(fn, /, *args, **kwargs)#

Schedules fn(*args, **kwargs) to run on a Lithops worker.

Parameters:

  • fn – The callable to run

  • args – Positional arguments for fn

  • kwargs – Keyword arguments for fn

Returns:

A Future representing the call

Raises:

  • RuntimeError – If the executor has been shut down

  • concurrent.futures.BrokenExecutor – If the executor stopped working and can no longer run calls

class lithops.concurrent.futures.Future(lithops_future=None, adapter=None)#

Bases: Future

A concurrent.futures.Future backed by a Lithops ResponseFuture.

Created by FunctionExecutor.submit(). The underlying Lithops future is available as lithops_future for stats and other Lithops-specific attributes.

done()#

Return True if the future was cancelled or finished executing.

exception(timeout=None)#

Returns the exception raised by the call, waiting for it to finish.

Parameters:

timeout – Seconds to wait if the call is not done yet. None waits without limit

Returns:

The exception raised by the call, or None if it returned normally

Raises:

  • concurrent.futures.CancelledError – If the future was cancelled

  • TimeoutError – If the call did not finish within timeout

property lithops_future#

The Lithops ResponseFuture this object is tracking.

result(timeout=None)#

Returns the result of the call, waiting for it to finish.

Parameters:

timeout – Seconds to wait if the call is not done yet. None waits without limit

Returns:

The value returned by the call

Raises:

  • concurrent.futures.CancelledError – If the future was cancelled

  • TimeoutError – If the call did not finish within timeout

  • Exception – The exception raised by the call, if it raised one

property stats#

Execution stats from the Lithops future, once they are available.

class lithops.concurrent.futures.LocalhostExecutor(max_workers=None, *, executor=None, initializer=None, initargs=(), runtime_memory=None, extra_env=None, execution_timeout=None, include_modules=None, exclude_modules=None, **kwargs)#

Bases: FunctionExecutor

FunctionExecutor pinned to the Lithops localhost backend.

class lithops.concurrent.futures.ProcessPoolExecutor(max_workers=None, *, executor=None, initializer=None, initargs=(), runtime_memory=None, extra_env=None, execution_timeout=None, include_modules=None, exclude_modules=None, **kwargs)#

Bases: FunctionExecutor

Drop-in replacement for concurrent.futures.ProcessPoolExecutor.

Tasks run on Lithops workers (localhost, serverless, or standalone) instead of a local multiprocessing pool. Constructor arguments that only apply to the standard library (mp_context, max_tasks_per_child) are ignored.

class lithops.concurrent.futures.ServerlessExecutor(max_workers=None, *, executor=None, initializer=None, initargs=(), runtime_memory=None, extra_env=None, execution_timeout=None, include_modules=None, exclude_modules=None, **kwargs)#

Bases: FunctionExecutor

FunctionExecutor pinned to a Lithops serverless backend.

class lithops.concurrent.futures.StandaloneExecutor(max_workers=None, *, executor=None, initializer=None, initargs=(), runtime_memory=None, extra_env=None, execution_timeout=None, include_modules=None, exclude_modules=None, **kwargs)#

Bases: FunctionExecutor

FunctionExecutor pinned to a Lithops standalone backend.

class lithops.concurrent.futures.ThreadPoolExecutor(max_workers=None, *, executor=None, initializer=None, initargs=(), runtime_memory=None, extra_env=None, execution_timeout=None, include_modules=None, exclude_modules=None, **kwargs)#

Bases: FunctionExecutor

Drop-in replacement for concurrent.futures.ThreadPoolExecutor.

Tasks still run on Lithops workers, not in local threads. Use this name when swapping from concurrent.futures import ThreadPoolExecutor. thread_name_prefix is ignored.

lithops.concurrent.futures.as_completed(fs, timeout=None)#

Yield futures as they complete. Same contract as concurrent.futures.as_completed.

lithops.concurrent.futures.wait(fs, timeout=None, return_when='ALL_COMPLETED')#

Wait for futures to complete. Same contract as concurrent.futures.wait.