issue5738: The distribution example was confusing, and out of date. It's too large to include inline in the docs as well. It belongs in an addons module outside the stdlib. Removing.
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index ea27991..67ac49c 100644
--- a/Lib/multiprocessing/queues.py
+++ b/Lib/multiprocessing/queues.py
@@ -47,6 +47,8 @@
if sys.platform != 'win32':
register_after_fork(self, Queue._after_fork)
+ self.getv = 0
+
def __getstate__(self):
assert_spawning(self)
return (self._maxsize, self._reader, self._writer,
@@ -71,6 +73,8 @@
self._poll = self._reader.poll
def put(self, obj, block=True, timeout=None):
+ if not isinstance(obj, list):
+ debug('put: %s', obj)
assert not self._closed
if not self._sem.acquire(block, timeout):
raise Full
@@ -85,11 +89,15 @@
self._notempty.release()
def get(self, block=True, timeout=None):
+ self.getv += 1
+ debug('self.getv: %s', self.getv)
if block and timeout is None:
self._rlock.acquire()
try:
res = self._recv()
self._sem.release()
+ if not isinstance(res, list):
+ debug('get: %s', res)
return res
finally:
self._rlock.release()
@@ -104,6 +112,8 @@
raise Empty
res = self._recv()
self._sem.release()
+ if not isinstance(res, list):
+ debug('get: %s', res)
return res
finally:
self._rlock.release()
@@ -229,16 +239,22 @@
try:
while 1:
obj = bpopleft()
+ if not isinstance(obj, list):
+ debug('feeder thread got: %s', obj)
if obj is sentinel:
debug('feeder thread got sentinel -- exiting')
close()
return
-
if wacquire is None:
+ if not isinstance(obj, list):
+ debug('sending to pipe: %s', obj)
send(obj)
else:
- wacquire()
+ debug('waiting on wacquire')
+ wacquire(timeout=30)
try:
+ if not isinstance(obj, list):
+ debug('sending to pipe: %s', obj)
send(obj)
finally:
wrelease()