Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

### Added

- [API] Added `lithops.concurrent.futures`, a `concurrent.futures`-compatible executor interface (`submit`, eager `map`, stdlib `wait`/`as_completed`) backed by Lithops. See issue #1427.
- [Docs] Renamed the native executor documentation from "Futures API" to "Core API".
- [Tests] Added a unit test suite for all non-backend modules (18 files, 876 tests).
- [Core] Added a `log_prefix()` helper for uniform log prefixes across core and backends.
- [Core] Added a cache of serialized functions to avoid re-uploading the same function.
Expand Down
16 changes: 14 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -148,9 +148,9 @@ Supported backends by platform:

## High-level API

Lithops provides two high-level compute APIs and two high-level storage APIs.
Lithops provides a native compute API, two drop-in replacements for standard Python APIs, and two storage APIs.

### [Futures API](docs/source/api_futures.rst)
### [Core API](docs/source/api_futures.rst)

```python
from lithops import FunctionExecutor
Expand All @@ -163,6 +163,18 @@ with FunctionExecutor() as fexec:
print(f.result())
```

### [Concurrent Futures API](docs/source/api_concurrent.rst)

```python
from lithops.concurrent.futures import ProcessPoolExecutor

def double(i):
return i * 2

with ProcessPoolExecutor() as executor:
print(list(executor.map(double, [1, 2, 3, 4])))
```

### [Multiprocessing API](docs/source/api_multiprocessing.rst)

```python
Expand Down
3 changes: 2 additions & 1 deletion docs/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,7 @@ Lithops is an open-source project, actively maintained and supported by a commun
:maxdepth: 0
:caption: Lithops Compute API

source/api_futures.rst
Core API <source/api_futures.rst>
source/functions.md
source/worker_granularity.rst
source/notebooks/function_chaining.ipynb
Expand All @@ -264,6 +264,7 @@ Lithops is an open-source project, actively maintained and supported by a commun
:maxdepth: 0
:caption: Integrations

source/api_concurrent.rst
source/api_multiprocessing.rst
source/api_storage_os.rst
source/sklearn_joblib.rst
Expand Down
112 changes: 112 additions & 0 deletions docs/source/api_concurrent.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,112 @@
Concurrent Futures API
======================

``lithops.concurrent.futures`` is a drop-in for Python's
`concurrent.futures <https://docs.python.org/3/library/concurrent.futures.html>`_
**Executor interface**. Swap the import and the same client keeps working:

.. code:: python

# 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 :doc:`Core API <api_futures>` (``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.

.. code:: python

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.

.. automodule:: lithops.concurrent.futures
:members: FunctionExecutor, LocalhostExecutor, ServerlessExecutor,
StandaloneExecutor, ProcessPoolExecutor, ThreadPoolExecutor,
Future, wait, as_completed
:show-inheritance:
9 changes: 5 additions & 4 deletions docs/source/api_futures.rst
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
.. _core-api:
.. _futures-api:

Lithops Futures API
===================
Lithops Core API
================

The core abstraction in Lithops is the **executor**, which is responsible for orchestrating the execution of your functions across different environments.

Expand Down Expand Up @@ -41,8 +42,8 @@ By default, executors load configuration from the Lithops configuration file (e.
This layered executor design lets Lithops provide a powerful, unified API for parallel function execution — from local development to multi-cloud production deployments, with fault tolerance and retries built in.


Futures API Reference
---------------------
Core API Reference
------------------

.. automodule:: lithops.executors
:members:
Expand Down
2 changes: 1 addition & 1 deletion docs/source/api_multiprocessing.rst
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ Before utilizing this API, you will need to install its dependencies:
Process and Pool
----------------

`Processes <https://docs.python.org/3/library/multiprocessing.html#the-process-class>`_ and `Pool <https://docs.python.org/3/library/multiprocessing.html#using-a-pool-of-workers>`_ are the abstractions used in multiprocessing to parallelize computation. They interact directly with Lithops' Futures API.
`Processes <https://docs.python.org/3/library/multiprocessing.html#the-process-class>`_ and `Pool <https://docs.python.org/3/library/multiprocessing.html#using-a-pool-of-workers>`_ are the abstractions used in multiprocessing to parallelize computation. They interact directly with Lithops' Core API.

.. code:: python

Expand Down
2 changes: 1 addition & 1 deletion docs/source/functions.md
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ This document describes how to invoke functions based on the *iterdata* variable

Reserved parameters
-------------------
Reserved parameters are only accessible when using the [Futures API](./api_futures.rst).
Reserved parameters are only accessible when using the [Core API](./api_futures.rst).


- **id**: To get the call id. For instance, if you spawn 10 activations of a function, you will get here a number from 0 to 9, for example: [map.py](https://github.com/lithops-cloud/lithops/blob/master/examples/map.py)
Expand Down
2 changes: 1 addition & 1 deletion docs/source/notebooks/function_chaining.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
"Lithops does not download the intermediate results to the local client; instead, the intermediate results are read\n",
"directly by the next function.\n",
"\n",
"It currently works with the Futures API, and you can chain the `map()`, `map_reduce()`, `wait()`, and `get_result()`\n",
"It currently works with the Core API, and you can chain the `map()`, `map_reduce()`, `wait()`, and `get_result()`\n",
"methods. Note that the returning value of one function must match the signature of the next function when chaining\n",
"multiple `map()` calls. View the next examples:\n",
"\n",
Expand Down
2 changes: 1 addition & 1 deletion examples/function_chaining.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""
Simple Lithops example using the function chaining pattern
in the Futures API.
in the Core API.
"""
import lithops

Expand Down
53 changes: 53 additions & 0 deletions lithops/concurrent/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
#
# 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.
#

from .futures import (
ALL_COMPLETED,
FIRST_COMPLETED,
FIRST_EXCEPTION,
BrokenExecutor,
CancelledError,
Executor,
FunctionExecutor,
Future,
InvalidStateError,
LocalhostExecutor,
ProcessPoolExecutor,
ServerlessExecutor,
StandaloneExecutor,
ThreadPoolExecutor,
TimeoutError,
as_completed,
wait,
)

__all__ = [
'ALL_COMPLETED',
'FIRST_COMPLETED',
'FIRST_EXCEPTION',
'BrokenExecutor',
'CancelledError',
'Executor',
'FunctionExecutor',
'Future',
'InvalidStateError',
'LocalhostExecutor',
'ProcessPoolExecutor',
'ServerlessExecutor',
'StandaloneExecutor',
'ThreadPoolExecutor',
'TimeoutError',
'as_completed',
'wait',
]
Loading
Loading