在 Python 中处理阻塞函数调用

Question

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

我导入了 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"

How do I still get print "hallo". ;）

谢谢， 马吕斯

Answer 1

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

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 2

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

在主线程中，您需要等待两种情况：a）top_block 完成，b）超时到期。繁忙等待是邪恶的:-)，因此您应该使用线程的 join-with-timeout 来等待线程。如果join后线程仍然存活，则需要停止top_run。

Answer 3

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

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 4

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_sent 的东西将永远不会被命中，也许你的意思是 'time.time() >那个测试中的“末日”？

Answer 5

您问题的简单部分与信号处理有关。从 Python 运行时的角度来看，解释器进行系统调用时收到的信号会作为 OSError 异常呈现给您的 Python 代码，并带有对应于 errno.EINTR 的 errno 属性。

因此，这可能大致按照您的预期工作：

    #!/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()< /code> 控制对象上的方法（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() 。

Answer 6

你可以尝试使用延迟执行... Twisted 框架大量使用它们

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

Answer 7

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

请参阅另一个问题的答案：
python：如何在多线程中发送数据包，然后线程杀死自己

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

Answer 8

如果要在阻塞函数上设置超时，可以使用 threading.Thread 作为 join(timeout) 方法，该方法会阻塞直到超时。

基本上，类似的事情应该做你想做的：

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