Многопроцессорный API

Question

num_folds = 3

def callModelScore(model, datax, datay, scoringType, folds):
    rating = model_selection.cross_val_score(gnb,
        X, y, scoring=scoringType, cv=num_folds)
    meanRating = str(round(100*rating.mean(), 2))
    print(scoringType + " " + meanRating + "%")
    return rating

from multiprocessing import Process

p1 = Process(target =  callModelScore,args =(gnb, X, y, 'accuracy', num_folds,))
p2 = Process(target =  callModelScore,args =(gnb, X, y, 'accuracy', num_folds,))
p3 = Process(target =  callModelScore,args =(gnb, X, y, 'accuracy', num_folds,))
p4 = Process(target =  callModelScore,args =(gnb, X, y, 'accuracy', num_folds,))
p1.start()
p2.start()
p3.start()
p4.start()
p1.join()
p2.join()
p3.join()
p4.join()

---

Ошибка:

BrokenPipeError                           Traceback (most recent call last)
<ipython-input-22-0aaad613b937> in <module>
      5 p3 = Process(target =  callModelScore,args =(gnb, X, y, 'accuracy', num_folds,))
      6 p4 = Process(target =  callModelScore,args =(gnb, X, y, 'accuracy', num_folds,))
----> 7 p1.start()
      8 p2.start()
      9 p3.start()

~\Anaconda3\lib\multiprocessing\process.py in start(self)
    110                'daemonic processes are not allowed to have children'
    111         _cleanup()
--> 112         self._popen = self._Popen(self)
    113         self._sentinel = self._popen.sentinel
    114         # Avoid a refcycle if the target function holds an indirect

~\Anaconda3\lib\multiprocessing\context.py in _Popen(process_obj)
    221     @staticmethod
    222     def _Popen(process_obj):
--> 223         return _default_context.get_context().Process._Popen(process_obj)
    224
    225 class DefaultContext(BaseContext):

~\Anaconda3\lib\multiprocessing\context.py in _Popen(process_obj)
    320         def _Popen(process_obj):
    321             from .popen_spawn_win32 import Popen
--> 322             return Popen(process_obj)
    323
    324     class SpawnContext(BaseContext):

~\Anaconda3\lib\multiprocessing\popen_spawn_win32.py in __init__(self, process_obj)
     87             try:
     88                 reduction.dump(prep_data, to_child)
---> 89                 reduction.dump(process_obj, to_child)
     90             finally:
     91                 set_spawning_popen(None)

~\Anaconda3\lib\multiprocessing\reduction.py in dump(obj, file, protocol)
     58 def dump(obj, file, protocol=None):
     59     '''Replacement for pickle.dump() using ForkingPickler.'''
---> 60     ForkingPickler(file, protocol).dump(obj)
     61
     62 #

BrokenPipeError: [Errno 32] Broken pipe

Я использовал python API для многопроцессорной обработки, чтобы протестировать API, и я не могу заставить что-либо работать. Я даже попробовал некоторые тестовые коды в документации для python 3. Я использую ноутбук Jupyter с Anaconda.

Answer 1

Я не эксперт в этом вопросе, но я считаю, что суть в том, что Jupyter Notebooks уже использует pickle для сериализации данных. Это означает, что существует неопределенность между потоками данных дочерних процессов и основным процессом, который их сгенерировал. К счастью, существует разветвленный проект многопроцессорной обработки, который, похоже, активно поддерживается. См. multiprocess API (обратите внимание, он отличается от «ing»). Этот API использует укроп вместо маринада. Что, помимо того, чтобы быть веселым, должно позволить вашей оболочке различать детей и основной процесс. Это также исключено в документации :

Функциональность этого пакета требует, чтобы модуль main был импортирован детьми. Это описано в Руководстве по программированию, однако, здесь стоит указать. Это означает, что некоторые примеры, такие как примеры multiprocessing.pool.Pool, не будут работать в интерактивном интерпретаторе.