Fix process hang in process-group shutdown (#7941)

Removing the file used as the file-store while the process-group is
still active is invalid as it is still in use.
If `reuse_dist_env` is `True` the process group is still active and the
processes will try reading from that file waiting for it to exists. In
the shutdown (`destroy_process_group`) they will wait for all threads to
join but (at least) one is still waiting for that file. This will cause
the process to hang until a PyTorch-internal timeout is reached, which
currently is ~ 5minutes

Solution is to create a unique file. I chose to put it in in `tmpdir`
and add a suffix to differentiate it.

Note that `tmpdir` is not enough as this method is called through the
fixture setup already once so that is not clean when called later in the
test execution

CC @mrwyattii , author of #3850 adding this code

---------

Signed-off-by: Alexander Grund <alexander.grund@tu-dresden.de>

Author: Flamefire

tests/unit/common.py

@@ -12,6 +12,7 @@
 from abc import ABC, abstractmethod
 from pathlib import Path
 import random
+import tempfile
 import numpy as np
 from typing import Callable, Any
 
@@ -336,18 +337,13 @@ def _dist_run(self, local_rank, num_procs, master_port, init_method, skip_msg=""
 
     def _launch_with_file_store(self, request, world_size):
         tmpdir = request.getfixturevalue("tmpdir")
-        dist_file_store = tmpdir.join("dist_file_store")
-        assert not os.path.exists(dist_file_store)
-        init_method = f"file://{dist_file_store}"
 
         if isinstance(world_size, int):
             world_size = [world_size]
         for procs in world_size:
-            try:
-                self._launch_procs(procs, init_method)
-            finally:
-                if os.path.exists(dist_file_store):
-                    os.remove(dist_file_store)
+            with tempfile.NamedTemporaryFile(delete=False, dir=str(tmpdir), suffix='_filestore') as fp:
+                init_method = f"file://{fp.name}"
+            self._launch_procs(procs, init_method)
             time.sleep(0.5)
 
     def _dist_destroy(self):
@@ -357,7 +353,7 @@ def _dist_destroy(self):
 
     def _close_pool(self, pool, num_procs, force=False):
         if force or not self.reuse_dist_env:
-            msg = pool.starmap(self._dist_destroy, [() for _ in range(num_procs)])
+            pool.starmap(self._dist_destroy, [() for _ in range(num_procs)])
             pool.close()
             pool.join()