10

シグナルで完全に動作するコンテキスト マネージャーがありtimeoutますが、シグナルはメイン スレッドでのみ機能するため、マルチスレッド モードでエラーが発生します。

def timeout_handler(signum, frame):
    raise TimeoutException()

@contextmanager
def timeout(seconds):
    old_handler = signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(seconds)
    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

のデコレータの実装を見てきましたが、から派生したクラス内timeoutで渡す方法がわかりません。私のバリアントは機能しません。yieldthreading.Thread

@contextmanager
def timelimit(seconds):
    class FuncThread(threading.Thread):
        def run(self):
            yield

    it = FuncThread()        
    it.start()
    it.join(seconds)

    if it.isAlive():
        raise TimeoutException()
4

4 に答える 4

13

コンテキスト マネージャーによって保護されたコードがループ ベースである場合は、スレッドの強制終了を処理する方法でこれを処理することを検討してください。通常、別のスレッドを強制終了することは安全ではないため、標準的な方法は、ワーカー スレッドから見えるフラグを制御スレッドに設定させることです。ワーカー スレッドは定期的にそのフラグをチェックし、正常にシャットダウンします。タイムアウトに類似したことを行う方法は次のとおりです。

class timeout(object):
    def __init__(self, seconds):
        self.seconds = seconds
    def __enter__(self):
        self.die_after = time.time() + self.seconds
        return self
    def __exit__(self, type, value, traceback):
        pass
    @property
    def timed_out(self):
        return time.time() > self.die_after

シングルスレッドの使用例を次に示します。

with timeout(1) as t:
    while True: # this will take a long time without a timeout
        # periodically check for timeouts
        if t.timed_out:
            break # or raise an exception
        # do some "useful" work
        print "."
        time.sleep(0.2)

そしてマルチスレッドのもの：

import thread
def print_for_n_secs(string, seconds):
    with timeout(seconds) as t:
        while True:
            if t.timed_out:
                break # or raise an exception
            print string,
            time.sleep(0.5)

for i in xrange(5):
    thread.start_new_thread(print_for_n_secs,
                            ('thread%d' % (i,), 2))
    time.sleep(0.25)

このアプローチは、シグナルを使用するよりも侵入的ですが、任意のスレッドに対して機能します。

于 2013-03-03T20:12:09.233 に答える
3

コンテキストマネージャーであなたが提案していることを行う方法がわかりませんyield。あるスレッドから別のスレッドへの流れはできません。私がすることは、タイムアウトのある割り込み可能なスレッドで関数をラップすることです。これがそのための レシピです。

余分なスレッドがあり、構文はそれほど良くありませんが、機能します。

于 2013-02-26T02:05:39.447 に答える
-3

システム コールのタイムアウトはシグナルで行われます。ほとんどのブロッキング システム コールは、シグナルが発生すると EINTR を返すため、アラームを使用してタイムアウトを実装できます。

以下は、ほとんどのシステム コールで動作するコンテキスト マネージャです。時間がかかりすぎると、ブロックしているシステム コールから IOError が発生します。

import signal, errno
from contextlib import contextmanager
import fcntl

@contextmanager
def timeout(seconds):
    def timeout_handler(signum, frame):
        pass

    original_handler = signal.signal(signal.SIGALRM, timeout_handler)

    try:
        signal.alarm(seconds)
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, original_handler)

with timeout(1):
    f = open("test.lck", "w")
    try:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
    except IOError, e:
        if e.errno != errno.EINTR:
            raise e
        print "Lock timed out"
于 2013-02-22T06:57:10.623 に答える