worker processes won't hang when any other of workers is already done (#52975)

python/paddle/fluid/dataloader/dataloader_iter.py

@@ -457,6 +457,7 @@ class _DataLoaderIterMultiProcess(_DataLoaderIterBase):
 
         for i in range(self._num_workers):
             indices_queue = multiprocessing.Queue()
+            indices_queue.cancel_join_thread()
             self._indices_queues.append(indices_queue)
             worker = multiprocessing.Process(
                 target=_worker_loop,

python/paddle/fluid/dataloader/worker.py

@@ -398,3 +398,6 @@ def _worker_loop(
     finally:
         if use_shared_memory:
             _cleanup_mmap()
+    if done_event.is_set():
+        out_queue.cancel_join_thread()
+        out_queue.close()