回答此问题可获得 20 贡献值,回答如果被采纳可获得 50 分。
<p>为了在windows上的交互式python(miniconda)中使用<code>multiprocessing</code>,我发现了一个非常有用的代码。但是,代码不能将类中的<code>self</code>参数传递给要合并的函数。以下是我在google colab上工作但在windows iPython上从未完成的代码:</p>
<pre><code>import multiprocessing
from multiprocessing import Pool
from poolable import make_applicable, make_mappable
def worker(d):
"""worker function"""
for i in range(10000):
j = i **(1/3) + d.bias
return j
class dummy():
def __init__(self):
self.bias = 1000
def calc(self):
pool = Pool(processes=12)
results = {}
for i in range(5):
results[i] = (pool.apply_async(*make_applicable(worker,self)))
pool.close()
pool.join()
print([results[i].get() for i in range(5)])
d=dummy()
d.calc()
</code></pre>
<p>如果我传递其他类型的变量,则该代码在windows上运行良好,例如:</p>
<pre><code>results[i] = (pool.apply_async(*make_applicable(worker,self.bias)))
</code></pre>
<p>但是当我将<code>self</code>传递给函数时,这个过程永远不会结束。我不知道该怎么办。你知道吗</p>
<p><code>poolable.py</code>来自<a href="https://stackoverflow.com/a/37140995/7490554">here</a>:</p>
<pre><code>from types import FunctionType
import marshal
def _applicable(*args, **kwargs):
name = kwargs['__pw_name']
code = marshal.loads(kwargs['__pw_code'])
gbls = globals() #gbls = marshal.loads(kwargs['__pw_gbls'])
defs = marshal.loads(kwargs['__pw_defs'])
clsr = marshal.loads(kwargs['__pw_clsr'])
fdct = marshal.loads(kwargs['__pw_fdct'])
func = FunctionType(code, gbls, name, defs, clsr)
func.fdct = fdct
del kwargs['__pw_name']
del kwargs['__pw_code']
del kwargs['__pw_defs']
del kwargs['__pw_clsr']
del kwargs['__pw_fdct']
return func(*args, **kwargs)
def make_applicable(f, *args, **kwargs):
if not isinstance(f, FunctionType): raise ValueError('argument must be a function')
kwargs['__pw_name'] = f.__name__ # edited
kwargs['__pw_code'] = marshal.dumps(f.__code__) # edited
kwargs['__pw_defs'] = marshal.dumps(f.__defaults__) # edited
kwargs['__pw_clsr'] = marshal.dumps(f.__closure__) # edited
kwargs['__pw_fdct'] = marshal.dumps(f.__dict__) # edited
return _applicable, args, kwargs
def _mappable(x):
x,name,code,defs,clsr,fdct = x
code = marshal.loads(code)
gbls = globals() #gbls = marshal.loads(gbls)
defs = marshal.loads(defs)
clsr = marshal.loads(clsr)
fdct = marshal.loads(fdct)
func = FunctionType(code, gbls, name, defs, clsr)
func.fdct = fdct
return func(x)
def make_mappable(f, iterable):
if not isinstance(f, FunctionType): raise ValueError('argument must be a function')
name = f.__name__ # edited
code = marshal.dumps(f.__code__) # edited
defs = marshal.dumps(f.__defaults__) # edited
clsr = marshal.dumps(f.__closure__) # edited
fdct = marshal.dumps(f.__dict__) # edited
return _mappable, ((i,name,code,defs,clsr,fdct) for i in iterable)
</code></pre>
<p><strong>编辑:</strong></p>
<p>似乎这个问题不仅存在于<code>self</code>,而且也存在于传递给<code>make_applicable</code>函数的任何其他类。以下代码也未完成:</p>
<pre><code>class dummy2():
def __init__(self):
self.bias = 1000
class dummy():
def __init__(self):
self.bias = 1000
def copy(self):
return copy.deepcopy(self)
def calc(self):
pool = Pool(processes=12)
results = {}
for i in range(5):
d = dummy2()
results[i] = pool.apply_async(*make_applicable(worker,d))
pool.close()
pool.join()
print([results[i].get() for i in range(5)])
</code></pre>