Передача класса в многопроцессорный пул в Python на Windows

Question

Чтобы использовать multiprocessing в интерактивном Python на Windows (Miniconda), я нашел очень полезный код, который работает очень хорошо. Однако код не может передать аргумент self внутри класса функции для объединения в пул. Вот мой код, который работает на Google Colab, но никогда не заканчивается на Windows iPython:

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()

Код хорошо работает на Windows, если я передаю другие типы переменных, например:

results[i] = (pool.apply_async(*make_applicable(worker,self.bias)))

Но когда я передаю self функции, процесс никогда не заканчивается. Я понятия не имею, что делать.

poolable.py с здесь :

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)

Редактировать:

Этокажется, что проблема существует не только для self, но и для любых других классов, которые передаются функции make_applicable. Следующий код также не заканчивается:

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)])

Answer 1

Использование консоли IPython: поместите свой код в модуль (mp.py), обеспечив выполнение экземпляра класса и вызов метода в if __name__ == '__main__': условном

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
      self.bias = 10
  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()
    return [results[i].get() for i in range(5)]

if __name__ == '__main__':
    d=Dummy()
    print(d.calc())

Затем в консоли запустите модуль:

In [1]: runfile('P:/pyProjects3/mp.py', wdir='P:/pyProjects3')
[31.543628731482663, 31.543628731482663, 31.543628731482663, 31.543628731482663, 31.543628731482663]

---

У меня есть Jupyter Notebook (Anaconda), но я не знаю, как его использовать, если я это выясню, я обновлю этот ответ.

Answer 2

попробуйте использовать этот код для проверки

import pickle

def worker(d):
    """worker function"""
    d = pickle.loads(d)
    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):
          x = pickle.dumps(self)
          results[i] = (pool.apply_async(*make_applicable(worker,x)))
      pool.close()
      pool.join()
      print([results[i].get() for i in range(5)])