Python iterable Cola

Question

Necesito saber cuándo se cierra una cola y no tengo más elementos para poder finalizar la iteración.

lo hice poniendo un centinela en la cola:

from Queue import Queue 

class IterableQueue(Queue): 

    _sentinel = object() 

    def __iter__(self): 
     return self 

    def close(self): 
     self.put(self._sentinel) 

    def next(self): 
     item = self.get() 
     if item is self._sentinel: 
      raise StopIteration 
     else: 
      return item 

Teniendo en cuenta que este es un uso muy común para una cola, no hay ninguna implementación incorporada?

Answer 1

Un centinela es una forma razonable para un productor para enviar un mensaje de que no hay más tareas de cola son próximas.

Fwiw, su código se puede simplificar un poco con la forma de dos argumentos iter():

from Queue import Queue 

class IterableQueue(Queue): 

    _sentinel = object() 

    def __iter__(self): 
     return iter(self.get, self._sentinel) 

    def close(self): 
     self.put(self._sentinel) 

Answer 2

El módulo de multiprocesamiento tiene su propia versión de Queue que incluye un método close. No estoy seguro de cómo funciona en el enhebrado, pero vale la pena intentarlo. No veo por qué no debería funcionar de la misma:

from multiprocessing import Queue 

q = Queue() 
q.put(1) 
q.get_nowait() 
# 1 
q.close() 
q.get_nowait() 
# ... 
# IOError: handle out of range in select() 

Se podía coger el IOError como la señal de cierre.

PRUEBA

from multiprocessing import Queue 
from threading import Thread 

def worker(q): 
    while True: 
     try: 
      item = q.get(timeout=.5) 
     except IOError: 
      print "Queue closed. Exiting thread." 
      return 
     except: 
      continue 
     print "Got item:", item 

q = Queue() 
for i in xrange(3): 
    q.put(i) 
t = Thread(target=worker, args=(q,)) 
t.start() 
# Got item: 0 
# Got item: 1 
# Got item: 2 
q.close() 
# Queue closed. Exiting thread. 

aunque para ser honestos, no es demasiado muy diferente que el establecimiento de una bandera en el Queue.Queue. El multiprocessing.Queue es sólo con un descriptor de archivo cerrado como bandera:

from Queue import Queue 

def worker2(q): 
    while True: 
     if q.closed: 
      print "Queue closed. Exiting thread." 
      return 
     try: 
      item = q.get(timeout=.5) 
     except: 
      continue 
     print "Got item:", item 

q = Queue() 
q.closed = False 
for i in xrange(3): 
    q.put(i) 
t = Thread(target=worker2, args=(q,)) 
t.start() 
# Got item: 0 
# Got item: 1 
# Got item: 2 
q.closed = True 
# Queue closed. Exiting thread.