pythonのマルチプロセシングを含むプログラム実行中にプログラムが止まってしまう

実現したいこと

pythonで以下のようなトレーニングループのコードを動かしています.

train(): torch.multiprosessing.Processを使って複数cpuで前処理を行なった後gpuで訓練
test(): 10epochごとに評価を行い, 評価指標の計算にtorch.multiprocessing.Pooolを用いる
multiprosessingのstart methodにはforkを用いています.

しかし, 全epochの途中でマルチプロセシングの段階でcpuの処理が0%になり, コードが止まってしまうことがあります. 具体的には, trainとtestのmultiprocessingの処理段階で進行状況確認のためtqdmを用いているのですがtqdmが0itのまま動かなくなってしまいます. cpu数が不足し, デッドロックのような状況が起きているかと思ったのですがマルチプロセシングに詳しくなくわかりかねている状況です.

train時に止まることも, test時に止まることもあります. また, 10epoch目のような早い段階で止まることもあれば200epochのような段階を経た状態で止まってしまうことがあります. また, 確かではないのですが同一のプログラムが同じタイミングで止まってしまうという傾向があるようです.

https://teratail.com/questions/266783
などは該当せず, また, メモリの影響で止まっているわけでもなさそうということが分かります. ぼんやりとした質問で大変恐縮ですが何かご存知の方教えていただけると幸いです.

該当のソースコード

python
1class Trainer:
2    def __init__(self, config, dataset, model):
3        self.config = config
4        self.dataset = dataset
5        self.metric = Metric(config, dataset)
6        self.sampling = UniformSampling(dataset, config)
7        print('setup sample')
8        self.device = self.config['device']
9        self.model = model
10        self.process_num = 8
11        self.max_recall = 0
12        self.model = self.model.to(self.device)
13        
14        
15    def train(self):
16        S = self.sampling.sample()　##ここでマルチプロセシングを用いています
17        loss = self.model.train()
18        return loss
19    
20    def test(self):
21            rating_list, groundTrue_list = self.model.evaluate()
22            X = zip(rating_list, groundTrue_list) 
23            pool = multiprocessing.Pool(self.process_num)##ここでマルチプロセシングを用いています
24            pre_results = pool.map(test_one_batch, X)
25            metrics = calc_metrics(pre_results)
26   
27    def train_epoch(self):
28        for epoch in range(world.TRAIN_epochs):
29            loss = self.train()
30            if epoch%10==0:
31                metrics = self.test()
32
33def test_one_batch(X):
34        metrics = calc_metric(X)
35        return metrics
36
37class UniformSampling:
38    def __init__(self, dataset: Loader, config, neg_ratio=1):
39         ....
40             
41    def sample_parallel(self, process_index, process_num, sample_users, allPos, returned_dict):
42        S = sample(sample_users, allPos)
43        returned_dict[process_index] = np.array(S)
44
45        
46    
47    def sample(self):
48        user_num = self.user_num
49        sample_users = np.random.randint(0, self.n_users, user_num)
50        allPos = self.allPos
51        manager = Manager()
52        returned_dict = manager.dict()
53        process_list = []
54        for i in range(self.process_num):
55            process = multiprocessing.Process(
56                target = self.sample_parallel,
57                kwargs={'process_index':i,
58                        'process_num':self.process_num,
59                        'sample_users': sample_users,
60                        'allPos': allPos,
61                        'returned_dict': returned_dict}
62            )
63            process.start()
64            process_list.append(process)
65            
66        for process in process_list:
67            process.join()
68            
69        return np.concatenate(returned_dict.values())
70        
71                
72            
73