Add timeout argument to python's Queue.join()

I want to be able to join() the Queue class but timeouting after some time if the call hasn't returned yet.开发者_如何学JAVA What is the best way to do it? Is it possible to do it by subclassing queue\using metaclass?

Subclassing Queue is probably the best way. Something like this should work (untested):

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()

The join() method is all about waiting for all the tasks to be done. If you don't care whether the tasks have actually finished, you can periodically poll the unfinished task count:

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

This loop will exist either when the tasks are done or when the timeout period has elapsed.

Raymond

At first, you should ensure that all your working threads in the queue exit with task_done()

To implement a timeout functionality with Queue, you can wrap the Queue's code in a Thread and add a timeout for this Thread using Thread.join([timeout])

untested example to outline what I suggest

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

As I tried to implement the accepted answer, it seems that all_tasks_done is not defined anymore. A quick solution is to use the timeout of the wait() function called in JoinableQueue.join.

For example overriding the join function in a subclass of JoinableQueue will add a 15s timeout on the waiting operation :

def join(self):
    with self._cond:
        if not self._unfinished_tasks._semlock._is_zero():
            self._cond.wait(15)