We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[REQUEST] Remove deprecated PyTorch code
**Is your feature request related to a problem? Please describe.**
DeepSpeed uses code that's deprecated. This causes problems for libraries built on deepspeed when testing.

**Describe the solution you'd like**
```
/usr/local/lib/python3.10/dist-packages/deepspeed/comm/torch.py:171: in get_global_rank
    return torch.distributed.distributed_c10d._get_global_rank(group, group_rank)
```
should instead use `get_global_rank` as recommended by the warning
```
E       UserWarning: torch.distributed.distributed_c10d._get_global_rank is deprecated please use torch.distributed.distributed_c10d.get_global_rank instead
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `deepspeed/comm/__init__.py`
Content:
```
1 import torch
2 from .utils import *
3 from deepspeed import utils
4 
5 supported_torch_version = False
6 
7 # See more details at: https://github.com/pytorch/pytorch/pull/48767
8 # The PG API in torch versions lesser than 1.8 are different so it is
9 # non-trivial to support both in the same API. We will just use the
10 # DS comm. backend in deepspeed/comm/comm.py if torch version if 1.8+.
11 
12 if older_torch():
13     # Add custom deepspeed torch comm functions here since we can't import deepspeed.comm
14     # NOTE: We can't call torch.distributed directly here. Current hack is to import functions before calling them.
15     supported_torch_version = False
16     from torch.distributed import *
17 
18     def get_world_group():
19         return group.WORLD
20 
21     def get_global_rank(group, group_rank):
22         from torch.distributed.distributed_c10d import _get_global_rank
23         return _get_global_rank(group, group_rank)
24 
25     def allgather_fn(output_tensor, input_tensor, group=None, async_op=False):
26         from torch.distributed import all_gather, get_world_size
27         from torch import chunk
28         output_tensors = list(chunk(output_tensor, get_world_size(group)))
29         return all_gather(output_tensors, input_tensor, group=group, async_op=async_op)
30 
31     def reduce_scatter_fn(output_tensor, input_tensor, group=None, async_op=False):
32         from torch.distributed import reduce_scatter, get_world_size
33         from torch import chunk
34         input_tensor_lst = list(chunk(input_tensor, get_world_size(group)))
35         return reduce_scatter(output_tensor, input_tensor_lst, group=group)
36 
37     def configure(deepspeed_config=None,
38                   enabled=None,
39                   prof_all=None,
40                   prof_ops=None,
41                   verbose=None):
42         utils.logger.warn(
43             "Communication logging is not supported in torch versions older than 1.8")
44 
45 else:
46     supported_torch_version = True
47     from .comm import *
48
```

Path: `deepspeed/comm/torch.py`
Content:
```
1 '''
2 Copyright 2021 The Microsoft DeepSpeed Team
3 '''
4 from deepspeed import utils
5 
6 from .utils import *
7 from .backend import *
8 from .comm import *
9 
10 
11 class TorchBackend(Backend):
12     """
13         A light-weight wrapper class for torch.distributed API.
14         Only a subset of functions are wrapped. Once the init_process_group
15         is initialized, standard torch.distributed.* can be used directly
16         so no need to wrap all the functions. We can keep adding wrappers as
17         needed.
18     """
19     def __init__(self, backend, timeout, init_method, name='torch'):
20         super(TorchBackend, self).__init__()
21         self.torch_version_before_18 = older_torch()
22         self.has_allgather_base = has_allgather_base()
23         self.has_reduce_scatter_base = has_reduce_scatter_base()
24         self.initialized = True
25         self.name = name
26         # Future functionality to support ds.initialize() on a single GPU
27         # The idea is to fake that dist backend is initialized even when
28         # it is not so we can run on a single GPU without doing any init_process_group
29         self.single_gpu_mode = True
30         self.init_process_group(backend, timeout, init_method)
31 
32     def init_process_group(self, backend, timeout, init_method):
33         if not torch.distributed.is_initialized():
34             torch.distributed.init_process_group(backend,
35                                                  timeout=timeout,
36                                                  init_method=init_method)
37         self.using_mpi = torch.distributed.get_backend() == 'mpi'
38 
39     def all_reduce(self,
40                    tensor,
41                    op=torch.distributed.ReduceOp.SUM,
42                    group=None,
43                    async_op=False):
44         op = self._reduce_op(op)
45         return torch.distributed.all_reduce(tensor=tensor,
46                                             op=op,
47                                             group=group,
48                                             async_op=async_op)
49 
50     def reduce(self, tensor, dst, op=ReduceOp.SUM, group=None, async_op=False):
51         return torch.distributed.reduce(tensor=tensor,
52                                         dst=dst,
53                                         op=self._reduce_op(op),
54                                         group=group,
55                                         async_op=async_op)
56 
57     def reduce_scatter(self,
58                        output,
59                        input_list,
60                        op=ReduceOp.SUM,
61                        group=None,
62                        async_op=False):
63         return torch.distributed.reduce_scatter(output=output,
64                                                 input_list=input_list,
65                                                 op=self._reduce_op(op),
66                                                 group=group,
67                                                 async_op=async_op)
68 
69     def broadcast(self, tensor, src, group=None, async_op=False):
70         return torch.distributed.broadcast(tensor=tensor,
71                                            src=src,
72                                            group=group,
73                                            async_op=async_op)
74 
75     def all_gather(self, tensor_list, tensor, group=None, async_op=False):
76         return torch.distributed.all_gather(tensor_list=tensor_list,
77                                             tensor=tensor,
78                                             group=group,
79                                             async_op=async_op)
80 
81     def all_gather_base(self, output_tensor, input_tensor, group=None, async_op=False):
82         if self.has_allgather_base:
83             return torch.distributed.distributed_c10d._all_gather_base(
84                 output_tensor=output_tensor,
85                 input_tensor=input_tensor,
86                 group=group,
87                 async_op=async_op)
88         else:
89             utils.logger.warning(
90                 "unable to find torch.distributed._all_gather_base. will fall back to "
91                 "torch.distributed.reduce_scatter which will result in suboptimal performance. "
92                 "please consider upgrading your pytorch installation.")
93             pass
94 
95     def reduce_scatter_base(self,
96                             output_tensor,
97                             input_tensor,
98                             group=None,
99                             async_op=False):
100         if self.has_reduce_scatter_base:
101             return torch.distributed._reduce_scatter_base(output_tensor,
102                                                           input_tensor,
103                                                           group=group,
104                                                           async_op=async_op)
105         else:
106             utils.logger.warning(
107                 "unable to find torch.distributed._reduce_scatter_base. will fall back to "
108                 "torch.distributed.reduce_scatter which will result in suboptimal performance. "
109                 "please consider upgrading your pytorch installation.")
110             pass
111 
112     def all_to_all_single(self,
113                           output,
114                           input,
115                           output_split_sizes=None,
116                           input_split_sizes=None,
117                           group=None,
118                           async_op=False):
119         return torch.distributed.all_to_all_single(output=output,
120                                                    input=input,
121                                                    output_split_sizes=output_split_sizes,
122                                                    input_split_sizes=input_split_sizes,
123                                                    group=group,
124                                                    async_op=async_op)
125 
126     def send(self, tensor, dst, group=None, tag=0):
127         return torch.distributed.send(tensor=tensor, dst=dst, group=group, tag=tag)
128 
129     def recv(self, tensor, src=None, group=None, tag=0):
130         return torch.distributed.recv(tensor=tensor, src=src, group=group, tag=tag)
131 
132     def isend(self, tensor, dst, group=None, tag=0):
133         return torch.distributed.isend(tensor=tensor, dst=dst, group=group, tag=tag)
134 
135     def irecv(self, tensor, src=None, group=None, tag=0):
136         return torch.distributed.irecv(tensor=tensor, src=src, group=group, tag=tag)
137 
138     def gather(self, tensor, gather_list=None, dst=0, group=None, async_op=False):
139         return torch.distributed.gather(tensor=tensor,
140                                         gather_list=gather_list,
141                                         dst=dst,
142                                         group=group,
143                                         async_op=async_op)
144 
145     def scatter(self, tensor, scatter_list=None, src=0, group=None, async_op=False):
146         return torch.distributed.scatter(tensor=tensor,
147                                          scatter_list=scatter_list,
148                                          src=src,
149                                          group=group,
150                                          async_op=async_op)
151 
152     def barrier(self):
153         return torch.distributed.barrier()
154 
155     def get_rank(self, group=None):
156         return torch.distributed.get_rank(group=group)
157 
158     def get_world_size(self, group=None):
159         return torch.distributed.get_world_size(group=group)
160 
161     def is_initialized(self):
162         return torch.distributed.is_initialized()
163 
164     def get_backend(self, group=None):
165         return torch.distributed.get_backend(group=group)
166 
167     def new_group(self, ranks):
168         return torch.distributed.new_group(ranks)
169 
170     def get_global_rank(self, group, group_rank):
171         return torch.distributed.distributed_c10d._get_global_rank(group, group_rank)
172 
173     def get_world_group(self):
174         return torch.distributed.group.WORLD
175 
176     def destroy_process_group(self, group=None):
177         return torch.distributed.destroy_process_group(group=group)
178 
179     def _reduce_op(self, op):
180         '''
181             Helper function. If the op provided is not a torch.dist.ReduceOp, convert it and return
182         '''
183         if not isinstance(op, torch.distributed.ReduceOp):
184             if op == ReduceOp.SUM:
185                 op = torch.distributed.ReduceOp.SUM
186             elif op == ReduceOp.PRODUCT:
187                 op = torch.distributed.ReduceOp.PRODUCT
188             elif op == ReduceOp.AVG:
189                 op = torch.distributed.ReduceOp.AVG
190             elif op == ReduceOp.MIN:
191                 op = torch.distributed.ReduceOp.MIN
192             elif op == ReduceOp.MAX:
193                 op = torch.distributed.ReduceOp.MAX
194             elif op == ReduceOp.BAND:
195                 op = torch.distributed.ReduceOp.BAND
196             elif op == ReduceOp.BOR:
197                 op = torch.distributed.ReduceOp.BOR
198             elif op == ReduceOp.BXOR:
199                 op = torch.distributed.ReduceOp.BXOR
200         return op
201 
202 
203 # This will become a light-weight wrapper around torch.distributed functions
204 # TODO: create some example to show how this wrapper can help profile communication
205 # TODO: make sure there is no performance regression with this approach
206 # TODO: explore monkey-patching if this does not work
207
```
--- 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/deepspeed/comm/__init__.py b/deepspeed/comm/__init__.py
--- a/deepspeed/comm/__init__.py
+++ b/deepspeed/comm/__init__.py
@@ -19,7 +19,10 @@
         return group.WORLD
 
     def get_global_rank(group, group_rank):
-        from torch.distributed.distributed_c10d import _get_global_rank
+        if hasattr(torch.distributed.distributed_c10d, "get_global_rank"):
+            from torch.distributed.distributed_c10d import get_global_rank as _get_global_rank
+        else:
+            from torch.distributed.distributed_c10d import _get_global_rank
         return _get_global_rank(group, group_rank)
 
     def allgather_fn(output_tensor, input_tensor, group=None, async_op=False):
diff --git a/deepspeed/comm/torch.py b/deepspeed/comm/torch.py
--- a/deepspeed/comm/torch.py
+++ b/deepspeed/comm/torch.py
@@ -168,7 +168,11 @@
         return torch.distributed.new_group(ranks)
 
     def get_global_rank(self, group, group_rank):
-        return torch.distributed.distributed_c10d._get_global_rank(group, group_rank)
+        if hasattr(torch.distributed.distributed_c10d, "get_global_rank"):
+            from torch.distributed.distributed_c10d import get_global_rank as _get_global_rank
+        else:
+            from torch.distributed.distributed_c10d import _get_global_rank
+        return _get_global_rank(group, group_rank)
 
     def get_world_group(self):
         return torch.distributed.group.WORLD

{"golden_diff": "diff --git a/deepspeed/comm/__init__.py b/deepspeed/comm/__init__.py\n--- a/deepspeed/comm/__init__.py\n+++ b/deepspeed/comm/__init__.py\n@@ -19,7 +19,10 @@\n         return group.WORLD\n \n     def get_global_rank(group, group_rank):\n-        from torch.distributed.distributed_c10d import _get_global_rank\n+        if hasattr(torch.distributed.distributed_c10d, \"get_global_rank\"):\n+            from torch.distributed.distributed_c10d import get_global_rank as _get_global_rank\n+        else:\n+            from torch.distributed.distributed_c10d import _get_global_rank\n         return _get_global_rank(group, group_rank)\n \n     def allgather_fn(output_tensor, input_tensor, group=None, async_op=False):\ndiff --git a/deepspeed/comm/torch.py b/deepspeed/comm/torch.py\n--- a/deepspeed/comm/torch.py\n+++ b/deepspeed/comm/torch.py\n@@ -168,7 +168,11 @@\n         return torch.distributed.new_group(ranks)\n \n     def get_global_rank(self, group, group_rank):\n-        return torch.distributed.distributed_c10d._get_global_rank(group, group_rank)\n+        if hasattr(torch.distributed.distributed_c10d, \"get_global_rank\"):\n+            from torch.distributed.distributed_c10d import get_global_rank as _get_global_rank\n+        else:\n+            from torch.distributed.distributed_c10d import _get_global_rank\n+        return _get_global_rank(group, group_rank)\n \n     def get_world_group(self):\n         return torch.distributed.group.WORLD\n", "issue": "[REQUEST] Remove deprecated PyTorch code\n**Is your feature request related to a problem? Please describe.**\r\nDeepSpeed uses code that's deprecated. This causes problems for libraries built on deepspeed when testing.\r\n\r\n**Describe the solution you'd like**\r\n```\r\n/usr/local/lib/python3.10/dist-packages/deepspeed/comm/torch.py:171: in get_global_rank\r\n    return torch.distributed.distributed_c10d._get_global_rank(group, group_rank)\r\n```\r\nshould instead use `get_global_rank` as recommended by the warning\r\n```\r\nE       UserWarning: torch.distributed.distributed_c10d._get_global_rank is deprecated please use torch.distributed.distributed_c10d.get_global_rank instead\r\n```\r\n\n", "before_files": [{"content": "import torch\nfrom .utils import *\nfrom deepspeed import utils\n\nsupported_torch_version = False\n\n# See more details at: https://github.com/pytorch/pytorch/pull/48767\n# The PG API in torch versions lesser than 1.8 are different so it is\n# non-trivial to support both in the same API. We will just use the\n# DS comm. backend in deepspeed/comm/comm.py if torch version if 1.8+.\n\nif older_torch():\n    # Add custom deepspeed torch comm functions here since we can't import deepspeed.comm\n    # NOTE: We can't call torch.distributed directly here. Current hack is to import functions before calling them.\n    supported_torch_version = False\n    from torch.distributed import *\n\n    def get_world_group():\n        return group.WORLD\n\n    def get_global_rank(group, group_rank):\n        from torch.distributed.distributed_c10d import _get_global_rank\n        return _get_global_rank(group, group_rank)\n\n    def allgather_fn(output_tensor, input_tensor, group=None, async_op=False):\n        from torch.distributed import all_gather, get_world_size\n        from torch import chunk\n        output_tensors = list(chunk(output_tensor, get_world_size(group)))\n        return all_gather(output_tensors, input_tensor, group=group, async_op=async_op)\n\n    def reduce_scatter_fn(output_tensor, input_tensor, group=None, async_op=False):\n        from torch.distributed import reduce_scatter, get_world_size\n        from torch import chunk\n        input_tensor_lst = list(chunk(input_tensor, get_world_size(group)))\n        return reduce_scatter(output_tensor, input_tensor_lst, group=group)\n\n    def configure(deepspeed_config=None,\n                  enabled=None,\n                  prof_all=None,\n                  prof_ops=None,\n                  verbose=None):\n        utils.logger.warn(\n            \"Communication logging is not supported in torch versions older than 1.8\")\n\nelse:\n    supported_torch_version = True\n    from .comm import *\n", "path": "deepspeed/comm/__init__.py"}, {"content": "'''\nCopyright 2021 The Microsoft DeepSpeed Team\n'''\nfrom deepspeed import utils\n\nfrom .utils import *\nfrom .backend import *\nfrom .comm import *\n\n\nclass TorchBackend(Backend):\n    \"\"\"\n        A light-weight wrapper class for torch.distributed API.\n        Only a subset of functions are wrapped. Once the init_process_group\n        is initialized, standard torch.distributed.* can be used directly\n        so no need to wrap all the functions. We can keep adding wrappers as\n        needed.\n    \"\"\"\n    def __init__(self, backend, timeout, init_method, name='torch'):\n        super(TorchBackend, self).__init__()\n        self.torch_version_before_18 = older_torch()\n        self.has_allgather_base = has_allgather_base()\n        self.has_reduce_scatter_base = has_reduce_scatter_base()\n        self.initialized = True\n        self.name = name\n        # Future functionality to support ds.initialize() on a single GPU\n        # The idea is to fake that dist backend is initialized even when\n        # it is not so we can run on a single GPU without doing any init_process_group\n        self.single_gpu_mode = True\n        self.init_process_group(backend, timeout, init_method)\n\n    def init_process_group(self, backend, timeout, init_method):\n        if not torch.distributed.is_initialized():\n            torch.distributed.init_process_group(backend,\n                                                 timeout=timeout,\n                                                 init_method=init_method)\n        self.using_mpi = torch.distributed.get_backend() == 'mpi'\n\n    def all_reduce(self,\n                   tensor,\n                   op=torch.distributed.ReduceOp.SUM,\n                   group=None,\n                   async_op=False):\n        op = self._reduce_op(op)\n        return torch.distributed.all_reduce(tensor=tensor,\n                                            op=op,\n                                            group=group,\n                                            async_op=async_op)\n\n    def reduce(self, tensor, dst, op=ReduceOp.SUM, group=None, async_op=False):\n        return torch.distributed.reduce(tensor=tensor,\n                                        dst=dst,\n                                        op=self._reduce_op(op),\n                                        group=group,\n                                        async_op=async_op)\n\n    def reduce_scatter(self,\n                       output,\n                       input_list,\n                       op=ReduceOp.SUM,\n                       group=None,\n                       async_op=False):\n        return torch.distributed.reduce_scatter(output=output,\n                                                input_list=input_list,\n                                                op=self._reduce_op(op),\n                                                group=group,\n                                                async_op=async_op)\n\n    def broadcast(self, tensor, src, group=None, async_op=False):\n        return torch.distributed.broadcast(tensor=tensor,\n                                           src=src,\n                                           group=group,\n                                           async_op=async_op)\n\n    def all_gather(self, tensor_list, tensor, group=None, async_op=False):\n        return torch.distributed.all_gather(tensor_list=tensor_list,\n                                            tensor=tensor,\n                                            group=group,\n                                            async_op=async_op)\n\n    def all_gather_base(self, output_tensor, input_tensor, group=None, async_op=False):\n        if self.has_allgather_base:\n            return torch.distributed.distributed_c10d._all_gather_base(\n                output_tensor=output_tensor,\n                input_tensor=input_tensor,\n                group=group,\n                async_op=async_op)\n        else:\n            utils.logger.warning(\n                \"unable to find torch.distributed._all_gather_base. will fall back to \"\n                \"torch.distributed.reduce_scatter which will result in suboptimal performance. \"\n                \"please consider upgrading your pytorch installation.\")\n            pass\n\n    def reduce_scatter_base(self,\n                            output_tensor,\n                            input_tensor,\n                            group=None,\n                            async_op=False):\n        if self.has_reduce_scatter_base:\n            return torch.distributed._reduce_scatter_base(output_tensor,\n                                                          input_tensor,\n                                                          group=group,\n                                                          async_op=async_op)\n        else:\n            utils.logger.warning(\n                \"unable to find torch.distributed._reduce_scatter_base. will fall back to \"\n                \"torch.distributed.reduce_scatter which will result in suboptimal performance. \"\n                \"please consider upgrading your pytorch installation.\")\n            pass\n\n    def all_to_all_single(self,\n                          output,\n                          input,\n                          output_split_sizes=None,\n                          input_split_sizes=None,\n                          group=None,\n                          async_op=False):\n        return torch.distributed.all_to_all_single(output=output,\n                                                   input=input,\n                                                   output_split_sizes=output_split_sizes,\n                                                   input_split_sizes=input_split_sizes,\n                                                   group=group,\n                                                   async_op=async_op)\n\n    def send(self, tensor, dst, group=None, tag=0):\n        return torch.distributed.send(tensor=tensor, dst=dst, group=group, tag=tag)\n\n    def recv(self, tensor, src=None, group=None, tag=0):\n        return torch.distributed.recv(tensor=tensor, src=src, group=group, tag=tag)\n\n    def isend(self, tensor, dst, group=None, tag=0):\n        return torch.distributed.isend(tensor=tensor, dst=dst, group=group, tag=tag)\n\n    def irecv(self, tensor, src=None, group=None, tag=0):\n        return torch.distributed.irecv(tensor=tensor, src=src, group=group, tag=tag)\n\n    def gather(self, tensor, gather_list=None, dst=0, group=None, async_op=False):\n        return torch.distributed.gather(tensor=tensor,\n                                        gather_list=gather_list,\n                                        dst=dst,\n                                        group=group,\n                                        async_op=async_op)\n\n    def scatter(self, tensor, scatter_list=None, src=0, group=None, async_op=False):\n        return torch.distributed.scatter(tensor=tensor,\n                                         scatter_list=scatter_list,\n                                         src=src,\n                                         group=group,\n                                         async_op=async_op)\n\n    def barrier(self):\n        return torch.distributed.barrier()\n\n    def get_rank(self, group=None):\n        return torch.distributed.get_rank(group=group)\n\n    def get_world_size(self, group=None):\n        return torch.distributed.get_world_size(group=group)\n\n    def is_initialized(self):\n        return torch.distributed.is_initialized()\n\n    def get_backend(self, group=None):\n        return torch.distributed.get_backend(group=group)\n\n    def new_group(self, ranks):\n        return torch.distributed.new_group(ranks)\n\n    def get_global_rank(self, group, group_rank):\n        return torch.distributed.distributed_c10d._get_global_rank(group, group_rank)\n\n    def get_world_group(self):\n        return torch.distributed.group.WORLD\n\n    def destroy_process_group(self, group=None):\n        return torch.distributed.destroy_process_group(group=group)\n\n    def _reduce_op(self, op):\n        '''\n            Helper function. If the op provided is not a torch.dist.ReduceOp, convert it and return\n        '''\n        if not isinstance(op, torch.distributed.ReduceOp):\n            if op == ReduceOp.SUM:\n                op = torch.distributed.ReduceOp.SUM\n            elif op == ReduceOp.PRODUCT:\n                op = torch.distributed.ReduceOp.PRODUCT\n            elif op == ReduceOp.AVG:\n                op = torch.distributed.ReduceOp.AVG\n            elif op == ReduceOp.MIN:\n                op = torch.distributed.ReduceOp.MIN\n            elif op == ReduceOp.MAX:\n                op = torch.distributed.ReduceOp.MAX\n            elif op == ReduceOp.BAND:\n                op = torch.distributed.ReduceOp.BAND\n            elif op == ReduceOp.BOR:\n                op = torch.distributed.ReduceOp.BOR\n            elif op == ReduceOp.BXOR:\n                op = torch.distributed.ReduceOp.BXOR\n        return op\n\n\n# This will become a light-weight wrapper around torch.distributed functions\n# TODO: create some example to show how this wrapper can help profile communication\n# TODO: make sure there is no performance regression with this approach\n# TODO: explore monkey-patching if this does not work\n", "path": "deepspeed/comm/torch.py"}], "after_files": [{"content": "import torch\nfrom .utils import *\nfrom deepspeed import utils\n\nsupported_torch_version = False\n\n# See more details at: https://github.com/pytorch/pytorch/pull/48767\n# The PG API in torch versions lesser than 1.8 are different so it is\n# non-trivial to support both in the same API. We will just use the\n# DS comm. backend in deepspeed/comm/comm.py if torch version if 1.8+.\n\nif older_torch():\n    # Add custom deepspeed torch comm functions here since we can't import deepspeed.comm\n    # NOTE: We can't call torch.distributed directly here. Current hack is to import functions before calling them.\n    supported_torch_version = False\n    from torch.distributed import *\n\n    def get_world_group():\n        return group.WORLD\n\n    def get_global_rank(group, group_rank):\n        if hasattr(torch.distributed.distributed_c10d, \"get_global_rank\"):\n            from torch.distributed.distributed_c10d import get_global_rank as _get_global_rank\n        else:\n            from torch.distributed.distributed_c10d import _get_global_rank\n        return _get_global_rank(group, group_rank)\n\n    def allgather_fn(output_tensor, input_tensor, group=None, async_op=False):\n        from torch.distributed import all_gather, get_world_size\n        from torch import chunk\n        output_tensors = list(chunk(output_tensor, get_world_size(group)))\n        return all_gather(output_tensors, input_tensor, group=group, async_op=async_op)\n\n    def reduce_scatter_fn(output_tensor, input_tensor, group=None, async_op=False):\n        from torch.distributed import reduce_scatter, get_world_size\n        from torch import chunk\n        input_tensor_lst = list(chunk(input_tensor, get_world_size(group)))\n        return reduce_scatter(output_tensor, input_tensor_lst, group=group)\n\n    def configure(deepspeed_config=None,\n                  enabled=None,\n                  prof_all=None,\n                  prof_ops=None,\n                  verbose=None):\n        utils.logger.warn(\n            \"Communication logging is not supported in torch versions older than 1.8\")\n\nelse:\n    supported_torch_version = True\n    from .comm import *\n", "path": "deepspeed/comm/__init__.py"}, {"content": "'''\nCopyright 2021 The Microsoft DeepSpeed Team\n'''\nfrom deepspeed import utils\n\nfrom .utils import *\nfrom .backend import *\nfrom .comm import *\n\n\nclass TorchBackend(Backend):\n    \"\"\"\n        A light-weight wrapper class for torch.distributed API.\n        Only a subset of functions are wrapped. Once the init_process_group\n        is initialized, standard torch.distributed.* can be used directly\n        so no need to wrap all the functions. We can keep adding wrappers as\n        needed.\n    \"\"\"\n    def __init__(self, backend, timeout, init_method, name='torch'):\n        super(TorchBackend, self).__init__()\n        self.torch_version_before_18 = older_torch()\n        self.has_allgather_base = has_allgather_base()\n        self.has_reduce_scatter_base = has_reduce_scatter_base()\n        self.initialized = True\n        self.name = name\n        # Future functionality to support ds.initialize() on a single GPU\n        # The idea is to fake that dist backend is initialized even when\n        # it is not so we can run on a single GPU without doing any init_process_group\n        self.single_gpu_mode = True\n        self.init_process_group(backend, timeout, init_method)\n\n    def init_process_group(self, backend, timeout, init_method):\n        if not torch.distributed.is_initialized():\n            torch.distributed.init_process_group(backend,\n                                                 timeout=timeout,\n                                                 init_method=init_method)\n        self.using_mpi = torch.distributed.get_backend() == 'mpi'\n\n    def all_reduce(self,\n                   tensor,\n                   op=torch.distributed.ReduceOp.SUM,\n                   group=None,\n                   async_op=False):\n        op = self._reduce_op(op)\n        return torch.distributed.all_reduce(tensor=tensor,\n                                            op=op,\n                                            group=group,\n                                            async_op=async_op)\n\n    def reduce(self, tensor, dst, op=ReduceOp.SUM, group=None, async_op=False):\n        return torch.distributed.reduce(tensor=tensor,\n                                        dst=dst,\n                                        op=self._reduce_op(op),\n                                        group=group,\n                                        async_op=async_op)\n\n    def reduce_scatter(self,\n                       output,\n                       input_list,\n                       op=ReduceOp.SUM,\n                       group=None,\n                       async_op=False):\n        return torch.distributed.reduce_scatter(output=output,\n                                                input_list=input_list,\n                                                op=self._reduce_op(op),\n                                                group=group,\n                                                async_op=async_op)\n\n    def broadcast(self, tensor, src, group=None, async_op=False):\n        return torch.distributed.broadcast(tensor=tensor,\n                                           src=src,\n                                           group=group,\n                                           async_op=async_op)\n\n    def all_gather(self, tensor_list, tensor, group=None, async_op=False):\n        return torch.distributed.all_gather(tensor_list=tensor_list,\n                                            tensor=tensor,\n                                            group=group,\n                                            async_op=async_op)\n\n    def all_gather_base(self, output_tensor, input_tensor, group=None, async_op=False):\n        if self.has_allgather_base:\n            return torch.distributed.distributed_c10d._all_gather_base(\n                output_tensor=output_tensor,\n                input_tensor=input_tensor,\n                group=group,\n                async_op=async_op)\n        else:\n            utils.logger.warning(\n                \"unable to find torch.distributed._all_gather_base. will fall back to \"\n                \"torch.distributed.reduce_scatter which will result in suboptimal performance. \"\n                \"please consider upgrading your pytorch installation.\")\n            pass\n\n    def reduce_scatter_base(self,\n                            output_tensor,\n                            input_tensor,\n                            group=None,\n                            async_op=False):\n        if self.has_reduce_scatter_base:\n            return torch.distributed._reduce_scatter_base(output_tensor,\n                                                          input_tensor,\n                                                          group=group,\n                                                          async_op=async_op)\n        else:\n            utils.logger.warning(\n                \"unable to find torch.distributed._reduce_scatter_base. will fall back to \"\n                \"torch.distributed.reduce_scatter which will result in suboptimal performance. \"\n                \"please consider upgrading your pytorch installation.\")\n            pass\n\n    def all_to_all_single(self,\n                          output,\n                          input,\n                          output_split_sizes=None,\n                          input_split_sizes=None,\n                          group=None,\n                          async_op=False):\n        return torch.distributed.all_to_all_single(output=output,\n                                                   input=input,\n                                                   output_split_sizes=output_split_sizes,\n                                                   input_split_sizes=input_split_sizes,\n                                                   group=group,\n                                                   async_op=async_op)\n\n    def send(self, tensor, dst, group=None, tag=0):\n        return torch.distributed.send(tensor=tensor, dst=dst, group=group, tag=tag)\n\n    def recv(self, tensor, src=None, group=None, tag=0):\n        return torch.distributed.recv(tensor=tensor, src=src, group=group, tag=tag)\n\n    def isend(self, tensor, dst, group=None, tag=0):\n        return torch.distributed.isend(tensor=tensor, dst=dst, group=group, tag=tag)\n\n    def irecv(self, tensor, src=None, group=None, tag=0):\n        return torch.distributed.irecv(tensor=tensor, src=src, group=group, tag=tag)\n\n    def gather(self, tensor, gather_list=None, dst=0, group=None, async_op=False):\n        return torch.distributed.gather(tensor=tensor,\n                                        gather_list=gather_list,\n                                        dst=dst,\n                                        group=group,\n                                        async_op=async_op)\n\n    def scatter(self, tensor, scatter_list=None, src=0, group=None, async_op=False):\n        return torch.distributed.scatter(tensor=tensor,\n                                         scatter_list=scatter_list,\n                                         src=src,\n                                         group=group,\n                                         async_op=async_op)\n\n    def barrier(self):\n        return torch.distributed.barrier()\n\n    def get_rank(self, group=None):\n        return torch.distributed.get_rank(group=group)\n\n    def get_world_size(self, group=None):\n        return torch.distributed.get_world_size(group=group)\n\n    def is_initialized(self):\n        return torch.distributed.is_initialized()\n\n    def get_backend(self, group=None):\n        return torch.distributed.get_backend(group=group)\n\n    def new_group(self, ranks):\n        return torch.distributed.new_group(ranks)\n\n    def get_global_rank(self, group, group_rank):\n        if hasattr(torch.distributed.distributed_c10d, \"get_global_rank\"):\n            from torch.distributed.distributed_c10d import get_global_rank as _get_global_rank\n        else:\n            from torch.distributed.distributed_c10d import _get_global_rank\n        return _get_global_rank(group, group_rank)\n\n    def get_world_group(self):\n        return torch.distributed.group.WORLD\n\n    def destroy_process_group(self, group=None):\n        return torch.distributed.destroy_process_group(group=group)\n\n    def _reduce_op(self, op):\n        '''\n            Helper function. If the op provided is not a torch.dist.ReduceOp, convert it and return\n        '''\n        if not isinstance(op, torch.distributed.ReduceOp):\n            if op == ReduceOp.SUM:\n                op = torch.distributed.ReduceOp.SUM\n            elif op == ReduceOp.PRODUCT:\n                op = torch.distributed.ReduceOp.PRODUCT\n            elif op == ReduceOp.AVG:\n                op = torch.distributed.ReduceOp.AVG\n            elif op == ReduceOp.MIN:\n                op = torch.distributed.ReduceOp.MIN\n            elif op == ReduceOp.MAX:\n                op = torch.distributed.ReduceOp.MAX\n            elif op == ReduceOp.BAND:\n                op = torch.distributed.ReduceOp.BAND\n            elif op == ReduceOp.BOR:\n                op = torch.distributed.ReduceOp.BOR\n            elif op == ReduceOp.BXOR:\n                op = torch.distributed.ReduceOp.BXOR\n        return op\n\n\n# This will become a light-weight wrapper around torch.distributed functions\n# TODO: create some example to show how this wrapper can help profile communication\n# TODO: make sure there is no performance regression with this approach\n# TODO: explore monkey-patching if this does not work\n", "path": "deepspeed/comm/torch.py"}]}