Внезапно Python многопроцессорный пул apply_async комбинация закрытия соединения не работает - PullRequest
Новая
       8

Внезапно Python многопроцессорный пул apply_async комбинация закрытия соединения не работает

0 голосов
/ 15 декабря 2018

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

pool = multiprocessing.Pool(12)
pool.apply_async(func=process, args=(arg1_1,arg2_1))
pool.apply_async(func=process, args=(arg1_2,arg2_2))
pool.close()
pool.join()

Я попытался отладить его и обнаружил, что строка pool.apply_async работает, но pool.close() завершит все.

Ниже приведены все моикод: 

tasks = ['Task1', 'Task2', 
'Task3', 'Task4', 
'Task5', 'Task6']
for task in tqdm(tasks):
    print(task)
    task_archive = tasks_archive[task] # task_archive is dict with keys like: 'images_Train', paths to image files for training; 'labels_Train', paths to ground truth files for training.

    def prep(ids, outDir):
        print("ids[0]:{}, current time:{}".format(ids[0], str(tinies.datestr())))
        for ID in ids:
            # ID=train_ids[0]
            img_path = os.path.join(config.BASEDIR, task, 'imagesTr', ID)
            lab_path = os.path.join(config.BASEDIR, task, 'labelsTr', ID)
            volume_list, label, weight, original_shape, [bbmin, bbmax] = utils.crop_brain_region(img_path, lab_path, with_gt=True)
            volumes = np.asarray(volume_list)
            np.save(os.path.join(outDir, ID+'_volumes.npy'), volumes)
            np.save(os.path.join(outDir, ID+'_label.npy'), label)
            np.save(os.path.join(outDir, ID+'_weight.npy'), weight)

            json_info = dict()
            json_info['original_shape'] = original_shape
            json_info['bbox'] = [bbmin, bbmax]
            with open(os.path.join(outDir, ID+'.json'), 'w') as f:
                json.dump(json_info, f, indent=4)

    # preprocessing train ids
    train_ids = sorted([os.path.basename(x) for x in task_archive['images_train']])
    trainDir = os.path.join(config.PREPDATADIR, task, "prepTrain")
    tinies.mkdir(trainDir) # make dir if not existing

    # preprocessing val ids
    val_ids = sorted([os.path.basename(x) for x in task_archive['images_val']])
    valDir = os.path.join(config.PREPDATADIR, task, "prepVal")
    tinies.mkdir(valDir)

    pool = multiprocessing.Pool(12) # processes=3
    pool.apply_async(func=prep, args=(train_ids, trainDir))
    pool.apply_async(func=prep, args=(val_ids, valDir))
    pool.close() # close pool, no more processes added to pool
    pool.join() # wait pool to finish, required and should be after .close()
...