14

I'm trying to write a Python 2.6 (OSX) program using multiprocessing, and I want to populate a Queue with more than the default of 32767 items.

from multiprocessing import Queue
Queue(2**15) # raises OSError

Queue(32767) works fine, but any higher number (e.g. Queue(32768)) fails with OSError: [Errno 22] Invalid argument

Is there a workaround for this issue?

Jason Sundram
  • 12,225
  • 19
  • 71
  • 86
  • What kind of data does the queue contain? Are you sure that any higher number fails, or could it be that the 32768th data node causes the error? (Are you using path names by chance?) – voithos May 05 '11 at 16:55
  • @voithos, I haven't populated the queue before it explodes. Just setting max size causes the OSError. – Jason Sundram May 05 '11 at 17:38

2 Answers2

4

One approach would be to wrap your multiprocessing.Queue with a custom class (just on the producer side, or transparently from the consumer perspective). Using that you would queue up items to be dispatched to the Queue object that you're wrapping, and only feed things from the local queue (Python list() object) into the multiprocess.Queue as space becomes available, with exception handling to throttle when the Queue is full.

That's probably the easiest approach since it should have the minimum impact on the rest of your code. The custom class should behave just like a Queue while hiding the underlying multiprocessing.Queue behind your abstraction.

(One approach might be to have your producer use threads, one thread to manage the dispatch from a threading Queue to your multiprocessing.Queue and any other threads actually just feeding the threading Queue).

Jim Dennis
  • 17,054
  • 13
  • 68
  • 116
  • 1
    The problem with that approach is that (on OS X in any case), there's no way to tell how many items are in a queue. Calling the `.qsize` method raises `NotImplementedError`, with the explanation "Raises NotImplementedError on Mac OSX because of broken sem_getvalue()". Similarly the `.full` method is totally unreliable on OS X. So unless I'm mistaken there's no reliable way to get the information necessary to implement the wrapper class you describe... – Peter M Dec 20 '11 at 23:58
  • @Peter McMahan: does the Multiprocessing.Queue raise a reasonable exception when you attempt to .put() something into it while its full? Note that I did suggest "with exception handling." – Jim Dennis Dec 21 '11 at 08:49
  • After a couple minutes with the default Python installation (version 2.7.1) on my MacOS X (Lion 10.7.2) I see that it, too, has a limit of 32767 on its multiprocess.Queue() capacity. .empty() seems to work as does .full() .... but there are bugs in this implementation. Defining a simple "while not q.empty(): q.get()" loop often shows the Queue as empty even when you've already pushed thousands of objects into it. I should report these (but won't promise to get to it so please feel free if you're so inclined). – Jim Dennis Dec 21 '11 at 08:57
  • Yes, when I said that `.full` is unreliable that's what I meant: it seems like there is a delay before a `Queue` will recognize that it is empty or full. It does throw sensible errors (`mutiprocessing.queues.Full` and `multiprocessing.queues.Empty`) but at least the `Empty` exception seems to have the same problem as the `.empty()` method. In my tests so far, though, calling `q.put(i,block=False)` *seems* to reliably through the `mutiprocessing.queues.Full` exception, so a wrapper class may actually work... – Peter M Dec 21 '11 at 20:26
2

I've already answered the original question but I do feel like adding that Redis lists are quite reliable and the Python module's support for them are extremely easy to use for implementing a Queue like object. These have the advantage of allowing one to scale out over multiple nodes (across a network) as well as just over multiple processes.

Basically to use those you'd just pick a key (string) for your queue name have your producers push into it and have your workers (task consumers) loop on blocking pops from that key.

The Redis BLPOP, and BRPOP commands all take a list of keys (lists/queues) and an optional timeout value. They return a tuple (key,value) or None (on timeout). So you can easily write up an event driven system that's very similar to the familiar structure of select() (but at a much higher level). The only thing you have to watch for are missing keys and invalid key types (just wrap your queue operations with exception handlers, of course). (If some other application stops on your shared Redis server removing keys or replacing keys that you were using as queues with string/integer or other types of values ... well, you have a different problem at that point). :)

Another advantage of this model is that Redis does persist its data to the disk. So your work queue could survive system restarts if you chose to allow it.

(Of course you could implement a simple Queue as a table in SQLlite or any other SQL system if you really wanted to do so; just using some sort of auto-incrementing index for the sequencing and a column to mark each item has having been "done" (consumed); but that does involve somewhat more complexity than using what Redis gives you "out of the box").

Jim Dennis
  • 17,054
  • 13
  • 68
  • 116