问题描述
multiprocessing.Pool
令我发疯...
我想升级许多软件包,对于每个软件包,我都必须检查是否有更大的版本.这是通过check_one
函数完成的.
主要代码在Updater.update
方法中:在这里创建Pool对象并调用map()
方法.
multiprocessing.Pool
is driving me crazy...
I want to upgrade many packages, and for every one of them I have to check whether there is a greater version or not. This is done by the check_one
function.
The main code is in the Updater.update
method: there I create the Pool object and call the map()
method.
这是代码:
def check_one(args):
res, total, package, version = args
i = res.qsize()
logger.info('\r[{0:.1%} - {1}, {2} / {3}]',
i / float(total), package, i, total, addn=False)
try:
json = PyPIJson(package).retrieve()
new_version = Version(json['info']['version'])
except Exception as e:
logger.error('Error: Failed to fetch data for {0} ({1})', package, e)
return
if new_version > version:
res.put_nowait((package, version, new_version, json))
class Updater(FileManager):
# __init__ and other methods...
def update(self):
logger.info('Searching for updates')
packages = Queue.Queue()
data = ((packages, self.set_len, dist.project_name, Version(dist.version)) \
for dist in self.working_set)
pool = multiprocessing.Pool()
pool.map(check_one, data)
pool.close()
pool.join()
while True:
try:
package, version, new_version, json = packages.get_nowait()
except Queue.Empty:
break
txt = 'A new release is avaiable for {0}: {1!s} (old {2}), update'.format(package,
new_version,
version)
u = logger.ask(txt, bool=('upgrade version', 'keep working version'), dont_ask=self.yes)
if u:
self.upgrade(package, json, new_version)
else:
logger.info('{0} has not been upgraded', package)
self._clean()
logger.success('Updating finished successfully')
当我运行它时,出现这个奇怪的错误:
When I run it I get this weird error:
Searching for updates
Exception in thread Thread-1:
Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 552, in __bootstrap_inner
self.run()
File "/usr/lib/python2.7/threading.py", line 505, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/local/lib/python2.7/dist-packages/multiprocessing/pool.py", line 225, in _handle_tasks
put(task)
PicklingError: Can't pickle <type 'thread.lock'>: attribute lookup thread.lock failed
推荐答案
多处理通过mp.SimpleQueue
将任务(包括check_one
和data
)传递给工作进程.与Queue.Queue
不同,放在mp.SimpleQueue
中的所有内容都必须是可挑选的. Queue.Queue
无法选择:
multiprocessing passes tasks (which include check_one
and data
) to the worker processes through a mp.SimpleQueue
. Unlike Queue.Queue
s, everything put in the mp.SimpleQueue
must be pickable. Queue.Queue
s are not pickable:
import multiprocessing as mp
import Queue
def foo(queue):
pass
pool=mp.Pool()
q=Queue.Queue()
pool.map(foo,(q,))
产生此异常:
UnpickleableError: Cannot pickle <type 'thread.lock'> objects
您的data
包括packages
,这是一个Queue.Queue.这可能是问题的根源.
Your data
includes packages
, which is a Queue.Queue. That might be the source of the problem.
这是一个可能的解决方法:Queue
被用于两个目的:
Here is a possible workaround: The Queue
is being used for two purposes:
- 找出近似大小(通过调用
qsize
) - 存储结果以供以后检索.
我们可以使用mp.Value
而不是调用qsize
在多个进程之间共享值.
Instead of calling qsize
, to share a value between multiple processes, we could use a mp.Value
.
我们可以(并且应该)仅将调用中的值返回到check_one
,而不是将结果存储在队列中. pool.map
将结果收集在自己的队列中,并将结果作为pool.map
的返回值返回.
Instead of storing results in a queue, we can (and should) just return values from calls to check_one
. The pool.map
collects the results in a queue of its own making, and returns the results as the return value of pool.map
.
例如:
import multiprocessing as mp
import Queue
import random
import logging
# logger=mp.log_to_stderr(logging.DEBUG)
logger = logging.getLogger(__name__)
qsize = mp.Value('i', 1)
def check_one(args):
total, package, version = args
i = qsize.value
logger.info('\r[{0:.1%} - {1}, {2} / {3}]'.format(
i / float(total), package, i, total))
new_version = random.randrange(0,100)
qsize.value += 1
if new_version > version:
return (package, version, new_version, None)
else:
return None
def update():
logger.info('Searching for updates')
set_len=10
data = ( (set_len, 'project-{0}'.format(i), random.randrange(0,100))
for i in range(set_len) )
pool = mp.Pool()
results = pool.map(check_one, data)
pool.close()
pool.join()
for result in results:
if result is None: continue
package, version, new_version, json = result
txt = 'A new release is avaiable for {0}: {1!s} (old {2}), update'.format(
package, new_version, version)
logger.info(txt)
logger.info('Updating finished successfully')
if __name__=='__main__':
logging.basicConfig(level=logging.DEBUG)
update()
这篇关于multiprocessing.Pool-PicklingError:无法酸洗< type'thread.lock'> ;:属性查找thread.lock失败的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!