12

バックグラウンド

遊んでいるとき、私はよく次のような単純な再帰関数を記述します。

def f(a,b):
    if a>=0 and b>=0:
        return min( f(a-1,b) , f(b,a-1) ) # + some cost that depends on a,b
    else:
        return 0

（たとえば、加重編集距離を計算する場合、または再帰的に定義された数式を評価する場合。）

次に、メモ化デコレータを使用して、結果を自動的にキャッシュします。

問題

f（200,10）のようなものを試してみると、次のようになります。

RuntimeError: maximum recursion depth exceeded

再帰的な実装はPythonのスタックスペース/再帰の制限を使い果たすため、これは予想どおりです。

回避策

私は通常、次のいずれかでこの問題を回避します。

  • sys.setrecursionlimitを使用して再帰制限を増やす（約1000の深さまでしか機能しません）
  • forループを使用して、小さい値のキャッシュをいっぱいにします
  • リストを手動スタックとして使用するように関数を変更する（appendおよびpop呼び出しを介して）（つまり、再帰的な実装から反復的な実装に移行する）
  • 代替プログラミング言語の使用

しかし、これらはすべてエラーが発生しやすいと思います。

質問

本当に大きなスタックを持つことの効果をシミュレートする@Bigstackデコレータを作成する方法はありますか？

私の関数は通常、複数の再帰関数呼び出しを行うため、これは末尾再帰と同じではないことに注意してください。実際には、各関数のすべての内部状態をスタックに保存したいと思います。

私が試したこと

ジェネレータ式のリストをスタックとして使用することを考えていました。スタックフレームをプローブすることで、関数が再帰的に呼び出されたときに解決し、例外をトリガーしてデコレーターコードに戻ることができます。ただし、これらのアイデアを組み合わせて実際に機能するものを作成する方法を見つけることはできません。

または、関数の抽象構文ツリーにアクセスして、再帰関数への呼び出しを変換してステートメントを生成することもできますが、これは間違った方向に向かっているようです。

助言がありますか？

編集

確かにPythonを誤用しているように見えますが、私が検討している別のアプローチは、たとえば500スタックフレームのブロックごとに異なるスレッドを使用してから、連続するスレッドの各ペアの間にキューを挿入することです。戻り値のキュー。（各キューには最大で1つのエントリが含まれます。）これはおそらく何らかの理由で機能しないと思いますが、実装を試みた後で理由を理解するだけです。

4

3 に答える 3

6

To get around the recursion limit, you can catch the RuntimeError exception to detect when you've run out of stack space, and then return a continuation-ish function that, when called, restarts the recursion at the level where you ran out of space. Call this (and its return value, and so on) until you get a value, then try again from the top. Once you've memoized the lower levels, the higher levels won't run into a recursion limit, so eventually this will work. Put the repeated-calling-until-it-works in a wrapper function. Basically it's a lazy version of your warming-up-the-cache idea.

Here's an example with a simple recursive "add numbers from 1 to n inclusive" function.

import functools

def memoize(func):
    cache = {}
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        key = args, tuple(sorted(kwargs.items()))
        if key in cache:
            return cache[key]
        else:
            result = func(*args, **kwargs)
            if not callable(result):
                cache[key] = result
            return result
    return wrapper

@memoize
def _addup(n):
    if n < 2:
        return n
    else:
        try:
            result = _addup(n - 1)
        except RuntimeError:
            return lambda: _addup(n)
        else:
            return result if callable(result) else result + n

def addup(n):
    result = _addup(n)
    while callable(result):
        while callable(result):
            result = result()
        result = _addup(n)
    return result

assert addup(5000) == sum(xrange(5001))

Rather than returning the lambda function all the way back up the call chain, we can raise an exception to short-circuit that, which both improves performance and simplifies the code:

# memoize function as above, or you can probably use functools.lru_cache

class UnwindStack(Exception):
    pass

@memoize
def _addup(n):
    if n < 2:
        return n
    else:
        try:
            return _addup(n - 1) + n
        except RuntimeError:
            raise UnwindStack(lambda: _addup(n))

def _try(func, *args, **kwargs):
    try:
        return func(*args, **kwargs)
    except UnwindStack as e:
        return e[0]

def addup(n):
    result = _try(_addup, n)
    while callable(result):
        while callable(result):
            result = _try(result)
        result = _try(_addup, n)
    return result

This remains pretty inelegant, though, and still has a fair amount of overhead, and I can't imagine how you'd make a decorator out it. Python isn't really suited to this kind of thing, I guess.

于 2012-11-22T06:31:53.770 に答える
3

ジェネレータ式のリストをスタックとして使用する実装は次のとおりです。

def run_stackless(frame):
    stack, return_stack = [(False, frame)], []
    while stack:
        active, frame = stack.pop()
        action, res = frame.send(return_stack.pop() if active else None)
        if action == 'call':
            stack.extend([(True, frame), (False, res)])
        elif action == 'tail':
            stack.append((False, res))
        elif action == 'return':
            return_stack.append(res)
        else:
            raise ValueError('Unknown action', action)
    return return_stack.pop()

これを使用するには、次のルールに従って再帰関数を変換する必要があります。

 return expr                    -> yield 'return', expr
 recursive_call(args...)        -> (yield 'call', recursive_call(args...))
 return recursive_call(args...) -> yield 'tail', recursive_call(args...)

たとえば、コスト関数をとして使用するa * bと、関数は次のようになります。

def f(a,b):
    if a>=0 and b>=0:
        yield 'return', min((yield 'call', f(a-1,b)),
                            (yield 'call', f(b,a-1))) + (a * b)
    else:
        yield 'return', 0

テスト：

In [140]: run_stackless(g(30, 4))
Out[140]: 410

Python 2.6.2では、直接呼び出しと比較して、パフォーマンスが最大8〜10倍向上するようです。

tailアクションは末尾再帰用です：

def factorial(n):
    acc = [1]
    def fact(n):
        if n == 0:
            yield 'return', 0
        else:
            acc[0] *= n
            yield 'tail', fact(n - 1)
    run_stackless(fact(n))
    return acc[0]

ジェネレーター再帰スタイルへの変換はかなり簡単で、おそらくバイトコードハックとして実行できます。

于 2012-11-22T17:51:05.590 に答える
2

このアプローチは、メモ化とスタックの深さの増加を1つのデコレータに組み合わせたものです。

スタックの64レベルを担当する各スレッドでスレッドのプールを生成します。
スレッドは一度だけ作成され、再実行されます（ただし、現在は削除されていません）。

キューはスレッド間で情報を渡すために使用されますが、現在のスタックの深さに対応するスレッドのみが実際に実行する必要があることに注意してください。

私の実験では、これにより、単純な再帰関数のオーバーヘッドが約10％増加することが示唆されています（より複雑な関数の場合は、オーバーヘッドが少なくなるはずです）。

import threading,Queue

class BigstackThread(threading.Thread):

    def __init__(self,send,recv,func):
        threading.Thread.__init__( self )
        self.daemon = True
        self.send = send
        self.recv = recv
        self.func = func

    def run(self):
        while 1:
            args = self.send.get()
            v = self.func(*args)
            self.recv.put(v)

class Bigstack(object):

    def __init__(self,func):
        self.func = func
        self.cache = {}
        self.depth = 0
        self.threadpool = {}

    def __call__(self,*args):
        if args in self.cache:
            return self.cache[args]
        self.depth+=1
        if self.depth&63:
            v = self.func(*args)
        else:
            T=self.threadpool
            if self.depth not in T:
                send = Queue.Queue(1)
                recv = Queue.Queue(1)
                t = BigstackThread(send,recv,self)
                T[self.depth] = send,recv,t
                t.start()
            else:
                send,recv,_ = T[self.depth]
            send.put(args)
            v = recv.get()

        self.depth-=1
        self.cache[args]=v
        return v

@Bigstack
def f(a,b):
    if a>=0 and b>=0:
        return min(f(a-1,b),f(b-1,a))+1
    return 0
于 2012-11-22T19:44:35.583 に答える