当解释器循环本身是递归的时,蹦床的堆栈安全性

2024-09-29 00:19:56 发布

您现在位置:Python中文网/ 问答频道 /正文

为了编写具有堆栈安全性的递归函数,我在Python中实现了一个蹦床(因为CPython没有TCO特性)。看起来是这样的:

from typing import Generic, TypeVar
from abc import ABC, abstractmethod

A = TypeVar('A', covariant=True)


class Trampoline(Generic[A], ABC):
    """
    Base class for Trampolines. Useful for writing stack safe-safe
    recursive functions.
    """
    @abstractmethod
    def _resume(self) -> 'Trampoline[A]':
        """
        Let this trampoline resume the interpreter loop
        """
        pass

    @abstractmethod
    def _handle_cont(
        self, cont: Callable[[A], 'Trampoline[B]']
    ) -> 'Trampoline[B]':
        """
        Handle continuation function passed to `and_then`
        """
        pass

    @property
    def _is_done(self) -> bool:
        return isinstance(self, Done)

    def and_then(self, f: Callable[[A], 'Trampoline[B]']) -> 'Trampoline[B]':
        """
        Apply ``f`` to the value wrapped by this trampoline.

        Args:
            f: function to apply the value in this trampoline
        Return:
            Result of applying ``f`` to the value wrapped by \
            this trampoline
        """
        return AndThen(self, f)

    def map(self, f: Callable[[A], B]) -> 'Trampoline[B]':
        """
        Map ``f`` over the value wrapped by this trampoline.

        Args:
            f: function to wrap over this trampoline
        Return:
            new trampoline wrapping the result of ``f``
        """
        return self.and_then(lambda a: Done(f(a)))

    def run(self) -> A:
        """
        Interpret a structure of trampolines to produce a result

        Return:
            result of intepreting this structure of \
            trampolines
        """
        trampoline = self
        while not trampoline._is_done:
            trampoline = trampoline._resume()

        return cast(Done[A], trampoline).a


class Done(Trampoline[A]):
    """
    Represents the result of a recursive computation.
    """
    a: A

    def _resume(self) -> Trampoline[A]:
        return self

    def _handle_cont(self,
                     cont: Callable[[A], Trampoline[B]]) -> Trampoline[B]:
        return cont(self.a)


class Call(Trampoline[A]):
    """
    Represents a recursive call.
    """
    thunk: Callable[[], Trampoline[A]]

    def _handle_cont(self,
                     cont: Callable[[A], Trampoline[B]]) -> Trampoline[B]:
        return self.thunk().and_then(cont)  # type: ignore

    def _resume(self) -> Trampoline[A]:
        return self.thunk()  # type: ignore


class AndThen(Generic[A, B], Trampoline[B]):
    """
    Represents monadic bind for trampolines as a class to avoid
    deep recursive calls to ``Trampoline.run`` during interpretation.
    """
    sub: Trampoline[A]
    cont: Callable[[A], Trampoline[B]]

    def _handle_cont(self,
                     cont: Callable[[B], Trampoline[C]]) -> Trampoline[C]:
        return self.sub.and_then(self.cont).and_then(cont)  # type: ignore

    def _resume(self) -> Trampoline[B]:
        return self.sub._handle_cont(self.cont)  # type: ignore

    def and_then(  # type: ignore
        self, f: Callable[[A], Trampoline[B]]
    ) -> Trampoline[B]:
        return AndThen(
            self.sub,
            lambda x: Call(lambda: self.cont(x).and_then(f))  # type: ignore
        )

现在,我需要一个一元序列算子。我最初的拍摄是这样的:

from typing import Iterable

from functools import reduce


def sequence(iterable: Iterable[Trampoline[A]]) -> Trampoline[Iterable[A]]:
    def combine(result: Trampoline[Iterable[A]], ta: Trampoline[A]) -> Trampoline[Iterable[A]]:
        return result.and_then(lambda as_: ta.map(lambda a: as_ + (a,)))

    return reduce(combine, iterable, Done(()))

这是可行的,但以这种方式减少一长串蹦床所导致的所有函数调用的开销绝对会降低性能

因此,我尝试了以下方法:

def sequence(iterable: Iterable[Trampoline[A]]) -> Trampoline[Iterable[A]]:
    def thunk() -> Trampoline[Iterable[A]]:
        return Done(tuple([t.run() for t in iterable]))
    
    return Call(thunk)

现在,我的直觉是sequence的第二个解决方案不是堆栈安全的,因为它调用的是run,这意味着run将在解释过程中调用run(通过Call.thunk,但不是更少)。然而,无论我如何混合和匹配,我似乎都不会产生堆栈溢出

例如,我认为应该这样做:

t, *ts = [sequence(Done(v) for v in range(2)) for _ in range(10000)]

def combine(t1, t2):
    return t1.and_then(lambda _: t2)

final = reduce(combine, ts, t)
final.run()  # My gut feeling says this should overflow the stack, but it doesn't

我尝试了无数其他示例,但没有堆栈溢出。我的直觉仍然认为这不应该起作用

我需要有人让我相信,以这种方式蹦床解释器循环实际上是堆栈安全的,或者给我一个例子,说明它会溢出堆栈


Tags: andthetorunselfreturndefthis