Combining PriorityQueue in python asyncio
There are several queue implementations as per the documentation . I'm interested in JoinableQueue and PriorityQueue as I would like to have a priority queue.
It looks like I can only get one of the features in lower versions. That is, in 3.5 I can do a distinciton with a Queue (which is compatible) and a PriorityQueue, but in python below 3.5 there is a JoinableQueue and a PriorityQueue (see example below).
Is it possible to combine them to get a general way to get an attachable PriorityQueue in 3.4?
try:
# Python 3.4.
from asyncio import JoinableQueue as Queue # joinable
from asyncio import PriorityQueue # I assume this one is not joinable
except ImportError:
# Python 3.5.
from asyncio import Queue # standard joinable
from asyncio import PriorityQueue # I assume this is the one I want
Another approach is to influence the queue in some way?
source to share
Due to the fact that the implemented methods JoinableQueue
, and PriorityQueue
you can get JoinablePriorityQueue
by subclass using multiple inheritance, if you first list JoinableQueue
.
The reason for this is that it is PriorityQueue
very simple to implement:
class PriorityQueue(Queue):
"""A subclass of Queue; retrieves entries in priority order (lowest first).
Entries are typically tuples of the form: (priority number, data).
"""
def _init(self, maxsize):
self._queue = []
def _put(self, item, heappush=heapq.heappush):
heappush(self._queue, item)
def _get(self, heappop=heapq.heappop):
return heappop(self._queue)
While JoinableQueue
more complex, the only method it both PriorityQueue
implements _put
and, crucially, JoinableQueue
calls super()._put(..)
in its own implementation put
, which means it will cooperate with PriorityQueue
.
Here's an example to demonstrate that it works:
from asyncio import PriorityQueue, JoinableQueue
import asyncio
import random
class JoinablePriorityQueue(JoinableQueue, PriorityQueue):
pass
@asyncio.coroutine
def consume(q):
while True:
a = yield from q.get()
print("got a {}".format(a))
if a[1] is None:
q.task_done()
return
asyncio.sleep(1)
q.task_done()
@asyncio.coroutine
def produce(q):
for i in range(10):
yield from q.put((random.randint(0,10), i))
yield from q.put((100, None)) # Will be last
asyncio.async(consume(q))
print("waiting...")
yield from q.join()
print("waited")
loop = asyncio.get_event_loop()
q = JoinablePriorityQueue()
loop.run_until_complete(produce(q))
Output:
waiting...
got a (1, 2)
got a (2, 1)
got a (4, 4)
got a (5, 0)
got a (6, 8)
got a (6, 9)
got a (8, 3)
got a (9, 5)
got a (9, 7)
got a (10, 6)
got a (100, None)
waited
source to share