Как поддерживать глобальные процессы в пуле, работающие рекурсивно?

Вы не можете присоединиться к пулу, который не был закрыт первым, так как join() будет ждать завершения рабочих процессов, а не выполнения заданий (https://docs.python.org/3.6/library/multiprocessing.html раздел 17.2.2.9).

Но поскольку это закроет пул, а это не то, что вам нужно, вы не можете использовать это. Таким образом, соединение отсутствует, и вам нужно реализовать «ожидание завершения всех заданий» самостоятельно.

Один из способов сделать это без циклов занятости — использовать очередь. Вы также можете работать с ограниченными семафорами, но они работают не во всех операционных системах.

counter = 0
lock_queue = multiprocessing.Queue()
counter_lock = multiprocessing.Lock()

def log_result(result):

    tempx , tempb, u = result

    X[:,u,np.newaxis], b[:,u,np.newaxis] = tempx , tempb
    with counter_lock:
        counter += 1
        if counter == 4:
            counter = 0
            lock_queue.put(42)



workers =  mp.Pool(processes = 4) 
for t in range(p,T):

    count = 0 #==========This is only master's job=============
    for l in range(p):
        for k in range(4):
            gn[count]=train[t-l-1,k]
            count+=1
    G = G*v +  gn @ gn.T#==================================

    if __name__ == '__main__':
        counter = 0
        for i in range(4):
            workers.apply_async(OULtraining, args=(train[t,i], X[:,i,np.newaxis], b[:,i,np.newaxis], i, gn), callback = log_result)


        lock_queue.get(block=True)

Это сбрасывает глобальный счетчик перед отправкой заданий. Как только задание завершено, ваш обратный вызов увеличивает глобальный счетчик. Когда счетчик достигает 4 (ваше количество заданий), обратный вызов знает, что он обработал последний результат. Затем фиктивное сообщение отправляется в очередь. Ваша основная программа ожидает в Queue.get() появления чего-то там.

Это позволяет вашей основной программе блокироваться до завершения всех заданий, не закрывая пул.

Если вы замените multiprocessing.Pool на ProcessPoolExecutor из concurrent.futures, вы можете пропустить эту часть и использовать

concurrent.futures.wait(fs, timeout=None, return_when=ALL_COMPLETED)

для блокировки, пока все отправленные задачи не будут завершены. С функциональной точки зрения между ними нет никакой разницы. Метод concurrent.futures короче на пару строк, но результат точно такой же.