diff options
Diffstat (limited to 'Lib/multiprocessing/queues.py')
| -rw-r--r-- | Lib/multiprocessing/queues.py | 20 |
1 files changed, 18 insertions, 2 deletions
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index ea279911b6..67ac49cf61 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -47,6 +47,8 @@ class Queue(object): 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 @@ class Queue(object): 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 @@ class Queue(object): 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 @@ class Queue(object): 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 @@ class Queue(object): 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() |