在Python中有一种简单的方法可以等到某些条件成立吗？

Question

我需要在脚本中等待,直到一定数量的条件成为现实？

我知道我可以使用条件变量和朋友滚动自己的事件,但我不想经历实现它的所有麻烦,因为一些对象属性更改来自包装C++库(Boost.Python)中的外部线程,所以我不能只是__setattr__在一个类中劫持并在那里放置一个条件变量,这让我尝试从C++创建并发出Python条件变量信号,或者包装一个本地变量并在Python中等待,这两个声音都是繁琐,不必要的复杂和无聊.

是否有更简单的方法,除非连续轮询病情？

理想情况下,它将沿着这条线

res = wait_until(lambda: some_predicate, timeout)
if (not res):
    print 'timed out'

Answer 1

不幸的是,满足您的限制的唯一可能性是定期轮询,例如...:

import time

def wait_until(somepredicate, timeout, period=0.25, *args, **kwargs):
  mustend = time.time() + timeout
  while time.time() < mustend:
    if somepredicate(*args, **kwargs): return True
    time.sleep(period)
  return False

等等.如果somepredicate可以分解,则可以通过多种方式对其进行优化(例如,如果已知它and是多个子句中的一个,特别是如果某些条款反过来通过threading.Events或其他等可检测到的优化等),但是你要求的一般术语,这种低效的方法是唯一的出路.

Answer 2

另一个不错的包是waiting- https://pypi.org/project/waiting/

安装：

pip install waiting

用法：您传递一个每次都会被调用的函数作为条件、超时，并且（这很有用）您可以传递等待的描述，如果您得到 TimeoutError 将显示该描述。

使用功能：

from waiting import wait


def is_something_ready(something):
    if something.ready():
        return True
    return False


# wait for something to be ready
something = # whatever

wait(lambda: is_something_ready(something), timeout_seconds=120, waiting_for="something to be ready")

# this code will only execute after "something" is ready
print("Done")

注意：函数必须返回一个布尔值 - 当等待结束时为真，否则为假

Answer 3

这是另一个解决方案。目标是让线程在以非常精确的顺序执行某些工作之前相互等待。这项工作可能需要未知的时间。持续轮询不好有两个原因：它会占用 CPU 时间，并且在满足条件后不会立即开始操作。

class Waiter():

    def __init__(self, init_value):
        self.var = init_value
        self.var_mutex = threading.Lock()
        self.var_event = threading.Event()

    def WaitUntil(self, v):
        while True:
            self.var_mutex.acquire()
            if self.var == v:
                self.var_mutex.release()
                return # Done waiting
            self.var_mutex.release()
            self.var_event.wait(1) # Wait 1 sec

    def Set(self, v):
        self.var_mutex.acquire()
        self.var = v
        self.var_mutex.release()
        self.var_event.set() # In case someone is waiting
        self.var_event.clear()

以及测试方法

class TestWaiter():

    def __init__(self):
        self.waiter = Waiter(0)
        threading.Thread(name='Thread0', target=self.Thread0).start()
        threading.Thread(name='Thread1', target=self.Thread1).start()
        threading.Thread(name='Thread2', target=self.Thread2).start()

    def Thread0(self):
        while True:
            self.waiter.WaitUntil(0)
            # Do some work
            time.sleep(np.random.rand()*2)
            self.waiter.Set(1)

    def Thread1(self):
        while True:
            self.waiter.WaitUntil(1)
            # Do some work
            time.sleep(np.random.rand())
            self.waiter.Set(2)

    def Thread2(self):
        while True:
            self.waiter.WaitUntil(2)
            # Do some work
            time.sleep(np.random.rand()/10)
            self.waiter.Set(0)

多处理服务员：

import multiprocessing as mp
import ctypes

class WaiterMP():
    def __init__(self, init_value, stop_value=-1):
        self.var = mp.Value(ctypes.c_int, init_value)
        self.stop_value = stop_value
        self.event = mp.Event()

    def Terminate(self):
        self.Set(self.stop_value)

    def Restart(self):
        self.var.value = self.init_value

    def WaitUntil(self, v):
        while True:
            if self.var.value == v or self.var.value == self.stop_value:
                return
            # Wait 1 sec and check aiagn (in case event was missed)
            self.event.wait(1)

    def Set(self, v):
        exit = self.var.value == self.stop_value
        if not exit: # Do not set var if threads are exiting
            self.var.value = v
        self.event.set() # In case someone is waiting
        self.event.clear()

如果这仍然不是最佳解决方案，请发表评论。