We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix backward compatibilty
## 🐛 Bug description

Recent PR on the change of `mp.spawn` by `mp.start_process` brakes compatibility with PyTorch 1.3.1 and 1.4.0 versions.

https://github.com/pytorch/ignite/actions?query=workflow%3A.github%2Fworkflows%2Fpytorch-version-tests.yml

We have to fix that.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ignite/distributed/comp_models/native.py`
Content:
```
1 import os
2 import subprocess
3 import warnings
4 from distutils.version import LooseVersion
5 from typing import Callable, Mapping, Optional, Tuple
6 
7 import torch
8 import torch.distributed as dist
9 import torch.multiprocessing as mp
10 
11 from ignite.distributed.comp_models.base import ComputationModel
12 
13 has_native_dist_support = dist.is_available()
14 
15 
16 if has_native_dist_support:
17 
18     NCCL = dist.Backend.NCCL
19     GLOO = dist.Backend.GLOO
20     MPI = dist.Backend.MPI
21 
22     class _NativeDistModel(ComputationModel):
23         """Private class for PyTorch native distributed computation model.
24 
25         Supported `backends <https://pytorch.org/docs/stable/distributed.html#backends>`_:
26 
27         - NCCL
28         - GLOO
29         - MPI
30 
31         In this implementation we assume the following mapping between backend and devices:
32 
33         - NCCL <-> GPU
34         - GLOO <-> CPU
35         - MPI  <-> CPU
36 
37         """
38 
39         name = "native-dist"
40         available_backends = tuple(
41             name for name in [NCCL, GLOO, MPI] if getattr(dist, "is_{}_available".format(name))()
42         )
43 
44         @staticmethod
45         def create_from_context() -> Optional["_NativeDistModel"]:
46             if not (dist.is_available() and dist.is_initialized()):
47                 return None
48             return _NativeDistModel()
49 
50         @staticmethod
51         def create_from_backend(backend: str, **kwargs) -> "_NativeDistModel":
52             if dist.is_available() and dist.is_initialized():
53                 raise RuntimeError("Can not create new distributed process group if default one is already initialized")
54             return _NativeDistModel(backend=backend, **kwargs)
55 
56         def __init__(self, backend=None, timeout=None, **kwargs):
57             """This is a private method. Please, use `create_from_backend` or `create_from_context`
58             """
59             super(_NativeDistModel, self).__init__()
60             self._env_backup = None
61             if backend is not None:
62                 self._create_from_backend(backend, timeout=timeout, **kwargs)
63             else:
64                 self._init_from_context()
65 
66         def _create_from_backend(self, backend, timeout=None, **kwargs):
67             if backend == dist.Backend.NCCL and not torch.cuda.is_available():
68                 raise RuntimeError("Nccl backend is required but no cuda capable devices")
69 
70             self.setup_env_vars()
71 
72             self._local_rank = int(os.environ["LOCAL_RANK"])
73             # for debug purposes
74             self._master_port = int(os.environ["MASTER_PORT"])
75             self._master_addr = os.environ["MASTER_ADDR"]
76 
77             init_pg_kwargs = {}
78             if timeout is not None:
79                 init_pg_kwargs["timeout"] = timeout
80 
81             dist.init_process_group(backend, init_method="env://", **init_pg_kwargs)
82             # https://github.com/facebookresearch/maskrcnn-benchmark/issues/172
83             dist.barrier()
84 
85             if backend == dist.Backend.NCCL:
86                 torch.cuda.set_device(self._local_rank)
87 
88             self._setup_attrs()
89 
90         def _init_from_context(self):
91 
92             self._identify_local_rank()
93 
94             # for debug purposes
95             self._master_port = None
96             self._master_addr = None
97             self._setup_attrs()
98 
99         def _compute_nproc_per_node(self):
100             tensor = torch.tensor([self.get_local_rank() + 1]).to(self.device())
101             dist.all_reduce(tensor, op=dist.ReduceOp.MAX)
102             return tensor.item()
103 
104         def _get_all_hostnames(self):
105             import socket
106 
107             device = "cpu"
108             if self.backend() == dist.Backend.NCCL:
109                 index = torch.cuda.current_device()
110                 device = "cuda:{}".format(index)
111             name = socket.gethostname()
112             name = torch.tensor(bytearray(name, "utf-8")).to(device)
113             padded_t_name = torch.zeros(256, device=device, dtype=torch.long)
114             padded_t_name[: len(name)] = name
115             out_t_names = [torch.zeros_like(padded_t_name) for _ in range(self.get_world_size())]
116             dist.all_gather(out_t_names, padded_t_name)
117             out_t_names = [tuple(t.cpu().tolist()) for t in out_t_names]
118             return out_t_names
119 
120         @staticmethod
121         def _compute_node_and_local_ranks(rank, hostnames):
122             from collections import Counter
123 
124             c = Counter(hostnames)
125             sizes = torch.tensor([0,] + list(c.values()))
126             cumsum_sizes = torch.cumsum(sizes, dim=0)
127             node_rank = (rank // cumsum_sizes[1:]).clamp(0, 1).sum().item()
128             local_rank = rank - cumsum_sizes[node_rank].item()
129             return local_rank, node_rank
130 
131         def _compute_local_rank_via_hostname(self):
132             # get all hostnames
133             hostnames = self._get_all_hostnames()
134             local_rank, self._node = self._compute_node_and_local_ranks(self.get_rank(), hostnames)
135 
136             if local_rank < 0 or self._node < 0:
137                 raise ValueError(
138                     "Failed to correctly estimate local rank. "
139                     "Debugging info: local rank: {}, node rank: {}, hostnames: {}".format(
140                         local_rank, self._node, hostnames
141                     )
142                 )
143             return local_rank
144 
145         def _identify_local_rank(self):
146 
147             if "SLURM_JOBID" in os.environ:
148                 os.environ["LOCAL_RANK"] = os.environ["SLURM_LOCALID"]
149 
150             if "LOCAL_RANK" in os.environ:
151                 self._local_rank = int(os.environ["LOCAL_RANK"])
152             elif self._ext_local_rank is not None:
153                 self._local_rank = self._ext_local_rank
154             else:
155                 warnings.warn(
156                     "Local rank information for native distributed setting will be initialized using "
157                     "heuristic approach based on hostname which can be different of real setup. Please, "
158                     "either set `os.environ['LOCAL_RANK']` "
159                     "or use `idist.set_local_rank(local_rank)` with correct local rank index."
160                 )
161                 # use socket gethostname heuristic to determine number of nodes => local rank
162                 self._local_rank = self._compute_local_rank_via_hostname()
163 
164         def setup_env_vars(self):
165 
166             self._env_backup = os.environ.copy()
167 
168             if "SLURM_JOBID" in os.environ:
169                 self._setup_env_in_slurm()
170                 return
171 
172             # check if all necessary env vars are set
173             # if partially defined raise an error
174             necessary_env_vars = ["RANK", "LOCAL_RANK", "WORLD_SIZE"]
175             all_env_vars_defined = [k in os.environ for k in necessary_env_vars]
176             if any(all_env_vars_defined) and not all(all_env_vars_defined):
177                 raise RuntimeError(
178                     "PyTorch distributed configuration should define env variables '{}'".format(necessary_env_vars)
179                 )
180 
181             os.environ["RANK"] = os.environ.get("RANK", "0")
182             os.environ["LOCAL_RANK"] = os.environ.get("LOCAL_RANK", "0")
183             os.environ["WORLD_SIZE"] = os.environ.get("WORLD_SIZE", "1")
184             os.environ["MASTER_ADDR"] = os.environ.get("MASTER_ADDR", "127.0.0.1")
185             os.environ["MASTER_PORT"] = os.environ.get("MASTER_PORT", "15000")
186 
187         def _setup_env_in_slurm(self):
188             for k in ["SLURM_PROCID", "SLURM_LOCALID", "SLURM_NTASKS", "SLURM_JOB_NODELIST"]:
189                 if k not in os.environ:
190                     raise RuntimeError("SLURM distributed configuration is missing '{}' in env variables".format(k))
191 
192             os.environ["RANK"] = os.environ["SLURM_PROCID"]
193             os.environ["LOCAL_RANK"] = os.environ["SLURM_LOCALID"]
194             os.environ["WORLD_SIZE"] = os.environ["SLURM_NTASKS"]
195             # port should be the same over all process
196             slurm_port = os.environ["SLURM_JOB_ID"]
197             slurm_port = slurm_port[-4:]
198             os.environ["MASTER_PORT"] = str(int(slurm_port) + 15000)
199             # master address is the first hostname of nodes list
200             hostnames = subprocess.check_output(["scontrol", "show", "hostnames", os.environ["SLURM_JOB_NODELIST"]])
201             os.environ["MASTER_ADDR"] = hostnames.split()[0].decode("utf-8")
202 
203         def get_local_rank(self) -> int:
204             return self._local_rank
205 
206         def get_rank(self) -> int:
207             return dist.get_rank()
208 
209         def get_world_size(self) -> int:
210             return dist.get_world_size()
211 
212         def get_nproc_per_node(self) -> int:
213             return self._nproc_per_node
214 
215         def get_nnodes(self) -> int:
216             return self._nnodes
217 
218         def get_node_rank(self) -> int:
219             return self._node
220 
221         def device(self) -> torch.device:
222             if self.backend() == dist.Backend.NCCL:
223                 index = torch.cuda.current_device()
224                 return torch.device("cuda:{}".format(index))
225             return torch.device("cpu")
226 
227         def backend(self) -> str:
228             return dist.get_backend()
229 
230         def finalize(self):
231             dist.destroy_process_group()
232             # restore backed-up env
233             if self._env_backup is not None:
234                 os.environ.clear()
235                 os.environ.update(self._env_backup)
236 
237         @staticmethod
238         def _dist_worker_task_fn(
239             local_rank, backend, fn, args, kw_dict, world_size, nprocs_per_node, node_rank, master_addr, master_port, kw
240         ):
241             from ignite.distributed.utils import _set_model, finalize
242 
243             copy_env_vars = os.environ.copy()
244 
245             os.environ["LOCAL_RANK"] = str(local_rank)
246             os.environ["RANK"] = str(node_rank * nprocs_per_node + local_rank)
247             os.environ["WORLD_SIZE"] = str(world_size)
248             os.environ["MASTER_ADDR"] = str(master_addr)
249             os.environ["MASTER_PORT"] = str(master_port)
250 
251             model = _NativeDistModel.create_from_backend(backend, **kw)
252             _set_model(model)
253             fn(local_rank, *args, **kw_dict)
254             finalize()
255 
256             os.environ.clear()
257             os.environ.update(copy_env_vars)
258 
259         @staticmethod
260         def spawn(
261             fn: Callable,
262             args: Tuple,
263             kwargs_dict: Optional[Mapping] = None,
264             nproc_per_node: int = 1,
265             nnodes: int = 1,
266             node_rank: int = 0,
267             master_addr: str = "127.0.0.1",
268             master_port: int = 2222,
269             backend: str = "nccl",
270             **kwargs
271         ):
272             world_size = nnodes * nproc_per_node
273 
274             spawn_kwargs = {
275                 "join": kwargs.get("join", True),
276                 "daemon": kwargs.get("daemon", False),
277             }
278             # start_method in pytorch >= 1.5
279             if LooseVersion(torch.__version__) >= LooseVersion("1.5.0"):
280                 spawn_kwargs["start_method"] = kwargs.get("start_method", "spawn")
281 
282             mp.start_processes(
283                 _NativeDistModel._dist_worker_task_fn,
284                 nprocs=nproc_per_node,
285                 args=(
286                     backend,
287                     fn,
288                     args,
289                     kwargs_dict,
290                     world_size,
291                     nproc_per_node,
292                     node_rank,
293                     master_addr,
294                     master_port,
295                     kwargs,
296                 ),
297                 **spawn_kwargs,
298             )
299 
300         _reduce_op_map = {
301             "SUM": dist.ReduceOp.SUM,
302             "PRODUCT": dist.ReduceOp.PRODUCT,
303             "MIN": dist.ReduceOp.MIN,
304             "MAX": dist.ReduceOp.MAX,
305             "AND": dist.ReduceOp.BAND,
306             "OR": dist.ReduceOp.BOR,
307         }
308 
309         def _do_all_reduce(self, tensor: torch.Tensor, op: str = "SUM") -> torch.Tensor:
310             if op not in self._reduce_op_map:
311                 raise ValueError("Unsupported reduction operation: '{}'".format(op))
312             op = self._reduce_op_map[op]
313             dist.all_reduce(tensor, op)
314             return tensor
315 
316         def _do_all_gather(self, tensor: torch.Tensor) -> torch.Tensor:
317             if tensor.ndimension() == 0:
318                 tensor = tensor.unsqueeze(0)
319             output = [torch.zeros_like(tensor) for _ in range(self.get_world_size())]
320             dist.all_gather(output, tensor)
321             return torch.cat(output, dim=0)
322 
323         def _do_broadcast(self, tensor: torch.Tensor, src: int) -> torch.Tensor:
324             dist.broadcast(tensor, src=src)
325             return tensor
326 
327         def barrier(self):
328             dist.barrier()
329
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ignite/distributed/comp_models/native.py b/ignite/distributed/comp_models/native.py
--- a/ignite/distributed/comp_models/native.py
+++ b/ignite/distributed/comp_models/native.py
@@ -275,11 +275,14 @@
                 "join": kwargs.get("join", True),
                 "daemon": kwargs.get("daemon", False),
             }
-            # start_method in pytorch >= 1.5
+
+            start_processes = mp.spawn
+            # start_method and start_processes in pytorch >= 1.5
             if LooseVersion(torch.__version__) >= LooseVersion("1.5.0"):
                 spawn_kwargs["start_method"] = kwargs.get("start_method", "spawn")
+                start_processes = mp.start_processes
 
-            mp.start_processes(
+            start_processes(
                 _NativeDistModel._dist_worker_task_fn,
                 nprocs=nproc_per_node,
                 args=(

{"golden_diff": "diff --git a/ignite/distributed/comp_models/native.py b/ignite/distributed/comp_models/native.py\n--- a/ignite/distributed/comp_models/native.py\n+++ b/ignite/distributed/comp_models/native.py\n@@ -275,11 +275,14 @@\n                 \"join\": kwargs.get(\"join\", True),\n                 \"daemon\": kwargs.get(\"daemon\", False),\n             }\n-            # start_method in pytorch >= 1.5\n+\n+            start_processes = mp.spawn\n+            # start_method and start_processes in pytorch >= 1.5\n             if LooseVersion(torch.__version__) >= LooseVersion(\"1.5.0\"):\n                 spawn_kwargs[\"start_method\"] = kwargs.get(\"start_method\", \"spawn\")\n+                start_processes = mp.start_processes\n \n-            mp.start_processes(\n+            start_processes(\n                 _NativeDistModel._dist_worker_task_fn,\n                 nprocs=nproc_per_node,\n                 args=(\n", "issue": "Fix backward compatibilty\n## \ud83d\udc1b Bug description\r\n\r\nRecent PR on the change of `mp.spawn` by `mp.start_process` brakes compatibility with PyTorch 1.3.1 and 1.4.0 versions.\r\n\r\nhttps://github.com/pytorch/ignite/actions?query=workflow%3A.github%2Fworkflows%2Fpytorch-version-tests.yml\r\n\r\nWe have to fix that.\n", "before_files": [{"content": "import os\nimport subprocess\nimport warnings\nfrom distutils.version import LooseVersion\nfrom typing import Callable, Mapping, Optional, Tuple\n\nimport torch\nimport torch.distributed as dist\nimport torch.multiprocessing as mp\n\nfrom ignite.distributed.comp_models.base import ComputationModel\n\nhas_native_dist_support = dist.is_available()\n\n\nif has_native_dist_support:\n\n    NCCL = dist.Backend.NCCL\n    GLOO = dist.Backend.GLOO\n    MPI = dist.Backend.MPI\n\n    class _NativeDistModel(ComputationModel):\n        \"\"\"Private class for PyTorch native distributed computation model.\n\n        Supported `backends <https://pytorch.org/docs/stable/distributed.html#backends>`_:\n\n        - NCCL\n        - GLOO\n        - MPI\n\n        In this implementation we assume the following mapping between backend and devices:\n\n        - NCCL <-> GPU\n        - GLOO <-> CPU\n        - MPI  <-> CPU\n\n        \"\"\"\n\n        name = \"native-dist\"\n        available_backends = tuple(\n            name for name in [NCCL, GLOO, MPI] if getattr(dist, \"is_{}_available\".format(name))()\n        )\n\n        @staticmethod\n        def create_from_context() -> Optional[\"_NativeDistModel\"]:\n            if not (dist.is_available() and dist.is_initialized()):\n                return None\n            return _NativeDistModel()\n\n        @staticmethod\n        def create_from_backend(backend: str, **kwargs) -> \"_NativeDistModel\":\n            if dist.is_available() and dist.is_initialized():\n                raise RuntimeError(\"Can not create new distributed process group if default one is already initialized\")\n            return _NativeDistModel(backend=backend, **kwargs)\n\n        def __init__(self, backend=None, timeout=None, **kwargs):\n            \"\"\"This is a private method. Please, use `create_from_backend` or `create_from_context`\n            \"\"\"\n            super(_NativeDistModel, self).__init__()\n            self._env_backup = None\n            if backend is not None:\n                self._create_from_backend(backend, timeout=timeout, **kwargs)\n            else:\n                self._init_from_context()\n\n        def _create_from_backend(self, backend, timeout=None, **kwargs):\n            if backend == dist.Backend.NCCL and not torch.cuda.is_available():\n                raise RuntimeError(\"Nccl backend is required but no cuda capable devices\")\n\n            self.setup_env_vars()\n\n            self._local_rank = int(os.environ[\"LOCAL_RANK\"])\n            # for debug purposes\n            self._master_port = int(os.environ[\"MASTER_PORT\"])\n            self._master_addr = os.environ[\"MASTER_ADDR\"]\n\n            init_pg_kwargs = {}\n            if timeout is not None:\n                init_pg_kwargs[\"timeout\"] = timeout\n\n            dist.init_process_group(backend, init_method=\"env://\", **init_pg_kwargs)\n            # https://github.com/facebookresearch/maskrcnn-benchmark/issues/172\n            dist.barrier()\n\n            if backend == dist.Backend.NCCL:\n                torch.cuda.set_device(self._local_rank)\n\n            self._setup_attrs()\n\n        def _init_from_context(self):\n\n            self._identify_local_rank()\n\n            # for debug purposes\n            self._master_port = None\n            self._master_addr = None\n            self._setup_attrs()\n\n        def _compute_nproc_per_node(self):\n            tensor = torch.tensor([self.get_local_rank() + 1]).to(self.device())\n            dist.all_reduce(tensor, op=dist.ReduceOp.MAX)\n            return tensor.item()\n\n        def _get_all_hostnames(self):\n            import socket\n\n            device = \"cpu\"\n            if self.backend() == dist.Backend.NCCL:\n                index = torch.cuda.current_device()\n                device = \"cuda:{}\".format(index)\n            name = socket.gethostname()\n            name = torch.tensor(bytearray(name, \"utf-8\")).to(device)\n            padded_t_name = torch.zeros(256, device=device, dtype=torch.long)\n            padded_t_name[: len(name)] = name\n            out_t_names = [torch.zeros_like(padded_t_name) for _ in range(self.get_world_size())]\n            dist.all_gather(out_t_names, padded_t_name)\n            out_t_names = [tuple(t.cpu().tolist()) for t in out_t_names]\n            return out_t_names\n\n        @staticmethod\n        def _compute_node_and_local_ranks(rank, hostnames):\n            from collections import Counter\n\n            c = Counter(hostnames)\n            sizes = torch.tensor([0,] + list(c.values()))\n            cumsum_sizes = torch.cumsum(sizes, dim=0)\n            node_rank = (rank // cumsum_sizes[1:]).clamp(0, 1).sum().item()\n            local_rank = rank - cumsum_sizes[node_rank].item()\n            return local_rank, node_rank\n\n        def _compute_local_rank_via_hostname(self):\n            # get all hostnames\n            hostnames = self._get_all_hostnames()\n            local_rank, self._node = self._compute_node_and_local_ranks(self.get_rank(), hostnames)\n\n            if local_rank < 0 or self._node < 0:\n                raise ValueError(\n                    \"Failed to correctly estimate local rank. \"\n                    \"Debugging info: local rank: {}, node rank: {}, hostnames: {}\".format(\n                        local_rank, self._node, hostnames\n                    )\n                )\n            return local_rank\n\n        def _identify_local_rank(self):\n\n            if \"SLURM_JOBID\" in os.environ:\n                os.environ[\"LOCAL_RANK\"] = os.environ[\"SLURM_LOCALID\"]\n\n            if \"LOCAL_RANK\" in os.environ:\n                self._local_rank = int(os.environ[\"LOCAL_RANK\"])\n            elif self._ext_local_rank is not None:\n                self._local_rank = self._ext_local_rank\n            else:\n                warnings.warn(\n                    \"Local rank information for native distributed setting will be initialized using \"\n                    \"heuristic approach based on hostname which can be different of real setup. Please, \"\n                    \"either set `os.environ['LOCAL_RANK']` \"\n                    \"or use `idist.set_local_rank(local_rank)` with correct local rank index.\"\n                )\n                # use socket gethostname heuristic to determine number of nodes => local rank\n                self._local_rank = self._compute_local_rank_via_hostname()\n\n        def setup_env_vars(self):\n\n            self._env_backup = os.environ.copy()\n\n            if \"SLURM_JOBID\" in os.environ:\n                self._setup_env_in_slurm()\n                return\n\n            # check if all necessary env vars are set\n            # if partially defined raise an error\n            necessary_env_vars = [\"RANK\", \"LOCAL_RANK\", \"WORLD_SIZE\"]\n            all_env_vars_defined = [k in os.environ for k in necessary_env_vars]\n            if any(all_env_vars_defined) and not all(all_env_vars_defined):\n                raise RuntimeError(\n                    \"PyTorch distributed configuration should define env variables '{}'\".format(necessary_env_vars)\n                )\n\n            os.environ[\"RANK\"] = os.environ.get(\"RANK\", \"0\")\n            os.environ[\"LOCAL_RANK\"] = os.environ.get(\"LOCAL_RANK\", \"0\")\n            os.environ[\"WORLD_SIZE\"] = os.environ.get(\"WORLD_SIZE\", \"1\")\n            os.environ[\"MASTER_ADDR\"] = os.environ.get(\"MASTER_ADDR\", \"127.0.0.1\")\n            os.environ[\"MASTER_PORT\"] = os.environ.get(\"MASTER_PORT\", \"15000\")\n\n        def _setup_env_in_slurm(self):\n            for k in [\"SLURM_PROCID\", \"SLURM_LOCALID\", \"SLURM_NTASKS\", \"SLURM_JOB_NODELIST\"]:\n                if k not in os.environ:\n                    raise RuntimeError(\"SLURM distributed configuration is missing '{}' in env variables\".format(k))\n\n            os.environ[\"RANK\"] = os.environ[\"SLURM_PROCID\"]\n            os.environ[\"LOCAL_RANK\"] = os.environ[\"SLURM_LOCALID\"]\n            os.environ[\"WORLD_SIZE\"] = os.environ[\"SLURM_NTASKS\"]\n            # port should be the same over all process\n            slurm_port = os.environ[\"SLURM_JOB_ID\"]\n            slurm_port = slurm_port[-4:]\n            os.environ[\"MASTER_PORT\"] = str(int(slurm_port) + 15000)\n            # master address is the first hostname of nodes list\n            hostnames = subprocess.check_output([\"scontrol\", \"show\", \"hostnames\", os.environ[\"SLURM_JOB_NODELIST\"]])\n            os.environ[\"MASTER_ADDR\"] = hostnames.split()[0].decode(\"utf-8\")\n\n        def get_local_rank(self) -> int:\n            return self._local_rank\n\n        def get_rank(self) -> int:\n            return dist.get_rank()\n\n        def get_world_size(self) -> int:\n            return dist.get_world_size()\n\n        def get_nproc_per_node(self) -> int:\n            return self._nproc_per_node\n\n        def get_nnodes(self) -> int:\n            return self._nnodes\n\n        def get_node_rank(self) -> int:\n            return self._node\n\n        def device(self) -> torch.device:\n            if self.backend() == dist.Backend.NCCL:\n                index = torch.cuda.current_device()\n                return torch.device(\"cuda:{}\".format(index))\n            return torch.device(\"cpu\")\n\n        def backend(self) -> str:\n            return dist.get_backend()\n\n        def finalize(self):\n            dist.destroy_process_group()\n            # restore backed-up env\n            if self._env_backup is not None:\n                os.environ.clear()\n                os.environ.update(self._env_backup)\n\n        @staticmethod\n        def _dist_worker_task_fn(\n            local_rank, backend, fn, args, kw_dict, world_size, nprocs_per_node, node_rank, master_addr, master_port, kw\n        ):\n            from ignite.distributed.utils import _set_model, finalize\n\n            copy_env_vars = os.environ.copy()\n\n            os.environ[\"LOCAL_RANK\"] = str(local_rank)\n            os.environ[\"RANK\"] = str(node_rank * nprocs_per_node + local_rank)\n            os.environ[\"WORLD_SIZE\"] = str(world_size)\n            os.environ[\"MASTER_ADDR\"] = str(master_addr)\n            os.environ[\"MASTER_PORT\"] = str(master_port)\n\n            model = _NativeDistModel.create_from_backend(backend, **kw)\n            _set_model(model)\n            fn(local_rank, *args, **kw_dict)\n            finalize()\n\n            os.environ.clear()\n            os.environ.update(copy_env_vars)\n\n        @staticmethod\n        def spawn(\n            fn: Callable,\n            args: Tuple,\n            kwargs_dict: Optional[Mapping] = None,\n            nproc_per_node: int = 1,\n            nnodes: int = 1,\n            node_rank: int = 0,\n            master_addr: str = \"127.0.0.1\",\n            master_port: int = 2222,\n            backend: str = \"nccl\",\n            **kwargs\n        ):\n            world_size = nnodes * nproc_per_node\n\n            spawn_kwargs = {\n                \"join\": kwargs.get(\"join\", True),\n                \"daemon\": kwargs.get(\"daemon\", False),\n            }\n            # start_method in pytorch >= 1.5\n            if LooseVersion(torch.__version__) >= LooseVersion(\"1.5.0\"):\n                spawn_kwargs[\"start_method\"] = kwargs.get(\"start_method\", \"spawn\")\n\n            mp.start_processes(\n                _NativeDistModel._dist_worker_task_fn,\n                nprocs=nproc_per_node,\n                args=(\n                    backend,\n                    fn,\n                    args,\n                    kwargs_dict,\n                    world_size,\n                    nproc_per_node,\n                    node_rank,\n                    master_addr,\n                    master_port,\n                    kwargs,\n                ),\n                **spawn_kwargs,\n            )\n\n        _reduce_op_map = {\n            \"SUM\": dist.ReduceOp.SUM,\n            \"PRODUCT\": dist.ReduceOp.PRODUCT,\n            \"MIN\": dist.ReduceOp.MIN,\n            \"MAX\": dist.ReduceOp.MAX,\n            \"AND\": dist.ReduceOp.BAND,\n            \"OR\": dist.ReduceOp.BOR,\n        }\n\n        def _do_all_reduce(self, tensor: torch.Tensor, op: str = \"SUM\") -> torch.Tensor:\n            if op not in self._reduce_op_map:\n                raise ValueError(\"Unsupported reduction operation: '{}'\".format(op))\n            op = self._reduce_op_map[op]\n            dist.all_reduce(tensor, op)\n            return tensor\n\n        def _do_all_gather(self, tensor: torch.Tensor) -> torch.Tensor:\n            if tensor.ndimension() == 0:\n                tensor = tensor.unsqueeze(0)\n            output = [torch.zeros_like(tensor) for _ in range(self.get_world_size())]\n            dist.all_gather(output, tensor)\n            return torch.cat(output, dim=0)\n\n        def _do_broadcast(self, tensor: torch.Tensor, src: int) -> torch.Tensor:\n            dist.broadcast(tensor, src=src)\n            return tensor\n\n        def barrier(self):\n            dist.barrier()\n", "path": "ignite/distributed/comp_models/native.py"}], "after_files": [{"content": "import os\nimport subprocess\nimport warnings\nfrom distutils.version import LooseVersion\nfrom typing import Callable, Mapping, Optional, Tuple\n\nimport torch\nimport torch.distributed as dist\nimport torch.multiprocessing as mp\n\nfrom ignite.distributed.comp_models.base import ComputationModel\n\nhas_native_dist_support = dist.is_available()\n\n\nif has_native_dist_support:\n\n    NCCL = dist.Backend.NCCL\n    GLOO = dist.Backend.GLOO\n    MPI = dist.Backend.MPI\n\n    class _NativeDistModel(ComputationModel):\n        \"\"\"Private class for PyTorch native distributed computation model.\n\n        Supported `backends <https://pytorch.org/docs/stable/distributed.html#backends>`_:\n\n        - NCCL\n        - GLOO\n        - MPI\n\n        In this implementation we assume the following mapping between backend and devices:\n\n        - NCCL <-> GPU\n        - GLOO <-> CPU\n        - MPI  <-> CPU\n\n        \"\"\"\n\n        name = \"native-dist\"\n        available_backends = tuple(\n            name for name in [NCCL, GLOO, MPI] if getattr(dist, \"is_{}_available\".format(name))()\n        )\n\n        @staticmethod\n        def create_from_context() -> Optional[\"_NativeDistModel\"]:\n            if not (dist.is_available() and dist.is_initialized()):\n                return None\n            return _NativeDistModel()\n\n        @staticmethod\n        def create_from_backend(backend: str, **kwargs) -> \"_NativeDistModel\":\n            if dist.is_available() and dist.is_initialized():\n                raise RuntimeError(\"Can not create new distributed process group if default one is already initialized\")\n            return _NativeDistModel(backend=backend, **kwargs)\n\n        def __init__(self, backend=None, timeout=None, **kwargs):\n            \"\"\"This is a private method. Please, use `create_from_backend` or `create_from_context`\n            \"\"\"\n            super(_NativeDistModel, self).__init__()\n            self._env_backup = None\n            if backend is not None:\n                self._create_from_backend(backend, timeout=timeout, **kwargs)\n            else:\n                self._init_from_context()\n\n        def _create_from_backend(self, backend, timeout=None, **kwargs):\n            if backend == dist.Backend.NCCL and not torch.cuda.is_available():\n                raise RuntimeError(\"Nccl backend is required but no cuda capable devices\")\n\n            self.setup_env_vars()\n\n            self._local_rank = int(os.environ[\"LOCAL_RANK\"])\n            # for debug purposes\n            self._master_port = int(os.environ[\"MASTER_PORT\"])\n            self._master_addr = os.environ[\"MASTER_ADDR\"]\n\n            init_pg_kwargs = {}\n            if timeout is not None:\n                init_pg_kwargs[\"timeout\"] = timeout\n\n            dist.init_process_group(backend, init_method=\"env://\", **init_pg_kwargs)\n            # https://github.com/facebookresearch/maskrcnn-benchmark/issues/172\n            dist.barrier()\n\n            if backend == dist.Backend.NCCL:\n                torch.cuda.set_device(self._local_rank)\n\n            self._setup_attrs()\n\n        def _init_from_context(self):\n\n            self._identify_local_rank()\n\n            # for debug purposes\n            self._master_port = None\n            self._master_addr = None\n            self._setup_attrs()\n\n        def _compute_nproc_per_node(self):\n            tensor = torch.tensor([self.get_local_rank() + 1]).to(self.device())\n            dist.all_reduce(tensor, op=dist.ReduceOp.MAX)\n            return tensor.item()\n\n        def _get_all_hostnames(self):\n            import socket\n\n            device = \"cpu\"\n            if self.backend() == dist.Backend.NCCL:\n                index = torch.cuda.current_device()\n                device = \"cuda:{}\".format(index)\n            name = socket.gethostname()\n            name = torch.tensor(bytearray(name, \"utf-8\")).to(device)\n            padded_t_name = torch.zeros(256, device=device, dtype=torch.long)\n            padded_t_name[: len(name)] = name\n            out_t_names = [torch.zeros_like(padded_t_name) for _ in range(self.get_world_size())]\n            dist.all_gather(out_t_names, padded_t_name)\n            out_t_names = [tuple(t.cpu().tolist()) for t in out_t_names]\n            return out_t_names\n\n        @staticmethod\n        def _compute_node_and_local_ranks(rank, hostnames):\n            from collections import Counter\n\n            c = Counter(hostnames)\n            sizes = torch.tensor([0,] + list(c.values()))\n            cumsum_sizes = torch.cumsum(sizes, dim=0)\n            node_rank = (rank // cumsum_sizes[1:]).clamp(0, 1).sum().item()\n            local_rank = rank - cumsum_sizes[node_rank].item()\n            return local_rank, node_rank\n\n        def _compute_local_rank_via_hostname(self):\n            # get all hostnames\n            hostnames = self._get_all_hostnames()\n            local_rank, self._node = self._compute_node_and_local_ranks(self.get_rank(), hostnames)\n\n            if local_rank < 0 or self._node < 0:\n                raise ValueError(\n                    \"Failed to correctly estimate local rank. \"\n                    \"Debugging info: local rank: {}, node rank: {}, hostnames: {}\".format(\n                        local_rank, self._node, hostnames\n                    )\n                )\n            return local_rank\n\n        def _identify_local_rank(self):\n\n            if \"SLURM_JOBID\" in os.environ:\n                os.environ[\"LOCAL_RANK\"] = os.environ[\"SLURM_LOCALID\"]\n\n            if \"LOCAL_RANK\" in os.environ:\n                self._local_rank = int(os.environ[\"LOCAL_RANK\"])\n            elif self._ext_local_rank is not None:\n                self._local_rank = self._ext_local_rank\n            else:\n                warnings.warn(\n                    \"Local rank information for native distributed setting will be initialized using \"\n                    \"heuristic approach based on hostname which can be different of real setup. Please, \"\n                    \"either set `os.environ['LOCAL_RANK']` \"\n                    \"or use `idist.set_local_rank(local_rank)` with correct local rank index.\"\n                )\n                # use socket gethostname heuristic to determine number of nodes => local rank\n                self._local_rank = self._compute_local_rank_via_hostname()\n\n        def setup_env_vars(self):\n\n            self._env_backup = os.environ.copy()\n\n            if \"SLURM_JOBID\" in os.environ:\n                self._setup_env_in_slurm()\n                return\n\n            # check if all necessary env vars are set\n            # if partially defined raise an error\n            necessary_env_vars = [\"RANK\", \"LOCAL_RANK\", \"WORLD_SIZE\"]\n            all_env_vars_defined = [k in os.environ for k in necessary_env_vars]\n            if any(all_env_vars_defined) and not all(all_env_vars_defined):\n                raise RuntimeError(\n                    \"PyTorch distributed configuration should define env variables '{}'\".format(necessary_env_vars)\n                )\n\n            os.environ[\"RANK\"] = os.environ.get(\"RANK\", \"0\")\n            os.environ[\"LOCAL_RANK\"] = os.environ.get(\"LOCAL_RANK\", \"0\")\n            os.environ[\"WORLD_SIZE\"] = os.environ.get(\"WORLD_SIZE\", \"1\")\n            os.environ[\"MASTER_ADDR\"] = os.environ.get(\"MASTER_ADDR\", \"127.0.0.1\")\n            os.environ[\"MASTER_PORT\"] = os.environ.get(\"MASTER_PORT\", \"15000\")\n\n        def _setup_env_in_slurm(self):\n            for k in [\"SLURM_PROCID\", \"SLURM_LOCALID\", \"SLURM_NTASKS\", \"SLURM_JOB_NODELIST\"]:\n                if k not in os.environ:\n                    raise RuntimeError(\"SLURM distributed configuration is missing '{}' in env variables\".format(k))\n\n            os.environ[\"RANK\"] = os.environ[\"SLURM_PROCID\"]\n            os.environ[\"LOCAL_RANK\"] = os.environ[\"SLURM_LOCALID\"]\n            os.environ[\"WORLD_SIZE\"] = os.environ[\"SLURM_NTASKS\"]\n            # port should be the same over all process\n            slurm_port = os.environ[\"SLURM_JOB_ID\"]\n            slurm_port = slurm_port[-4:]\n            os.environ[\"MASTER_PORT\"] = str(int(slurm_port) + 15000)\n            # master address is the first hostname of nodes list\n            hostnames = subprocess.check_output([\"scontrol\", \"show\", \"hostnames\", os.environ[\"SLURM_JOB_NODELIST\"]])\n            os.environ[\"MASTER_ADDR\"] = hostnames.split()[0].decode(\"utf-8\")\n\n        def get_local_rank(self) -> int:\n            return self._local_rank\n\n        def get_rank(self) -> int:\n            return dist.get_rank()\n\n        def get_world_size(self) -> int:\n            return dist.get_world_size()\n\n        def get_nproc_per_node(self) -> int:\n            return self._nproc_per_node\n\n        def get_nnodes(self) -> int:\n            return self._nnodes\n\n        def get_node_rank(self) -> int:\n            return self._node\n\n        def device(self) -> torch.device:\n            if self.backend() == dist.Backend.NCCL:\n                index = torch.cuda.current_device()\n                return torch.device(\"cuda:{}\".format(index))\n            return torch.device(\"cpu\")\n\n        def backend(self) -> str:\n            return dist.get_backend()\n\n        def finalize(self):\n            dist.destroy_process_group()\n            # restore backed-up env\n            if self._env_backup is not None:\n                os.environ.clear()\n                os.environ.update(self._env_backup)\n\n        @staticmethod\n        def _dist_worker_task_fn(\n            local_rank, backend, fn, args, kw_dict, world_size, nprocs_per_node, node_rank, master_addr, master_port, kw\n        ):\n            from ignite.distributed.utils import _set_model, finalize\n\n            copy_env_vars = os.environ.copy()\n\n            os.environ[\"LOCAL_RANK\"] = str(local_rank)\n            os.environ[\"RANK\"] = str(node_rank * nprocs_per_node + local_rank)\n            os.environ[\"WORLD_SIZE\"] = str(world_size)\n            os.environ[\"MASTER_ADDR\"] = str(master_addr)\n            os.environ[\"MASTER_PORT\"] = str(master_port)\n\n            model = _NativeDistModel.create_from_backend(backend, **kw)\n            _set_model(model)\n            fn(local_rank, *args, **kw_dict)\n            finalize()\n\n            os.environ.clear()\n            os.environ.update(copy_env_vars)\n\n        @staticmethod\n        def spawn(\n            fn: Callable,\n            args: Tuple,\n            kwargs_dict: Optional[Mapping] = None,\n            nproc_per_node: int = 1,\n            nnodes: int = 1,\n            node_rank: int = 0,\n            master_addr: str = \"127.0.0.1\",\n            master_port: int = 2222,\n            backend: str = \"nccl\",\n            **kwargs\n        ):\n            world_size = nnodes * nproc_per_node\n\n            spawn_kwargs = {\n                \"join\": kwargs.get(\"join\", True),\n                \"daemon\": kwargs.get(\"daemon\", False),\n            }\n\n            start_processes = mp.spawn\n            # start_method and start_processes in pytorch >= 1.5\n            if LooseVersion(torch.__version__) >= LooseVersion(\"1.5.0\"):\n                spawn_kwargs[\"start_method\"] = kwargs.get(\"start_method\", \"spawn\")\n                start_processes = mp.start_processes\n\n            start_processes(\n                _NativeDistModel._dist_worker_task_fn,\n                nprocs=nproc_per_node,\n                args=(\n                    backend,\n                    fn,\n                    args,\n                    kwargs_dict,\n                    world_size,\n                    nproc_per_node,\n                    node_rank,\n                    master_addr,\n                    master_port,\n                    kwargs,\n                ),\n                **spawn_kwargs,\n            )\n\n        _reduce_op_map = {\n            \"SUM\": dist.ReduceOp.SUM,\n            \"PRODUCT\": dist.ReduceOp.PRODUCT,\n            \"MIN\": dist.ReduceOp.MIN,\n            \"MAX\": dist.ReduceOp.MAX,\n            \"AND\": dist.ReduceOp.BAND,\n            \"OR\": dist.ReduceOp.BOR,\n        }\n\n        def _do_all_reduce(self, tensor: torch.Tensor, op: str = \"SUM\") -> torch.Tensor:\n            if op not in self._reduce_op_map:\n                raise ValueError(\"Unsupported reduction operation: '{}'\".format(op))\n            op = self._reduce_op_map[op]\n            dist.all_reduce(tensor, op)\n            return tensor\n\n        def _do_all_gather(self, tensor: torch.Tensor) -> torch.Tensor:\n            if tensor.ndimension() == 0:\n                tensor = tensor.unsqueeze(0)\n            output = [torch.zeros_like(tensor) for _ in range(self.get_world_size())]\n            dist.all_gather(output, tensor)\n            return torch.cat(output, dim=0)\n\n        def _do_broadcast(self, tensor: torch.Tensor, src: int) -> torch.Tensor:\n            dist.broadcast(tensor, src=src)\n            return tensor\n\n        def barrier(self):\n            dist.barrier()\n", "path": "ignite/distributed/comp_models/native.py"}]}