We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
DistributedOptimizer cannot be used when underlying optimizer overwrites compute_gradients and apply_gradients
In our use case we have a TensorFlow optimizer that explicitly defines `compute_gradients` and `apply_gradients` functions. The current version of `DistributedOptimizer` cannot be used in our case because it does not overwrite `apply_gradients` to use` self._optimizer.apply_gradients` and it calls (lines 178-179 of `horovod/tensorflow/__init__.py`)
```python
gradients = super(DistributedOptimizer, self).compute_gradients(*args, **kwargs)
```
instead of `gradients = self._optimizer.compute_gradients(*args, **kwargs)`.

Is it possible to update `DistributedOptimizer` to work in our use case and similar situations when `compute_gradients` and `apply_gradients` are redefined? As far as I can see simply changing lines 178-179 of `horovod/tensorflow/__init__.py` to 
```python
gradients = self._optimizer.compute_gradients(*args, **kwargs)
```
and adding additional method:
```python
  def apply_gradients(self, grads_and_vars, global_step=None, name=None):
    """Calls this same method on the underlying optimizer."""
    return self._optimizer.apply_gradients(grads_and_vars, global_step, name)
```
should make it work.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `horovod/__init__.py`
Content:
```
1 __version__ = '0.13.0'
2
```

Path: `horovod/tensorflow/__init__.py`
Content:
```
1 # Copyright 2016 The TensorFlow Authors. All Rights Reserved.
2 # Modifications copyright (C) 2017 Uber Technologies, Inc.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 # ==============================================================================
16 # pylint: disable=g-short-docstring-punctuation
17 """## Communicating Between Processes with MPI
18 
19 TensorFlow natively provides inter-device communication through send and
20 receive ops and inter-node communication through Distributed TensorFlow, based
21 on the same send and receive abstractions. On HPC clusters where Infiniband or
22 other high-speed node interconnects are available, these can end up being
23 insufficient for synchronous data-parallel training (without asynchronous
24 gradient descent). This module implements a variety of MPI ops which can take
25 advantage of hardware-specific MPI libraries for efficient communication.
26 """
27 
28 from __future__ import absolute_import
29 from __future__ import division
30 from __future__ import print_function
31 
32 from horovod.common import init
33 from horovod.common import size
34 from horovod.common import local_size
35 from horovod.common import rank
36 from horovod.common import local_rank
37 from horovod.common import mpi_threads_supported
38 from horovod.common import check_extension
39 
40 check_extension('horovod.tensorflow', 'HOROVOD_WITH_TENSORFLOW', __file__, 'mpi_lib')
41 
42 from horovod.tensorflow.mpi_ops import allgather
43 from horovod.tensorflow.mpi_ops import broadcast
44 from horovod.tensorflow.mpi_ops import _allreduce
45 
46 import tensorflow as tf
47 
48 
49 def allreduce(tensor, average=True, device_dense='', device_sparse=''):
50     """Perform an allreduce on a tf.Tensor or tf.IndexedSlices.
51 
52     Arguments:
53         tensor: tf.Tensor, tf.Variable, or tf.IndexedSlices to reduce.
54         The shape of the input must be identical across all ranks.
55         average: If True, computes the average over all ranks.
56                  Otherwise, computes the sum over all ranks.
57         device_dense: Device to be used for dense tensors. Uses GPU by default
58                       if Horovod was build with HOROVOD_GPU_ALLREDUCE.
59         device_sparse: Device to be used for sparse tensors. Uses GPU by default
60                        if Horovod was build with HOROVOD_GPU_ALLGATHER.
61 
62     This function performs a bandwidth-optimal ring allreduce on the input
63     tensor. If the input is an tf.IndexedSlices, the function instead does an
64     allgather on the values and the indices, effectively doing an allreduce on
65     the represented tensor.
66     """
67     if isinstance(tensor, tf.IndexedSlices):
68         with tf.device(device_sparse):
69             # For IndexedSlices, do two allgathers intead of an allreduce.
70             horovod_size = tf.cast(size(), tensor.values.dtype)
71             values = allgather(tensor.values)
72             indices = allgather(tensor.indices)
73 
74             # To make this operation into an average, divide all gathered values by
75             # the Horovod size.
76             new_values = tf.div(values, horovod_size) if average else values
77         return tf.IndexedSlices(new_values, indices,
78                                 dense_shape=tensor.dense_shape)
79     else:
80         with tf.device(device_dense):
81             horovod_size = tf.cast(size(), tensor.dtype)
82             summed_tensor = _allreduce(tensor)
83             new_tensor = (tf.div(summed_tensor, horovod_size)
84                           if average else summed_tensor)
85         return new_tensor
86 
87 
88 def broadcast_global_variables(root_rank):
89     """Broadcasts all global variables from root rank to all other processes.
90 
91     Arguments:
92         root_rank: rank of the process from which global variables will be broadcasted
93         to all other processes.
94     """
95     return tf.group(*[tf.assign(var, broadcast(var, root_rank))
96                       for var in tf.global_variables()])
97 
98 
99 class BroadcastGlobalVariablesHook(tf.train.SessionRunHook):
100     """
101     SessionRunHook that will broadcast all global variables from root rank
102     to all other processes during initialization.
103 
104     This is necessary to ensure consistent initialization of all workers when
105     training is started with random weights or restored from a checkpoint.
106     """
107 
108     def __init__(self, root_rank, device=''):
109         """Construct a new BroadcastGlobalVariablesHook that will broadcast all
110         global variables from root rank to all other processes during initialization.
111 
112         Args:
113           root_rank:
114             Rank that will send data, other ranks will receive data.
115           device:
116             Device to be used for broadcasting. Uses GPU by default
117             if Horovod was build with HOROVOD_GPU_BROADCAST.
118         """
119         super(BroadcastGlobalVariablesHook, self).__init__()
120         self.root_rank = root_rank
121         self.bcast_op = None
122         self.device = device
123 
124     def begin(self):
125         if not self.bcast_op or self.bcast_op.graph != tf.get_default_graph():
126             with tf.device(self.device):
127                 self.bcast_op = broadcast_global_variables(self.root_rank)
128 
129     def after_create_session(self, session, coord):
130         session.run(self.bcast_op)
131 
132 
133 class DistributedOptimizer(tf.train.Optimizer):
134     """An optimizer that wraps another tf.Optimizer, using an allreduce to
135     average gradient values before applying gradients to model weights."""
136 
137     def __init__(self, optimizer, name=None, use_locking=False, device_dense='',
138                  device_sparse=''):
139         """Construct a new DistributedOptimizer, which uses another optimizer
140         under the hood for computing single-process gradient values and
141         applying gradient updates after the gradient values have been averaged
142         across all the Horovod ranks.
143 
144         Args:
145           optimizer:
146             Optimizer to use for computing gradients and applying updates.
147           name:
148             Optional name prefix for the operations created when applying
149             gradients. Defaults to "Distributed" followed by the provided
150             optimizer type.
151           use_locking:
152             Whether to use locking when updating variables.
153             See Optimizer.__init__ for more info.
154           device_dense:
155             Device to be used for dense tensors. Uses GPU by default
156             if Horovod was build with HOROVOD_GPU_ALLREDUCE.
157           device_sparse:
158             Device to be used for sparse tensors. Uses GPU by default
159             if Horovod was build with HOROVOD_GPU_ALLGATHER.
160         """
161         if name is None:
162             name = "Distributed{}".format(type(optimizer).__name__)
163 
164         self._optimizer = optimizer
165         self._device_dense = device_dense
166         self._device_sparse = device_sparse
167         super(DistributedOptimizer, self).__init__(
168             name=name, use_locking=use_locking)
169 
170     def compute_gradients(self, *args, **kwargs):
171         """Compute gradients of all trainable variables.
172 
173         See Optimizer.compute_gradients() for more info.
174 
175         In DistributedOptimizer, compute_gradients() is overriden to also
176         allreduce the gradients before returning them.
177         """
178         gradients = (super(DistributedOptimizer, self)
179                      .compute_gradients(*args, **kwargs))
180         if size() > 1:
181             averaged_gradients = []
182             with tf.name_scope(self._name + "_Allreduce"):
183                 for grad, var in gradients:
184                     if grad is not None:
185                         avg_grad = allreduce(grad, device_dense=self._device_dense,
186                                              device_sparse=self._device_sparse)
187                         averaged_gradients.append((avg_grad, var))
188                     else:
189                         averaged_gradients.append((None, var))
190             return averaged_gradients
191         else:
192             return gradients
193 
194     def _apply_dense(self, *args, **kwargs):
195         """Calls this same method on the underlying optimizer."""
196         return self._optimizer._apply_dense(*args, **kwargs)
197 
198     def _resource_apply_dense(self, *args, **kwargs):
199         """Calls this same method on the underlying optimizer."""
200         return self._optimizer._resource_apply_dense(*args, **kwargs)
201 
202     def _resource_apply_sparse_duplicate_indices(self, *args, **kwargs):
203         """Calls this same method on the underlying optimizer."""
204         return self._optimizer._resource_apply_sparse_duplicate_indices(*args, **kwargs)
205 
206     def _resource_apply_sparse(self, *args, **kwargs):
207         """Calls this same method on the underlying optimizer."""
208         return self._optimizer._resource_apply_sparse(*args, **kwargs)
209 
210     def _apply_sparse_duplicate_indices(self, *args, **kwargs):
211         """Calls this same method on the underlying optimizer."""
212         return self._optimizer._apply_sparse_duplicate_indices(*args, **kwargs)
213 
214     def _apply_sparse(self, *args, **kwargs):
215         """Calls this same method on the underlying optimizer."""
216         return self._optimizer._apply_sparse(*args, **kwargs)
217 
218     def _prepare(self, *args, **kwargs):
219         """Calls this same method on the underlying optimizer."""
220         return self._optimizer._prepare(*args, **kwargs)
221 
222     def _create_slots(self, *args, **kwargs):
223         """Calls this same method on the underlying optimizer."""
224         return self._optimizer._create_slots(*args, **kwargs)
225 
226     def _valid_dtypes(self, *args, **kwargs):
227         """Calls this same method on the underlying optimizer."""
228         return self._optimizer._valid_dtypes(*args, **kwargs)
229 
230     def _finish(self, *args, **kwargs):
231         """Calls this same method on the underlying optimizer."""
232         return self._optimizer._finish(*args, **kwargs)
233
```
--- 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/horovod/__init__.py b/horovod/__init__.py
--- a/horovod/__init__.py
+++ b/horovod/__init__.py
@@ -1 +1 @@
-__version__ = '0.13.0'
+__version__ = '0.13.1'
diff --git a/horovod/tensorflow/__init__.py b/horovod/tensorflow/__init__.py
--- a/horovod/tensorflow/__init__.py
+++ b/horovod/tensorflow/__init__.py
@@ -175,8 +175,7 @@
         In DistributedOptimizer, compute_gradients() is overriden to also
         allreduce the gradients before returning them.
         """
-        gradients = (super(DistributedOptimizer, self)
-                     .compute_gradients(*args, **kwargs))
+        gradients = self._optimizer.compute_gradients(*args, **kwargs)
         if size() > 1:
             averaged_gradients = []
             with tf.name_scope(self._name + "_Allreduce"):
@@ -191,42 +190,18 @@
         else:
             return gradients
 
-    def _apply_dense(self, *args, **kwargs):
+    def apply_gradients(self, *args, **kwargs):
         """Calls this same method on the underlying optimizer."""
-        return self._optimizer._apply_dense(*args, **kwargs)
+        return self._optimizer.apply_gradients(*args, **kwargs)
 
-    def _resource_apply_dense(self, *args, **kwargs):
+    def get_slot(self, *args, **kwargs):
         """Calls this same method on the underlying optimizer."""
-        return self._optimizer._resource_apply_dense(*args, **kwargs)
+        return self._optimizer.get_slot(*args, **kwargs)
 
-    def _resource_apply_sparse_duplicate_indices(self, *args, **kwargs):
+    def get_slot_names(self, *args, **kwargs):
         """Calls this same method on the underlying optimizer."""
-        return self._optimizer._resource_apply_sparse_duplicate_indices(*args, **kwargs)
+        return self._optimizer.get_slot_names(*args, **kwargs)
 
-    def _resource_apply_sparse(self, *args, **kwargs):
+    def variables(self, *args, **kwargs):
         """Calls this same method on the underlying optimizer."""
-        return self._optimizer._resource_apply_sparse(*args, **kwargs)
-
-    def _apply_sparse_duplicate_indices(self, *args, **kwargs):
-        """Calls this same method on the underlying optimizer."""
-        return self._optimizer._apply_sparse_duplicate_indices(*args, **kwargs)
-
-    def _apply_sparse(self, *args, **kwargs):
-        """Calls this same method on the underlying optimizer."""
-        return self._optimizer._apply_sparse(*args, **kwargs)
-
-    def _prepare(self, *args, **kwargs):
-        """Calls this same method on the underlying optimizer."""
-        return self._optimizer._prepare(*args, **kwargs)
-
-    def _create_slots(self, *args, **kwargs):
-        """Calls this same method on the underlying optimizer."""
-        return self._optimizer._create_slots(*args, **kwargs)
-
-    def _valid_dtypes(self, *args, **kwargs):
-        """Calls this same method on the underlying optimizer."""
-        return self._optimizer._valid_dtypes(*args, **kwargs)
-
-    def _finish(self, *args, **kwargs):
-        """Calls this same method on the underlying optimizer."""
-        return self._optimizer._finish(*args, **kwargs)
+        return self._optimizer.variables(*args, **kwargs)

{"golden_diff": "diff --git a/horovod/__init__.py b/horovod/__init__.py\n--- a/horovod/__init__.py\n+++ b/horovod/__init__.py\n@@ -1 +1 @@\n-__version__ = '0.13.0'\n+__version__ = '0.13.1'\ndiff --git a/horovod/tensorflow/__init__.py b/horovod/tensorflow/__init__.py\n--- a/horovod/tensorflow/__init__.py\n+++ b/horovod/tensorflow/__init__.py\n@@ -175,8 +175,7 @@\n         In DistributedOptimizer, compute_gradients() is overriden to also\n         allreduce the gradients before returning them.\n         \"\"\"\n-        gradients = (super(DistributedOptimizer, self)\n-                     .compute_gradients(*args, **kwargs))\n+        gradients = self._optimizer.compute_gradients(*args, **kwargs)\n         if size() > 1:\n             averaged_gradients = []\n             with tf.name_scope(self._name + \"_Allreduce\"):\n@@ -191,42 +190,18 @@\n         else:\n             return gradients\n \n-    def _apply_dense(self, *args, **kwargs):\n+    def apply_gradients(self, *args, **kwargs):\n         \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._apply_dense(*args, **kwargs)\n+        return self._optimizer.apply_gradients(*args, **kwargs)\n \n-    def _resource_apply_dense(self, *args, **kwargs):\n+    def get_slot(self, *args, **kwargs):\n         \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._resource_apply_dense(*args, **kwargs)\n+        return self._optimizer.get_slot(*args, **kwargs)\n \n-    def _resource_apply_sparse_duplicate_indices(self, *args, **kwargs):\n+    def get_slot_names(self, *args, **kwargs):\n         \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._resource_apply_sparse_duplicate_indices(*args, **kwargs)\n+        return self._optimizer.get_slot_names(*args, **kwargs)\n \n-    def _resource_apply_sparse(self, *args, **kwargs):\n+    def variables(self, *args, **kwargs):\n         \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._resource_apply_sparse(*args, **kwargs)\n-\n-    def _apply_sparse_duplicate_indices(self, *args, **kwargs):\n-        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._apply_sparse_duplicate_indices(*args, **kwargs)\n-\n-    def _apply_sparse(self, *args, **kwargs):\n-        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._apply_sparse(*args, **kwargs)\n-\n-    def _prepare(self, *args, **kwargs):\n-        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._prepare(*args, **kwargs)\n-\n-    def _create_slots(self, *args, **kwargs):\n-        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._create_slots(*args, **kwargs)\n-\n-    def _valid_dtypes(self, *args, **kwargs):\n-        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._valid_dtypes(*args, **kwargs)\n-\n-    def _finish(self, *args, **kwargs):\n-        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n-        return self._optimizer._finish(*args, **kwargs)\n+        return self._optimizer.variables(*args, **kwargs)\n", "issue": "DistributedOptimizer cannot be used when underlying optimizer overwrites compute_gradients and apply_gradients\nIn our use case we have a TensorFlow optimizer that explicitly defines `compute_gradients` and `apply_gradients` functions. The current version of `DistributedOptimizer` cannot be used in our case because it does not overwrite `apply_gradients` to use` self._optimizer.apply_gradients` and it calls (lines 178-179 of `horovod/tensorflow/__init__.py`)\r\n```python\r\ngradients = super(DistributedOptimizer, self).compute_gradients(*args, **kwargs)\r\n```\r\ninstead of `gradients = self._optimizer.compute_gradients(*args, **kwargs)`.\r\n\r\nIs it possible to update `DistributedOptimizer` to work in our use case and similar situations when `compute_gradients` and `apply_gradients` are redefined? As far as I can see simply changing lines 178-179 of `horovod/tensorflow/__init__.py` to \r\n```python\r\ngradients = self._optimizer.compute_gradients(*args, **kwargs)\r\n```\r\nand adding additional method:\r\n```python\r\n  def apply_gradients(self, grads_and_vars, global_step=None, name=None):\r\n    \"\"\"Calls this same method on the underlying optimizer.\"\"\"\r\n    return self._optimizer.apply_gradients(grads_and_vars, global_step, name)\r\n```\r\nshould make it work.\n", "before_files": [{"content": "__version__ = '0.13.0'\n", "path": "horovod/__init__.py"}, {"content": "# Copyright 2016 The TensorFlow Authors. All Rights Reserved.\n# Modifications copyright (C) 2017 Uber Technologies, Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n# ==============================================================================\n# pylint: disable=g-short-docstring-punctuation\n\"\"\"## Communicating Between Processes with MPI\n\nTensorFlow natively provides inter-device communication through send and\nreceive ops and inter-node communication through Distributed TensorFlow, based\non the same send and receive abstractions. On HPC clusters where Infiniband or\nother high-speed node interconnects are available, these can end up being\ninsufficient for synchronous data-parallel training (without asynchronous\ngradient descent). This module implements a variety of MPI ops which can take\nadvantage of hardware-specific MPI libraries for efficient communication.\n\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nfrom horovod.common import init\nfrom horovod.common import size\nfrom horovod.common import local_size\nfrom horovod.common import rank\nfrom horovod.common import local_rank\nfrom horovod.common import mpi_threads_supported\nfrom horovod.common import check_extension\n\ncheck_extension('horovod.tensorflow', 'HOROVOD_WITH_TENSORFLOW', __file__, 'mpi_lib')\n\nfrom horovod.tensorflow.mpi_ops import allgather\nfrom horovod.tensorflow.mpi_ops import broadcast\nfrom horovod.tensorflow.mpi_ops import _allreduce\n\nimport tensorflow as tf\n\n\ndef allreduce(tensor, average=True, device_dense='', device_sparse=''):\n    \"\"\"Perform an allreduce on a tf.Tensor or tf.IndexedSlices.\n\n    Arguments:\n        tensor: tf.Tensor, tf.Variable, or tf.IndexedSlices to reduce.\n        The shape of the input must be identical across all ranks.\n        average: If True, computes the average over all ranks.\n                 Otherwise, computes the sum over all ranks.\n        device_dense: Device to be used for dense tensors. Uses GPU by default\n                      if Horovod was build with HOROVOD_GPU_ALLREDUCE.\n        device_sparse: Device to be used for sparse tensors. Uses GPU by default\n                       if Horovod was build with HOROVOD_GPU_ALLGATHER.\n\n    This function performs a bandwidth-optimal ring allreduce on the input\n    tensor. If the input is an tf.IndexedSlices, the function instead does an\n    allgather on the values and the indices, effectively doing an allreduce on\n    the represented tensor.\n    \"\"\"\n    if isinstance(tensor, tf.IndexedSlices):\n        with tf.device(device_sparse):\n            # For IndexedSlices, do two allgathers intead of an allreduce.\n            horovod_size = tf.cast(size(), tensor.values.dtype)\n            values = allgather(tensor.values)\n            indices = allgather(tensor.indices)\n\n            # To make this operation into an average, divide all gathered values by\n            # the Horovod size.\n            new_values = tf.div(values, horovod_size) if average else values\n        return tf.IndexedSlices(new_values, indices,\n                                dense_shape=tensor.dense_shape)\n    else:\n        with tf.device(device_dense):\n            horovod_size = tf.cast(size(), tensor.dtype)\n            summed_tensor = _allreduce(tensor)\n            new_tensor = (tf.div(summed_tensor, horovod_size)\n                          if average else summed_tensor)\n        return new_tensor\n\n\ndef broadcast_global_variables(root_rank):\n    \"\"\"Broadcasts all global variables from root rank to all other processes.\n\n    Arguments:\n        root_rank: rank of the process from which global variables will be broadcasted\n        to all other processes.\n    \"\"\"\n    return tf.group(*[tf.assign(var, broadcast(var, root_rank))\n                      for var in tf.global_variables()])\n\n\nclass BroadcastGlobalVariablesHook(tf.train.SessionRunHook):\n    \"\"\"\n    SessionRunHook that will broadcast all global variables from root rank\n    to all other processes during initialization.\n\n    This is necessary to ensure consistent initialization of all workers when\n    training is started with random weights or restored from a checkpoint.\n    \"\"\"\n\n    def __init__(self, root_rank, device=''):\n        \"\"\"Construct a new BroadcastGlobalVariablesHook that will broadcast all\n        global variables from root rank to all other processes during initialization.\n\n        Args:\n          root_rank:\n            Rank that will send data, other ranks will receive data.\n          device:\n            Device to be used for broadcasting. Uses GPU by default\n            if Horovod was build with HOROVOD_GPU_BROADCAST.\n        \"\"\"\n        super(BroadcastGlobalVariablesHook, self).__init__()\n        self.root_rank = root_rank\n        self.bcast_op = None\n        self.device = device\n\n    def begin(self):\n        if not self.bcast_op or self.bcast_op.graph != tf.get_default_graph():\n            with tf.device(self.device):\n                self.bcast_op = broadcast_global_variables(self.root_rank)\n\n    def after_create_session(self, session, coord):\n        session.run(self.bcast_op)\n\n\nclass DistributedOptimizer(tf.train.Optimizer):\n    \"\"\"An optimizer that wraps another tf.Optimizer, using an allreduce to\n    average gradient values before applying gradients to model weights.\"\"\"\n\n    def __init__(self, optimizer, name=None, use_locking=False, device_dense='',\n                 device_sparse=''):\n        \"\"\"Construct a new DistributedOptimizer, which uses another optimizer\n        under the hood for computing single-process gradient values and\n        applying gradient updates after the gradient values have been averaged\n        across all the Horovod ranks.\n\n        Args:\n          optimizer:\n            Optimizer to use for computing gradients and applying updates.\n          name:\n            Optional name prefix for the operations created when applying\n            gradients. Defaults to \"Distributed\" followed by the provided\n            optimizer type.\n          use_locking:\n            Whether to use locking when updating variables.\n            See Optimizer.__init__ for more info.\n          device_dense:\n            Device to be used for dense tensors. Uses GPU by default\n            if Horovod was build with HOROVOD_GPU_ALLREDUCE.\n          device_sparse:\n            Device to be used for sparse tensors. Uses GPU by default\n            if Horovod was build with HOROVOD_GPU_ALLGATHER.\n        \"\"\"\n        if name is None:\n            name = \"Distributed{}\".format(type(optimizer).__name__)\n\n        self._optimizer = optimizer\n        self._device_dense = device_dense\n        self._device_sparse = device_sparse\n        super(DistributedOptimizer, self).__init__(\n            name=name, use_locking=use_locking)\n\n    def compute_gradients(self, *args, **kwargs):\n        \"\"\"Compute gradients of all trainable variables.\n\n        See Optimizer.compute_gradients() for more info.\n\n        In DistributedOptimizer, compute_gradients() is overriden to also\n        allreduce the gradients before returning them.\n        \"\"\"\n        gradients = (super(DistributedOptimizer, self)\n                     .compute_gradients(*args, **kwargs))\n        if size() > 1:\n            averaged_gradients = []\n            with tf.name_scope(self._name + \"_Allreduce\"):\n                for grad, var in gradients:\n                    if grad is not None:\n                        avg_grad = allreduce(grad, device_dense=self._device_dense,\n                                             device_sparse=self._device_sparse)\n                        averaged_gradients.append((avg_grad, var))\n                    else:\n                        averaged_gradients.append((None, var))\n            return averaged_gradients\n        else:\n            return gradients\n\n    def _apply_dense(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._apply_dense(*args, **kwargs)\n\n    def _resource_apply_dense(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._resource_apply_dense(*args, **kwargs)\n\n    def _resource_apply_sparse_duplicate_indices(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._resource_apply_sparse_duplicate_indices(*args, **kwargs)\n\n    def _resource_apply_sparse(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._resource_apply_sparse(*args, **kwargs)\n\n    def _apply_sparse_duplicate_indices(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._apply_sparse_duplicate_indices(*args, **kwargs)\n\n    def _apply_sparse(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._apply_sparse(*args, **kwargs)\n\n    def _prepare(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._prepare(*args, **kwargs)\n\n    def _create_slots(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._create_slots(*args, **kwargs)\n\n    def _valid_dtypes(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._valid_dtypes(*args, **kwargs)\n\n    def _finish(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer._finish(*args, **kwargs)\n", "path": "horovod/tensorflow/__init__.py"}], "after_files": [{"content": "__version__ = '0.13.1'\n", "path": "horovod/__init__.py"}, {"content": "# Copyright 2016 The TensorFlow Authors. All Rights Reserved.\n# Modifications copyright (C) 2017 Uber Technologies, Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n# ==============================================================================\n# pylint: disable=g-short-docstring-punctuation\n\"\"\"## Communicating Between Processes with MPI\n\nTensorFlow natively provides inter-device communication through send and\nreceive ops and inter-node communication through Distributed TensorFlow, based\non the same send and receive abstractions. On HPC clusters where Infiniband or\nother high-speed node interconnects are available, these can end up being\ninsufficient for synchronous data-parallel training (without asynchronous\ngradient descent). This module implements a variety of MPI ops which can take\nadvantage of hardware-specific MPI libraries for efficient communication.\n\"\"\"\n\nfrom __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nfrom horovod.common import init\nfrom horovod.common import size\nfrom horovod.common import local_size\nfrom horovod.common import rank\nfrom horovod.common import local_rank\nfrom horovod.common import mpi_threads_supported\nfrom horovod.common import check_extension\n\ncheck_extension('horovod.tensorflow', 'HOROVOD_WITH_TENSORFLOW', __file__, 'mpi_lib')\n\nfrom horovod.tensorflow.mpi_ops import allgather\nfrom horovod.tensorflow.mpi_ops import broadcast\nfrom horovod.tensorflow.mpi_ops import _allreduce\n\nimport tensorflow as tf\n\n\ndef allreduce(tensor, average=True, device_dense='', device_sparse=''):\n    \"\"\"Perform an allreduce on a tf.Tensor or tf.IndexedSlices.\n\n    Arguments:\n        tensor: tf.Tensor, tf.Variable, or tf.IndexedSlices to reduce.\n        The shape of the input must be identical across all ranks.\n        average: If True, computes the average over all ranks.\n                 Otherwise, computes the sum over all ranks.\n        device_dense: Device to be used for dense tensors. Uses GPU by default\n                      if Horovod was build with HOROVOD_GPU_ALLREDUCE.\n        device_sparse: Device to be used for sparse tensors. Uses GPU by default\n                       if Horovod was build with HOROVOD_GPU_ALLGATHER.\n\n    This function performs a bandwidth-optimal ring allreduce on the input\n    tensor. If the input is an tf.IndexedSlices, the function instead does an\n    allgather on the values and the indices, effectively doing an allreduce on\n    the represented tensor.\n    \"\"\"\n    if isinstance(tensor, tf.IndexedSlices):\n        with tf.device(device_sparse):\n            # For IndexedSlices, do two allgathers intead of an allreduce.\n            horovod_size = tf.cast(size(), tensor.values.dtype)\n            values = allgather(tensor.values)\n            indices = allgather(tensor.indices)\n\n            # To make this operation into an average, divide all gathered values by\n            # the Horovod size.\n            new_values = tf.div(values, horovod_size) if average else values\n        return tf.IndexedSlices(new_values, indices,\n                                dense_shape=tensor.dense_shape)\n    else:\n        with tf.device(device_dense):\n            horovod_size = tf.cast(size(), tensor.dtype)\n            summed_tensor = _allreduce(tensor)\n            new_tensor = (tf.div(summed_tensor, horovod_size)\n                          if average else summed_tensor)\n        return new_tensor\n\n\ndef broadcast_global_variables(root_rank):\n    \"\"\"Broadcasts all global variables from root rank to all other processes.\n\n    Arguments:\n        root_rank: rank of the process from which global variables will be broadcasted\n        to all other processes.\n    \"\"\"\n    return tf.group(*[tf.assign(var, broadcast(var, root_rank))\n                      for var in tf.global_variables()])\n\n\nclass BroadcastGlobalVariablesHook(tf.train.SessionRunHook):\n    \"\"\"\n    SessionRunHook that will broadcast all global variables from root rank\n    to all other processes during initialization.\n\n    This is necessary to ensure consistent initialization of all workers when\n    training is started with random weights or restored from a checkpoint.\n    \"\"\"\n\n    def __init__(self, root_rank, device=''):\n        \"\"\"Construct a new BroadcastGlobalVariablesHook that will broadcast all\n        global variables from root rank to all other processes during initialization.\n\n        Args:\n          root_rank:\n            Rank that will send data, other ranks will receive data.\n          device:\n            Device to be used for broadcasting. Uses GPU by default\n            if Horovod was build with HOROVOD_GPU_BROADCAST.\n        \"\"\"\n        super(BroadcastGlobalVariablesHook, self).__init__()\n        self.root_rank = root_rank\n        self.bcast_op = None\n        self.device = device\n\n    def begin(self):\n        if not self.bcast_op or self.bcast_op.graph != tf.get_default_graph():\n            with tf.device(self.device):\n                self.bcast_op = broadcast_global_variables(self.root_rank)\n\n    def after_create_session(self, session, coord):\n        session.run(self.bcast_op)\n\n\nclass DistributedOptimizer(tf.train.Optimizer):\n    \"\"\"An optimizer that wraps another tf.Optimizer, using an allreduce to\n    average gradient values before applying gradients to model weights.\"\"\"\n\n    def __init__(self, optimizer, name=None, use_locking=False, device_dense='',\n                 device_sparse=''):\n        \"\"\"Construct a new DistributedOptimizer, which uses another optimizer\n        under the hood for computing single-process gradient values and\n        applying gradient updates after the gradient values have been averaged\n        across all the Horovod ranks.\n\n        Args:\n          optimizer:\n            Optimizer to use for computing gradients and applying updates.\n          name:\n            Optional name prefix for the operations created when applying\n            gradients. Defaults to \"Distributed\" followed by the provided\n            optimizer type.\n          use_locking:\n            Whether to use locking when updating variables.\n            See Optimizer.__init__ for more info.\n          device_dense:\n            Device to be used for dense tensors. Uses GPU by default\n            if Horovod was build with HOROVOD_GPU_ALLREDUCE.\n          device_sparse:\n            Device to be used for sparse tensors. Uses GPU by default\n            if Horovod was build with HOROVOD_GPU_ALLGATHER.\n        \"\"\"\n        if name is None:\n            name = \"Distributed{}\".format(type(optimizer).__name__)\n\n        self._optimizer = optimizer\n        self._device_dense = device_dense\n        self._device_sparse = device_sparse\n        super(DistributedOptimizer, self).__init__(\n            name=name, use_locking=use_locking)\n\n    def compute_gradients(self, *args, **kwargs):\n        \"\"\"Compute gradients of all trainable variables.\n\n        See Optimizer.compute_gradients() for more info.\n\n        In DistributedOptimizer, compute_gradients() is overriden to also\n        allreduce the gradients before returning them.\n        \"\"\"\n        gradients = self._optimizer.compute_gradients(*args, **kwargs)\n        if size() > 1:\n            averaged_gradients = []\n            with tf.name_scope(self._name + \"_Allreduce\"):\n                for grad, var in gradients:\n                    if grad is not None:\n                        avg_grad = allreduce(grad, device_dense=self._device_dense,\n                                             device_sparse=self._device_sparse)\n                        averaged_gradients.append((avg_grad, var))\n                    else:\n                        averaged_gradients.append((None, var))\n            return averaged_gradients\n        else:\n            return gradients\n\n    def apply_gradients(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer.apply_gradients(*args, **kwargs)\n\n    def get_slot(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer.get_slot(*args, **kwargs)\n\n    def get_slot_names(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer.get_slot_names(*args, **kwargs)\n\n    def variables(self, *args, **kwargs):\n        \"\"\"Calls this same method on the underlying optimizer.\"\"\"\n        return self._optimizer.variables(*args, **kwargs)\n", "path": "horovod/tensorflow/__init__.py"}]}