在Python中处理阻塞函数调用

首先,应不惜一切代价避免使用信号:

1) 这可能导致僵局。SIGALRM可能在阻塞系统调用之前到达进程(想象一下系统中的超高负载!)系统调用不会被中断。僵局。

相信我-我以前遇到过两个问题,它们一点都不好玩。

在某些情况下,可以明确地避免阻塞—我强烈建议使用select()和friends(在Python中选中select模块)来处理阻塞的写入和读取。不过,这并不能解决阻塞open()调用的问题。

为此,我已经测试了这个解决方案,它对命名管道很有效。它以非阻塞方式打开,然后将其关闭,如果没有可用的内容,则使用select()调用最终超时。

import sys, os, select, fcntl

f = os.open(sys.argv[1], os.O_RDONLY | os.O_NONBLOCK)

flags = fcntl.fcntl(f, fcntl.F_GETFL, 0)
fcntl.fcntl(f, fcntl.F_SETFL, flags & ~os.O_NONBLOCK)

r, w, e = select.select([f], [], [], 2.0)

if r == [f]:
    print 'ready'
    print os.read(f, 100)
else:
    print 'unready'

os.close(f)

mkfifo /tmp/fifo
python <code_above.py> /tmp/fifo (1st terminal)
echo abcd > /tmp/fifo (2nd terminal)

通过一些额外的努力,select()调用可以用作整个程序的主循环,聚合所有事件-您可以使用libev或libevent,或者使用一些Python包装器来包装它们。

恐怕一般来说,你不能用一种强有力的方法来解决这个问题——这真的取决于你阻止了什么。

IIUC,每个顶层块都有一个停止方法。因此,您实际上可以在线程中运行top_块,并在超时到达时发出停止。如果top_块的wait()也有超时,那会更好,但唉,它没有。

在主线程中,您需要等待两种情况:a)top_块完成,b)超时过期。忙碌的等待是有害的:-),所以应该使用线程的带超时连接来等待线程。如果连接后线程仍处于活动状态,则需要停止顶部运行。

您可以设置一个信号警报,它会在超时时中断您的通话:

http://docs.python.org/library/signal.html

signal.alarm(1) # 1 second

my_blocking_call()
signal.alarm(0)

如果要确保不会破坏应用程序,也可以设置信号处理程序:

def my_handler(signum, frame):
    pass

signal.signal(signal.SIGALRM, my_handler)

编辑:

import signal, time

def handler(signum, frame):
    print "Timed-out"

def foo():
    # Set the signal handler and a 5-second alarm
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(3)

    # This open() may hang indefinitely
    time.sleep(5)
    signal.alarm(0)          # Disable the alarm


foo()
print "hallo"

问题是:

1. SIGALRM的默认处理程序是中止应用程序,如果设置了处理程序,则它将不再停止应用程序。

2. 接收信号通常会中断系统调用(然后取消阻止应用程序)

你问题中最简单的部分是信号处理。从Python运行时的角度来看,在解释器进行系统调用时接收到的信号作为OSError异常呈现给Python代码,其中 errno 归因于 errno.EINTR

所以这大概和你想的差不多:

    #!/usr/bin/env python
    import signal, os, errno, time

    def handler(signum, frame):
            # print 'Signal handler called with signal', signum
            #raise IOError("Couldn't open device!")
            print "timed out"
            time.sleep(3)


    def foo():
        # Set the signal handler and a 5-second alarm
        signal.signal(signal.SIGALRM, handler)

        try:
            signal.alarm(3)
            # This open() may hang indefinitely
            fd = os.open('/dev/ttys0', os.O_RDWR)
        except OSError, e:
            if e.errno != errno.EINTR:
                raise e
        signal.alarm(0)          # Disable the alarm

    foo()
    print "hallo"

注意我已经移动了 time

我试图说明的关键点是,任何(不被忽略的)信号都会中断Python代码执行的主线。将使用参数调用处理程序,这些参数指示触发执行的信号号(允许一个Python函数用于处理许多不同的信号)和一个frame对象(可用于某种调试或检测)。

厄尔诺 通过循环返回以重试或分支到主线中的某个备选方案(例如继续执行某个其他文件,或不执行任何文件/输入等)。

正如其他人在回答您的问题时所指出的,基于SIGALARM的方法可能会充满可移植性和可靠性问题。更糟糕的是,其中一些问题可能是您在测试环境中永远不会遇到的竞争条件,并且可能只在极难再现的条件下发生。丑陋的细节往往是在重新进入的情况下——如果在执行信号处理程序期间发送信号,会发生什么情况?

我在一些脚本中使用过SIGALARM,在Linux下,这对我来说不是问题。我正在编写的代码适合这项任务。它也许能满足你的需要。

浏览一下你链接到的文档,我发现它们似乎没有提供任何可以用来直接限制阻塞行为的“超时”参数或设置。在“控制流图”下面的表格中,我看到他们特别指出 .run() 可以无限期执行或直到收到SIGINT。我也注意到 .start() 可以在应用程序中启动线程,并且似乎可以在这些线程运行时将控制权返回到Python代码行。(这似乎取决于你的流程图的性质,我还没有完全理解)。

听起来你可以创建流图, .start() 然后(经过一段时间的处理或在Python代码的主线中睡眠之后)调用 .lock() .run() .wait() 打电话之后 ;和 显然会一直运行,直到所有块“指示它们已完成”或者直到调用对象的 .stop() 方法。

.start() 也没有 也不是 ;然后打电话 在进行任何其他处理后(包括 time.sleep() ).

也许是简单的事情:

    tb = send_seq_2.top_block()
    tb.start()
    time.sleep(endtime - time.time())
    tb.stop()
    seq1_sent = True
    tb = send_seq_2.top_block()
    tb.start()
    seq2_sent = True

.. 尽管我怀疑 时间。睡眠() 在那里。也许你想做一些其他的事情 tb 对象的状态(可能需要较小的睡眠间隔,调用 .lock() 方法,访问我一无所知的属性,然后调用 .unlock()

if not seq1_sent:
        tb = send_seq_2.top_block()
        tb.Run(True)
        seq1_sent = True
        if time.time() < endtime:
            break

如果“If time.time()<endtime:”,那么您将跳出循环,seq2发送的内容将永远不会被命中,也许您在该测试中是指“time.time()>endtime”?

您可以尝试使用延迟执行。。。Twisted框架经常使用它们

http://www6.uniovi.es/python/pycon/papers/deferex/

另一个问题的答案如下: python: how to send packets in multi thread and then the thread kill itself

或者google for killable python threads获取更多类似的细节: http://code.activestate.com/recipes/496960-thread2-killable-threads/

如果要设置阻塞函数的超时,则将threading.Thread作为方法连接(timeout),它将阻塞到超时。

基本上,像这样的事情应该做你想做的:

import threading
my_thread = threading.Thread(target=send_seq_2.top_block)
my_thread.start()
my_thread.join(TIMEOUT)