如何使用外部模块在joblib中的线程之间共享变量答案

【问题标题】：How to share a variable among threads in joblib using external module如何使用外部模块在joblib中的线程之间共享变量
【发布时间】：2019-06-19 12:02:17
【问题描述】：

我正在尝试修改 sklearn 源代码。特别是，我正在修改 GridSearch 源代码，以使评估不同模型配置的单独进程/线程在它们之间共享一个变量。我需要每个线程/进程在运行时读取/更新该变量，以便根据其他线程获得的内容修改它们的执行。更具体地说，我想分享的参数是 best，在下面的 sn-p 中：

out = parallel(delayed(_fit_and_score)(clone(base_estimator), X, y, best, self.early,train=train, test=test,parameters=parameters,**fit_and_score_kwargs) for parameters, (train, test) in product(candidate_params, cv.split(X, y, groups)))

注意 _fit_and_score 函数位于单独的模块中。 Sklearn 利用 joblib 进行并行化，但我无法理解如何使用外部模块有效地做到这一点。在 joblib doc 中提供了此代码：

>>> shared_set = set()
>>> def collect(x):
...    shared_set.add(x)
...
>>> Parallel(n_jobs=2, require='sharedmem')(
...     delayed(collect)(i) for i in range(5))
[None, None, None, None, None]
>>> sorted(shared_set)
[0, 1, 2, 3, 4]

但我无法理解如何让它在我的上下文中运行。你可以在这里找到源代码：

【问题讨论】：

标签： scikit-learn joblib

【解决方案1】：

你可以用python的Manager（https://docs.python.org/3/library/multiprocessing.html#multiprocessing.sharedctypes.multiprocessing.Manager）来做，简单的代码例如：

from joblib import Parallel, delayed
from multiprocessing import Manager

manager = Manager()
q = manager.Namespace()
q.flag = False

def test(i, q):
    #update shared var in 0 process
    if i == 0:
        q.flag = True

    # do nothing for few seconds
    for n in range(100000000):
        if q.flag == True:
            return f'process {i} was updated'

    return 'process {i} was not updated'

out = Parallel(n_jobs=4)(delayed(test)(i, q) for i in range(4))

out：

['process 0 was updated',
 'process 1 was updated',
 'process 2 was updated',
 'process 3 was updated']

【讨论】：