Add concurrent.futures API (#1485)

Author: JosepSampe

CHANGELOG.md

@@ -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.

README.md

@@ -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
@@ -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

docs/index.rst

@@ -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
@@ -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

docs/source/api_concurrent.rst

@@ -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:

docs/source/api_futures.rst

@@ -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.
 
@@ -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:

docs/source/api_multiprocessing.rst

@@ -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
 

docs/source/functions.md

@@ -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)

docs/source/notebooks/function_chaining.ipynb

@@ -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",

examples/function_chaining.py

@@ -1,6 +1,6 @@
 """
 Simple Lithops example using the function chaining pattern
-in the Futures API.
+in the Core API.
 """
 import lithops
 

lithops/concurrent/__init__.py

@@ -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',
+]