thread.lock во время класса поиска пользовательских параметров с использованием распределенного Dask

Я написал свою собственную реализацию поиска параметров в основном из-за того, что мне не нужна перекрестная проверка GridSearch и RandomizedSearch scikit-learn.

Я использую dask для обеспечения оптимальной распределенной производительности.

Вот что у меня есть:

from scipy.stats import uniform
class Params(object):
    def __init__(self,fixed,loc=0.0,scale=1.0):
        self.fixed=fixed
        self.sched=uniform(loc=loc,scale=scale)

    def _getsched(self,i,size):
        return self.sched.rvs(size=size,random_state=i)

    def param(self,i,size=None):
        tmp=self.fixed.copy()
        if size is None:
            size=tmp['niter']
        tmp.update({'schd':self._getsched(i,size)})
        return tmp    

class Mymodel(object):
    def __init__(self,func,params_object,score,ntries,client):
        self.params=params_object
        self.func=func
        self.score=score
        self.ntries=ntries
        self.client=client

    def _run(self,params,train,test):
        return self.func(params,train,test,self.score)

    def build(self,train,test):
        res=[]
        for i in range(self.ntries):
            cparam=self.params.param(i)
            res.append( (cparam, self.client.submit(self._run, cparam, train,test)) )
        self._results=res
        return res

    def compute_optimal(self,res=None):
        from operator import itemgetter
        if res is None:
            res=self._results
        self._sorted=sorted(self.client.compute(res),key=itemgetter(1))

        return self._sorted[0]


def score(test,correct):
    return np.linalg.norm(test-correct)

def myfunc(params,ldata,data,score):
    schd=params['schd']
    niter=len(schd)
    #here I do some magic after which ldata is changing
    return score(test=ldata,correct=data)

После запуска dask.distributed:

from distributed import Client
scheduler_host='myhostname:8786'
cli=Client(scheduler_host)

Я запускаю это так:

%%time
params=Params({'niter':50},loc=1.0e-06,scale=1.0)
model=Mymodel(myfunc,params,score,100,cli)
ptdata=bad_data_example.copy()
graph=model.build(ptdata,good_data)

И получите это:

distributed.protocol.pickle - INFO - Failed to serialize
<bound method Mymodel._run of <__main__.Mymodel object at 0x2b8961903050>>.
Exception: can't pickle thread.lock objects

Не могли бы вы помочь мне понять, что происходит и как это исправить?

Мне также любопытно, как я нахожу минимум среди результатов всех параметров. Есть ли лучший способ сделать это с помощью Dask?

Я написал этот код довольно быстро и никогда не пробовал его в последовательном режиме. Я изучаю Dask вместе со многими другими темами (машинное обучение, программирование gpu, Numba, Python OOP и т. д.), поэтому этот код ни в коем случае не оптимален...

P.S. Чтобы выполнить его, я использую этот вызов: model.compute_optimal(). Еще не добрался сюда - из-за ошибки выше.