Get multiprocessing.Queue to do priorities

[uuid]

Hello,
I was wondering whether there was a way to make multiprocessing.Queue
behave in a priority queue-like fashion. Subclassing with heappush and
heappop for put and get doesn't work the old way (multiprocessing.Queue
seems to use different data structures than Queue.Queue?)

Could one create a heapq within the producer as a proxy, and then feed
a proper queue from that? Does anyone have an idea on how to deal with
the queue item flow control? (Since all of the sorting has to happen
within the heapq, it should only pass items to the real Queue if it's
empty?)

Thanks in advance!

 

[uuid]

I just read up and it seems that no matter the approach, it's futile to
use multiprocessing.Queue, since there is a bug that prevents true
FIFO. (http://bugs.python.org/issue4999)

Any recommendation on an alternate way to build a priority queue to use
with a "one producer, many consumers" type multiprocessing setup would
be welcomed!

:|

 

[uuid]

The Queue module, apparently, is thread safe, but *not* process safe.
If you try to use an ordinary Queue, it appears inaccessible to the
worker process. (Which, after all, is quite logical, since methods for
moving items between the threads of the same process are quite
different from inter-process communication.) It appears that creating a
manager that holds a shared queue might be an option
(http://stackoverflow.com/questions/342556/python-2-6-multiprocessing-queue-compatible-with-threads).

Just

for illustration: This shows that Queue.Queue doesn't work with processes:

------------------------
def worker(queue):
while True:
item = queue.get()
print item
queue.task_done()

queue_queue = Queue.Queue()

worker_thread = multiprocessing.Process(target=worker, args=(queue_queue,))
worker_thread.start()
for i in range(10):
queue_queue.put(str(i))
time.sleep(10)
while True:
try:
print 'still on queue: ' + queue_queue.get(False)
except Queue.Empty:
break
worker_thread.join()
------------------------

This yields:

still on queue: 0
still on queue: 1
still on queue: 2
still on queue: 3
still on queue: 4
still on queue: 5
still on queue: 6
still on queue: 7
still on queue: 8
still on queue: 9

So no queue item ever arrives at the worker process.

 

[Jesse Noller]

The Queue module, apparently, is thread safe, but *not* process safe. If you
try to use an ordinary Queue, it appears inaccessible to the worker process.
(Which, after all, is quite logical, since methods for moving items between
the threads of the same process are quite different from inter-process
communication.) It appears that creating a manager that holds a shared queue
might be an option
(http://stackoverflow.com/questions/342556/python-2-6-multiprocessing-queue-compatible-with-threads).

Using a manager, or submitting a patch which adds priority queue to
the multiprocessing.queue module is the correct solution for this.

You can file an enhancement in the tracker, and assign/add me to it,
but without a patch it may take me a bit (wicked busy right now).

jesse

 

[uuid]

Dear Jesse,
thanks for the hint.

I see you are already assigned to the FIFO bug
(http://bugs.python.org/issue4999), so I won't burden you even more.
Clearly, a reliable FIFO behavior of multiprocessing.Queue helps more
than a priority queue, since it can be used to build one, so that
should really be the first thing to fix.

In the meantime, I think I'll whip up a hack that uses sort of a
bucket- strategy: fill up a prioritized heapq, and then, in regular
intervals, unload its contents into a size-limited multiprocessing
queue.

I'll post this as soon as it works.

-u

 

[madeleine.udell]

Was this issue ever resolved? What is the current best practice for those wishing to use a priority queue with multiprocessing?

 

[madeleine.udell]

Was this issue ever resolved? What is the current best practice for those wishing to use a priority queue with multiprocessing?