Handling slow data decomposition with Queues and Locks

[SeanInSeattle]

With this code:

#!/usr/bin/env python

from processing import Pool, Process, Manager, Lock
import sys

def slow_operation(i, q, lk):
print "id: ", i
# some really slow operation...
print "qcquiring lock..."
try:
lk.acquire(blocking=True)
q.put("result")
lk.release()
except:
print sys.exc_info()

mgr = Manager()
qlk = mgr.Lock()
slow_operation_result = mgr.Queue()
pool = Pool(10)
oid_range = [(i, slow_operation_result, qlk) for i in range(1,200)]
async_result = pool.map_async(slow_operation, oid_range)
async_result.wait()




I get a bunch of these errors:




Process PoolWorker-#: Traceback (most recent call last): File "/usr/
lib/python2.5/site-packages/processing/process.py", line 227, in
_bootstrap self.run() File "/usr/lib/python2.5/site-packages/
processing/process.py", line 85, in run self._target(*self._args,
**self._kwargs) File "/usr/lib/python2.5/site-packages/processing/
pool.py", line 54, in worker for job, i, func, args, kwds in
iter(inqueue.get, None): File "/usr/lib/python2.5/site-packages/
processing/queue.py", line 327, in get return recv() File "/usr/lib/
python2.5/site-packages/processing/managers.py", line 869, in
RebuildProxy incref=incref, **kwds) File "/usr/lib/python2.5/site-
packages/processing/managers.py", line 907, in MakeAutoProxy proxy =
ProxyType(token, manager=manager, authkey=authkey, incref=incref) File
"/usr/lib/python2.5/site-packages/processing/managers.py", line 718,
in __init__ self._incref() File "/usr/lib/python2.5/site-packages/
processing/managers.py", line 763, in _incref assert self._id not in
self._idset AssertionError: (AssertionError(), , (,
Token(typeid='Queue', address='/tmp/pyc-1705-0-PUnclX', id=25638024),
{'exposed': ('empty', 'full', 'get', 'get_nowait', 'join', 'put',
'put_nowait', 'qsize', 'task_done')}))
Interpreter version: Python 2.5.2 (what ships with debian stable)
Processing lib version: 0.52

Why does this fail?