Add trampolines support, refs #413 (#1668)

* Add trampolines support, refs #413

* More tests

Author: sobolevn

.github/workflows/test.yml

@@ -11,17 +11,20 @@ on:
       - 'docs/requirements.txt'
   workflow_dispatch:
 
+permissions:
+  contents: read
+
 concurrency:
   group: test-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
 jobs:
   build:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
+        python-version: ['3.8', '3.9', '3.10', '3.11']
         task: ['tests', 'typesafety']
 
     steps:

CHANGELOG.md

@@ -6,6 +6,14 @@ incremental in minor, bugfixes only are patches.
 See [0Ver](https://0ver.org/).
 
 
+## 0.22.0 WIP
+
+### Features
+
+- *Breaking*: Drops `python3.7` support
+- Adds `trampolines` support
+
+
 ## 0.21.0
 
 ### Features

docs/index.rst

@@ -39,6 +39,7 @@ Contents
   pages/do-notation.rst
   pages/functions.rst
   pages/curry.rst
+  pages/trampolines.rst
   pages/types.rst
 
 .. toctree::

docs/pages/trampolines.rst

@@ -0,0 +1,44 @@
+.. _trampolines:
+
+Trampolines
+===========
+
+Python does not support TCO (tail call optimization), so any recursion-based
+algorithms become dangerous.
+
+We cannot be sure that
+they won't cause ``RecursionError`` on deeply nested data.
+
+Here's why we need trampolines: they allow to replicate tail call optimization
+by wrapping function calls into :class:`returns.trampolines.Trampoline` objects,
+making recursion-based function *always* safe.
+
+Example:
+
+.. code:: python
+
+  >>> from typing import Union, List
+  >>> from returns.trampolines import Trampoline, trampoline
+
+  >>> @trampoline
+  ... def accumulate(
+  ...     numbers: List[int],
+  ...     acc: int = 0,
+  ... ) -> Union[int, Trampoline[int]]:
+  ...    if not numbers:
+  ...        return acc
+  ...    number = number = numbers.pop()
+  ...    return Trampoline(accumulate, numbers, acc + number)
+
+  >>> assert accumulate([1, 2]) == 3
+  >>> assert accumulate([1, 2, 3]) == 6
+
+The following function is still fully type-safe:
+- ``Trampoline`` object uses ``ParamSpec`` to be sure that passed arguments are correct
+- Final return type of the function is narrowed to contain only an original type (without ``Trampoline`` implementation detail)
+
+API Reference
+-------------
+
+.. automodule:: returns.trampolines
+   :members:

poetry.lock

@@ -46,7 +46,7 @@ _ = "returns.contrib.hypothesis._entrypoint"
 
 
 [tool.poetry.dependencies]
-python = "^3.7"
+python = "^3.8"
 
 typing-extensions = ">=4.0,<5.0"
 mypy = { version = "^1.4.0", optional = true }

pyproject.toml

@@ -0,0 +1,94 @@
+from functools import wraps
+from typing import Callable, Generic, TypeVar, Union
+
+from typing_extensions import ParamSpec, final
+
+_ReturnType = TypeVar('_ReturnType')
+_FuncParams = ParamSpec('_FuncParams')
+
+
+@final
+class Trampoline(Generic[_ReturnType]):
+    """
+    Represents a wrapped function call.
+
+    Primitive to convert recursion into an actual object.
+    """
+
+    __slots__ = ('func', 'args', 'kwargs')
+
+    def __init__(  # noqa: WPS451
+        self,
+        func: Callable[_FuncParams, _ReturnType],
+        /,  # We use pos-only here to be able to store `kwargs` correctly.
+        *args: _FuncParams.args,
+        **kwargs: _FuncParams.kwargs,
+    ) -> None:
+        """Save function and given arguments."""
+        self.func = getattr(func, '_orig_func', func)
+        self.args = args
+        self.kwargs = kwargs
+
+    def __call__(self) -> _ReturnType:
+        """Call wrapped function with given arguments."""
+        return self.func(*self.args, **self.kwargs)
+
+
+def trampoline(
+    func: Callable[_FuncParams, Union[_ReturnType, Trampoline[_ReturnType]]],
+) -> Callable[_FuncParams, _ReturnType]:
+    """
+    Convert functions using recursion to regular functions.
+
+    Trampolines allow to unwrap recursion into a regular ``while`` loop,
+    which does not raise any ``RecursionError`` ever.
+
+    Since python does not have TCO (tail call optimization),
+    we have to provide this helper.
+
+    This is done by wrapping real function calls into
+    :class:`returns.trampolines.Trampoline` objects:
+
+    .. code:: python
+
+        >>> from typing import Union
+        >>> from returns.trampolines import Trampoline, trampoline
+
+        >>> @trampoline
+        ... def get_factorial(
+        ...     for_number: int,
+        ...     current_number: int = 0,
+        ...     acc: int = 1,
+        ... ) -> Union[int, Trampoline[int]]:
+        ...     assert for_number >= 0
+        ...     if for_number <= current_number:
+        ...         return acc
+        ...     return Trampoline(
+        ...         get_factorial,
+        ...         for_number,
+        ...         current_number=current_number + 1,
+        ...         acc=acc * (current_number + 1),
+        ...     )
+
+        >>> assert get_factorial(0) == 1
+        >>> assert get_factorial(3) == 6
+        >>> assert get_factorial(4) == 24
+
+    See also:
+        - eli.thegreenplace.net/2017/on-recursion-continuations-and-trampolines
+        - https://en.wikipedia.org/wiki/Tail_call
+
+    """
+
+    @wraps(func)
+    def decorator(
+        *args: _FuncParams.args,
+        **kwargs: _FuncParams.kwargs,
+    ) -> _ReturnType:
+        trampoline_result = func(*args, **kwargs)
+        while isinstance(trampoline_result, Trampoline):
+            trampoline_result = trampoline_result()
+        return trampoline_result
+
+    decorator._orig_func = func  # type: ignore[attr-defined]  # noqa: WPS437
+    return decorator

returns/trampolines.py

@@ -0,0 +1,50 @@
+import sys
+from typing import Callable, Iterator, Union
+
+import pytest
+
+from returns.trampolines import Trampoline, trampoline
+
+
+@trampoline
+def _accumulate(
+    numbers: Iterator[int],
+    acc: int = 0,
+) -> Union[int, Trampoline[int]]:
+    number = next(numbers, None)
+    if number is None:
+        return acc
+    return Trampoline(_accumulate, numbers, acc + number)
+
+
+@trampoline
+def _with_func_kwarg(
+    numbers: Iterator[int],
+    func: int = 0,  # we need this name to match `Trampoline` constructor
+) -> Union[int, Trampoline[int]]:
+    number = next(numbers, None)
+    if number is None:
+        return func
+    return Trampoline(_with_func_kwarg, numbers, func=func + number)
+
+
+@pytest.mark.parametrize('trampoline_func', [
+    _accumulate,
+    _with_func_kwarg,
+])
+@pytest.mark.parametrize('given_range', [
+    range(0),
+    range(1),
+    range(2),
+    range(5),
+    range(sys.getrecursionlimit()),
+    range(sys.getrecursionlimit() + 1),
+])
+def test_recursion_limit(
+    trampoline_func: Callable[[Iterator[int]], int],
+    given_range: range,
+) -> None:
+    """Test that accumulation is correct and no ``RecursionError`` happens."""
+    accumulated = trampoline_func(iter(given_range))
+
+    assert accumulated == sum(given_range)

tests/test_trampolines/test_trampoline_decorator.py

@@ -0,0 +1,47 @@
+- case: trampoline_missing_args
+  disable_cache: false
+  main: |
+    from typing import List, Union
+    from returns.trampolines import Trampoline, trampoline
+
+    @trampoline
+    def _accumulate(
+        numbers: List[int],
+        acc: int = 0,
+    ) -> Union[int, Trampoline[int]]:
+        return Trampoline(_accumulate)
+  out: |
+    main:9: error: Missing positional argument "numbers" in call to "Trampoline"  [call-arg]
+
+
+- case: trampoline_wrong_args
+  disable_cache: false
+  main: |
+    from typing import List, Union
+    from returns.trampolines import Trampoline, trampoline
+
+    @trampoline
+    def _accumulate(
+        numbers: List[int],
+        acc: int = 0,
+    ) -> Union[int, Trampoline[int]]:
+        return Trampoline(_accumulate, ['a'], 'b')
+  out: |
+    main:9: error: List item 0 has incompatible type "str"; expected "int"  [list-item]
+    main:9: error: Argument 3 to "Trampoline" has incompatible type "str"; expected "int"  [arg-type]
+
+
+- case: trampoline_return_type
+  disable_cache: false
+  main: |
+    from typing import List, Union
+    from returns.trampolines import Trampoline, trampoline
+
+    @trampoline
+    def _accumulate(
+        numbers: List[int],
+        acc: int = 0,
+    ) -> Union[int, Trampoline[int]]:
+        return Trampoline(_accumulate, [1], 2)
+
+    reveal_type(_accumulate([1, 2]))  # N: Revealed type is "builtins.int"