Добавить аргумент таймаута в python Queue.join()

Я хочу, чтобы иметь возможность присоединиться() к классу Queue, но время ожидания через некоторое время, если вызов еще не вернулся. Каков наилучший способ сделать это? Возможно ли это сделать путем подклассификации очереди\с помощью метакласса?

Ответ 1

Подкласс Queue, вероятно, лучший способ. Что-то вроде этого должно работать (непроверено):

def join_with_timeout(self, timeout):
    self.all_tasks_done.acquire()
    try:
        endtime = time() + timeout
        while self.unfinished_tasks:
            remaining = endtime - time()
            if remaining <= 0.0:
                raise NotFinished
            self.all_tasks_done.wait(remaining)
    finally:
        self.all_tasks_done.release()

Ответ 2

Метод join() - это ожидание выполнения всех задач. Если вам все равно, действительно ли задачи завершены, вы можете периодически проверять количество незавершенных задач:

stop = time() + timeout
while q.unfinished_tasks and time() < stop:
    sleep(1)

Этот цикл будет существовать либо при выполнении задач, либо по истечении периода ожидания.

Раймонда

Ответ 3

Сначала вы должны убедиться, что все ваши рабочие потоки в очереди выходят с task_done()

Чтобы реализовать функцию тайм-аута с помощью Queue, вы можете обернуть код очереди в потоке и добавить тайм-аут для этого потока с помощью Thread.join([timeout])

непроверенный пример, чтобы описать, что я предлагаю

def worker():
    while True:
        item = q.get()
        do_work(item)
        q.task_done()

def queuefunc():
    q = Queue()
    for i in range(num_worker_threads):
        t = Thread(target=worker)
        t.setDaemon(True)
        t.start()

    for item in source():
        q.put(item)

    q.join()       # block until all tasks are done

t = Thread(target=queuefunc)
t.start()
t.join(100) # timeout applies here