在 Python 中处理阻塞函数调用

Question

我正在使用Gnuradio framework。我处理我为发送/接收信号而生成的流程图。这些流程图初始化并启动，但它们不会将控制流返回给我的应用程序：

我导入了time

while time.time() < endtime:
        # invoke GRC flowgraph for 1st sequence
        if not seq1_sent:
            tb = send_seq_2.top_block()
            tb.Run(True)
            seq1_sent = True
            if time.time() < endtime:
                break

        # invoke GRC flowgraph for 2nd sequence
        if not seq2_sent:
            tb = send_seq_2.top_block()
            tb.Run(True)
            seq2_sent = True
            if time.time() < endtime:
                break

问题是：只有第一个 if 语句调用流程图（与硬件交互）。我陷入了困境。我可以使用线程，但我不知道如何在 Python 中超时线程。我怀疑这是可能的，因为似乎杀死线程不在 API 中。该脚本只需要在 Linux 上运行...

如何使用 Python 正确处理阻塞函数 - 而不杀死整个程序。 这个问题的另一个更具体的例子是：

import signal, os

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


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

    # This open() may hang indefinitely
    fd = os.open('/dev/ttys0', os.O_RDWR)
    signal.alarm(0)          # Disable the alarm


foo()
print "hallo"

我如何仍然获得print "hallo"。 ;)

谢谢， 马吕斯

Answer 1

如果你想在一个阻塞函数上设置超时，threading.Thread 作为方法 join(timeout) 直到超时。

基本上，这样的事情应该做你想做的事：

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

Answer 2

您可以设置一个信号警报，以超时中断您的通话：

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. 接收信号通常会中断系统调用（然后解除对应用程序的阻塞）

Answer 3

IIUC，每个 top_block 都有一个 stop 方法。因此，您实际上可以在线程中运行 top_block，并在超时到达时发出停止。如果 top_block 的 wait() 也有超时就好了，可惜没有。

在主线程中，您需要等待两种情况：a）top_block 完成，b）超时到期。 Busy-waits 是邪恶的 :-)，所以你应该使用线程的 join-with-timeout 来等待线程。如果join后线程还活着，则需要停止top_run。

Answer 4

首先 - 应不惜一切代价避免使用信号：

1) 可能会导致死锁。 SIGALRM 可能在阻塞系统调用之前到达进程（想象系统中的超高负载！）并且系统调用不会被中断。死锁。

2) 使用信号可能会产生一些令人讨厌的非本地后果。例如，其他线程中的系统调用可能会被中断，这通常不是您想要的。通常，系统调用会在收到（不是致命的）信号时重新启动。当您设置信号处理程序时，它会自动关闭整个进程或线程组的这种行为。检查'man siginterrupt'。

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

在某些情况下可以显式避免阻塞 - 我强烈建议使用 select() 和朋友（检查 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 包装器。

当你不能明确地强制非阻塞行为时，比如说你只使用一个外部库，那么它会变得更加困难。线程可能会，但显然它不是最先进的解决方案，通常是错误的。

恐怕您通常无法以稳健的方式解决此问题 - 这实际上取决于您阻止的内容。

Answer 5

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

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

Answer 6

您提到在 Python 中杀死线程 - 这是部分可能的，尽管您只能在 Python 代码运行时杀死/中断另一个线程，而不是在 C 代码中，所以这可能无法如您所愿。

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

或 google 可杀死 python 线程以获取更多详细信息，如下所示： http://code.activestate.com/recipes/496960-thread2-killable-threads/

Answer 7

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' 在那个测试中？

Answer 8

您问题的简单部分与信号处理有关。从 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 函数用于处理许多不同的信号）和一个框架对象（可用于调试或某种类型的检测）。

由于代码的主要流程被中断，因此您必须将该代码包装在一些异常处理中，以便在此类事件发生后重新获得控制权。 （顺便说一句，如果您使用 C 编写代码，您也会有同样的担忧；您必须准备好使用底层系统调用返回错误并处理系统中的 -EINTR 的任何库函数errno 通过循环返回重试或分支到主行中的某个替代项（例如继续执行某个其他文件，或没有任何文件/输入等）。

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

我在一些脚本中使用了 SIGALARM，在 Linux 下这对我来说不是问题。我正在处理的代码适合这项任务。它可能足以满足您的需求。

如果不进一步了解此 Gnuradio 代码的行为方式、您从中实例化的对象类型以及它们返回的对象类型，则很难回答您的主要问题。

浏览您链接到的文档，我发现它们似乎没有提供任何可用于直接限制阻塞行为的“超时”参数或设置。在“控制流图”下的表格中，我看到他们明确表示.run() 可以无限期执行或直到收到 SIGINT。我还注意到.start() 可以在您的应用程序中启动线程，并且似乎在这些线程运行时将控制权返回给您的 Python 代码行。 （这似乎取决于您的流程图的性质，我不太了解）。

听起来你可以创建你的流程图，.start() 它们，然后（在你的 Python 代码主行中处理或休眠一段时间后）在你的控制对象上调用 .lock() 方法（tb？）。我猜，这会将状态的 Python 表示……Python 对象……置于静止模式，以允许您查询状态，或者如他们所说，重新配置您的流程图。如果你调用.run()，它会在调用.start()之后调用.wait()； .wait() 显然会一直运行，直到所有块“表明它们已完成”或直到您调用对象的 .stop() 方法。

所以听起来你想使用.start()，既不想使用.run()，也不想使用.wait()；然后在进行任何其他处理（包括time.sleep()）后调用.stop()。

也许很简单：

    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

.. 虽然我怀疑我的time.sleep() 在那里。也许您想做其他事情来查询tb 对象的状态（可能需要以更短的时间间隔休眠，调用它的.lock() 方法，并访问我一无所知的属性，然后在再次休眠之前调用它的.unlock()。