We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
【ELASTIC HOROVOD】It perhaps some  deadlock when all workers are recorded.
**Environment:**
1. Framework: (TensorFlow, Keras, PyTorch, MXNet): Pytorch
2. Framework version: 1.6.0
3. Horovod version: 0.21.3
4. MPI version: 4.0.3
5. CUDA version: 10.2
6. NCCL version: 2.7.6
7. Python version: 3.6

**Checklist:**
1. Did you search issues to find if somebody asked this question before? Yes.
2. If your question is about hang, did you read [this doc](https://github.com/horovod/horovod/blob/master/docs/running.rst)?
3. If your question is about docker, did you read [this doc](https://github.com/horovod/horovod/blob/master/docs/docker.rst)?
4. Did you check if you question is answered in the [troubleshooting guide]  (https://github.com/horovod/horovod/blob/master/docs/troubleshooting.rst)? Yes

**Bug report:**
After all workers have report their status, the driver will execute the `_on_workers_recorded` function which get and update host slot info and activate workers if necessary. The workers wait until the `_on_workers_recorded` function is completed.  When executing `_on_workers_recorded` function, if one READY worker becomes FAILED，the deadlock will caused.(e.g. `_on_workers_recorded` call host discovery script that maybe consume seconds and one ready worker failed during this time). 

The deadlock details as follows:
1. If one ready worker(we call it `worker-A`) become failed during `_on_workers_recorded`, it will result in `_barrier.reset()` and holds  the `self._lock`, relevant code at [here](https://github.com/horovod/horovod/blob/1a0a6f2c5ec536c44fd5292875064b65545de6e0/horovod/runner/elastic/registration.py#L98). So it hang in `_barrier.reset()` until `_on_workers_recorded` function finish.

2. But `_on_workers_recorded` function also try to acquire `self._lock` which already held by `worker-A` at [self.reset](https://github.com/horovod/horovod/blob/1a0a6f2c5ec536c44fd5292875064b65545de6e0/horovod/runner/elastic/registration.py#L52).  Deadlock occurs!

**Steps to reproduce.**
1. In order to easily reproduce the problem, we sleep 15s  at `wait_for_available_slots` in `horovod/runner/elastic/driver.py` file (It simulates the case that host discovery script consume a few seconds):
```
...
    def wait_for_available_slots(self, min_np, min_hosts=1):
        print(f"wait_for_available_slots sleep 15s")
        time.sleep(15)

        extra_message = ' An elastic job also requires that at least two hosts ' \
                        'are available to resolve compatible network interfaces. If you know which interfaces ' \
                        'are compatible in your network, set `--network-interface` to skip this check.' \
            if min_hosts > 1 else ''
...
```

2. Run elastic horovod:
```
horovodrun -np 1 --host-discovery-script ./discovery_hosts.sh --network-interface eth1 --min-np 1 --log-level DEBUG --verbose  python3 pytorch_synthetic_benchmark_elastic.py --num-iters=1000
```

3. After some iteration passed, we add a new worker in host-discovery-script to raise `HostsUpdatedInterrupt`. The driver will record all workers as ready and call  `_activate_workers` to new a worker and go to `wait_for_available_slots`, finally sleeping 15s in `wait_for_available_slots`.

4. We immediately  kill one worker and the driver updates this worker as failed. Driver blocks in `_barrier.reset()`, holding the `self._lock`.

5. After 15s, `_activate_workers` call `_worker_registry.reset()` to  acquire `_lock` which already is held. Deadlock! 

**Solution.**
I think this issue is completely caused by worker updates during `_on_workers_recorded`.  Maybe should we prohibit any worker updates in `_on_workers_recorded` function? If any worker updated during this time,  we delay this to next rendezvous.  

--- 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/runner/elastic/registration.py`
Content:
```
1 # Copyright 2020 Uber Technologies, Inc. All Rights Reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 # ==============================================================================
15 
16 import logging
17 import threading
18 
19 from collections import defaultdict
20 
21 from horovod.runner.elastic import constants
22 
23 READY = 'READY'
24 SUCCESS = 'SUCCESS'
25 FAILURE = 'FAILURE'
26 
27 
28 class WorkerStateRegistry(object):
29     def __init__(self, driver, host_manager, reset_limit=None, verbose=False):
30         self._driver = driver
31         self._host_manager = host_manager
32         self._reset_limit = reset_limit
33         self._reset_count = 0
34         self._lock = threading.Lock()
35         self._states = {}
36         self._workers = defaultdict(set)
37         self._barrier = None
38         self._rendezvous_id = 0
39         self._verbose = verbose
40         self._size = 0
41 
42     def get_recorded_slots(self):
43         return self._states.keys()
44 
45     def get(self, state):
46         return self._workers[state]
47 
48     def count(self, state):
49         return len(self._workers[state])
50 
51     def reset(self, size):
52         with self._lock:
53             logging.info('reset workers: {}'.format(size))
54             self._states.clear()
55             self._workers.clear()
56             self._barrier = threading.Barrier(parties=size, action=self._action)
57             self._rendezvous_id += 1
58             self._size = size
59 
60     def size(self):
61         return self._size
62 
63     def last_rendezvous(self):
64         return self._rendezvous_id
65 
66     def record_ready(self, host, slot):
67         return self._record_state(host, slot, READY)
68 
69     def record_success(self, host, slot):
70         return self._record_state(host, slot, SUCCESS)
71 
72     def record_failure(self, host, slot):
73         return self._record_state(host, slot, FAILURE)
74 
75     def _record_state(self, host, slot, state):
76         if self._driver.finished():
77             logging.info('driver finished, ignoring registration: {}[{}] = {}'.format(host, slot, state))
78             return self._rendezvous_id
79 
80         if self._host_manager.is_blacklisted(host):
81             logging.warning('host registers state %s but is already blacklisted, ignoring: %s', state, host)
82             return self._rendezvous_id
83 
84         key = (host, slot)
85         with self._lock:
86             if key in self._states:
87                 if state == FAILURE:
88                     # Worker originally recorded itself as READY, but the worker failed while waiting at the barrier. As
89                     # such, we need to update the state to FAILURE, and we don't want two threads coming from the same
90                     # worker at the barrier.
91                     #
92                     # In order to ensure that the new failing thread can record results in cases of total job failure,
93                     # we also need to block this thread by waiting on the barrier. This requires us to reset the barrier,
94                     # as otherwise this worker will be double-counted (once for the READY thread and once for FAILURE),
95                     # which would cause the barrier to complete too early.
96                     logging.info('key exists, reset barrier: {}[{}] = {} -> {}'
97                                  .format(host, slot, self._states[key], state))
98                     self._barrier.reset()
99                 else:
100                     logging.error('key exists and new state %s not FAILURE, '
101                                   'ignoring (current state is %s)', state, self._states[key])
102 
103             if key not in self._states or state == FAILURE:
104                 logging.info('record state: {}[{}] = {}'.format(host, slot, state))
105                 self._states[key] = state
106                 self._workers[state].add(key)
107 
108             rendezvous_id = self._rendezvous_id
109 
110         rendezvous_id = self._wait(key, state, rendezvous_id)
111         return rendezvous_id
112 
113     def _wait(self, key, state, rendezvous_id):
114         while True:
115             try:
116                 self._barrier.wait()
117                 return rendezvous_id
118             except threading.BrokenBarrierError:
119                 if self._barrier.broken:
120                     # Timeout or other non-recoverable error, so exit
121                     raise
122 
123                 # Barrier has been reset
124                 with self._lock:
125                     # Check to make sure the reset was not caused by a change of state for this key
126                     rendezvous_id = self._rendezvous_id
127                     saved_state = self._states.get(key, state)
128                     if saved_state != state:
129                         # This worker changed its state, so do not attempt to wait again to avoid double-counting
130                         raise RuntimeError('State {} overridden by {}'.format(state, saved_state))
131 
132     def _action(self):
133         self._on_workers_recorded()
134 
135     def _on_workers_recorded(self):
136         logging.info('all {} workers recorded'.format(self.size()))
137 
138         # Check for success state, if any process succeeded, shutdown all other processes
139         if self.count(SUCCESS) > 0:
140             logging.info('success count == {} -> stop running'.format(self.count(SUCCESS)))
141             self._driver.stop()
142             return
143 
144         # Check that all processes failed, indicating that processing should stop
145         if self.count(FAILURE) == self._size:
146             logging.error('failure count == {} -> stop running'.format(self._size))
147             self._driver.stop()
148             return
149 
150         # Check for failures, and add them to the blacklisted hosts list
151         failures = self.get(FAILURE)
152         for host, slot in failures:
153             self._host_manager.blacklist(host)
154 
155         # If every active host is blacklisted, then treat this as job failure
156         if all([self._host_manager.is_blacklisted(host) for host, slot in self.get_recorded_slots()]):
157             logging.error('blacklisted slots count == {} -> stop running'.format(self._size))
158             self._driver.stop()
159             return
160 
161         # Check that we have already reset the maximum number of allowed times
162         if self._reset_limit is not None and self._reset_count >= self._reset_limit:
163             logging.error('reset count {} has exceeded limit {} -> stop running'
164                           .format(self._reset_count, self._reset_limit))
165             self._driver.stop(error_message=constants.RESET_LIMIT_EXCEEDED_MESSAGE.format(self._reset_limit))
166             return
167 
168         try:
169             self._reset_count += 1
170             self._driver.resume()
171         except Exception:
172             logging.exception('failed to activate new hosts -> stop running')
173             self._driver.stop()
174
```
--- 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/runner/elastic/registration.py b/horovod/runner/elastic/registration.py
--- a/horovod/runner/elastic/registration.py
+++ b/horovod/runner/elastic/registration.py
@@ -38,6 +38,8 @@
         self._rendezvous_id = 0
         self._verbose = verbose
         self._size = 0
+        self._action_event = threading.Event()
+        self._action_event.set()
 
     def get_recorded_slots(self):
         return self._states.keys()
@@ -81,6 +83,9 @@
             logging.warning('host registers state %s but is already blacklisted, ignoring: %s', state, host)
             return self._rendezvous_id
 
+        # we should wait for _action finished if a _record_state called when _action is running
+        self._action_event.wait()
+
         key = (host, slot)
         with self._lock:
             if key in self._states:
@@ -130,7 +135,9 @@
                         raise RuntimeError('State {} overridden by {}'.format(state, saved_state))
 
     def _action(self):
+        self._action_event.clear()
         self._on_workers_recorded()
+        self._action_event.set()
 
     def _on_workers_recorded(self):
         logging.info('all {} workers recorded'.format(self.size()))

{"golden_diff": "diff --git a/horovod/runner/elastic/registration.py b/horovod/runner/elastic/registration.py\n--- a/horovod/runner/elastic/registration.py\n+++ b/horovod/runner/elastic/registration.py\n@@ -38,6 +38,8 @@\n         self._rendezvous_id = 0\n         self._verbose = verbose\n         self._size = 0\n+        self._action_event = threading.Event()\n+        self._action_event.set()\n \n     def get_recorded_slots(self):\n         return self._states.keys()\n@@ -81,6 +83,9 @@\n             logging.warning('host registers state %s but is already blacklisted, ignoring: %s', state, host)\n             return self._rendezvous_id\n \n+        # we should wait for _action finished if a _record_state called when _action is running\n+        self._action_event.wait()\n+\n         key = (host, slot)\n         with self._lock:\n             if key in self._states:\n@@ -130,7 +135,9 @@\n                         raise RuntimeError('State {} overridden by {}'.format(state, saved_state))\n \n     def _action(self):\n+        self._action_event.clear()\n         self._on_workers_recorded()\n+        self._action_event.set()\n \n     def _on_workers_recorded(self):\n         logging.info('all {} workers recorded'.format(self.size()))\n", "issue": "\u3010ELASTIC HOROVOD\u3011It perhaps some  deadlock when all workers are recorded.\n**Environment:**\r\n1. Framework: (TensorFlow, Keras, PyTorch, MXNet): Pytorch\r\n2. Framework version: 1.6.0\r\n3. Horovod version: 0.21.3\r\n4. MPI version: 4.0.3\r\n5. CUDA version: 10.2\r\n6. NCCL version: 2.7.6\r\n7. Python version: 3.6\r\n\r\n**Checklist:**\r\n1. Did you search issues to find if somebody asked this question before? Yes.\r\n2. If your question is about hang, did you read [this doc](https://github.com/horovod/horovod/blob/master/docs/running.rst)?\r\n3. If your question is about docker, did you read [this doc](https://github.com/horovod/horovod/blob/master/docs/docker.rst)?\r\n4. Did you check if you question is answered in the [troubleshooting guide]  (https://github.com/horovod/horovod/blob/master/docs/troubleshooting.rst)? Yes\r\n\r\n**Bug report:**\r\nAfter all workers have report their status, the driver will execute the `_on_workers_recorded` function which get and update host slot info and activate workers if necessary. The workers wait until the `_on_workers_recorded` function is completed.  When executing `_on_workers_recorded` function, if one READY worker becomes FAILED\uff0cthe deadlock will caused.(e.g. `_on_workers_recorded` call host discovery script that maybe consume seconds and one ready worker failed during this time). \r\n\r\nThe deadlock details as follows:\r\n1. If one ready worker(we call it `worker-A`) become failed during `_on_workers_recorded`, it will result in `_barrier.reset()` and holds  the `self._lock`, relevant code at [here](https://github.com/horovod/horovod/blob/1a0a6f2c5ec536c44fd5292875064b65545de6e0/horovod/runner/elastic/registration.py#L98). So it hang in `_barrier.reset()` until `_on_workers_recorded` function finish.\r\n\r\n2. But `_on_workers_recorded` function also try to acquire `self._lock` which already held by `worker-A` at [self.reset](https://github.com/horovod/horovod/blob/1a0a6f2c5ec536c44fd5292875064b65545de6e0/horovod/runner/elastic/registration.py#L52).  Deadlock occurs!\r\n\r\n**Steps to reproduce.**\r\n1. In order to easily reproduce the problem, we sleep 15s  at `wait_for_available_slots` in `horovod/runner/elastic/driver.py` file (It simulates the case that host discovery script consume a few seconds):\r\n```\r\n...\r\n    def wait_for_available_slots(self, min_np, min_hosts=1):\r\n        print(f\"wait_for_available_slots sleep 15s\")\r\n        time.sleep(15)\r\n\r\n        extra_message = ' An elastic job also requires that at least two hosts ' \\\r\n                        'are available to resolve compatible network interfaces. If you know which interfaces ' \\\r\n                        'are compatible in your network, set `--network-interface` to skip this check.' \\\r\n            if min_hosts > 1 else ''\r\n...\r\n```\r\n\r\n2. Run elastic horovod:\r\n```\r\nhorovodrun -np 1 --host-discovery-script ./discovery_hosts.sh --network-interface eth1 --min-np 1 --log-level DEBUG --verbose  python3 pytorch_synthetic_benchmark_elastic.py --num-iters=1000\r\n```\r\n\r\n3. After some iteration passed, we add a new worker in host-discovery-script to raise `HostsUpdatedInterrupt`. The driver will record all workers as ready and call  `_activate_workers` to new a worker and go to `wait_for_available_slots`, finally sleeping 15s in `wait_for_available_slots`.\r\n\r\n4. We immediately  kill one worker and the driver updates this worker as failed. Driver blocks in `_barrier.reset()`, holding the `self._lock`.\r\n\r\n5. After 15s, `_activate_workers` call `_worker_registry.reset()` to  acquire `_lock` which already is held. Deadlock! \r\n\r\n**Solution.**\r\nI think this issue is completely caused by worker updates during `_on_workers_recorded`.  Maybe should we prohibit any worker updates in `_on_workers_recorded` function? If any worker updated during this time,  we delay this to next rendezvous.  \n", "before_files": [{"content": "# Copyright 2020 Uber Technologies, Inc. All Rights Reserved.\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\nimport logging\nimport threading\n\nfrom collections import defaultdict\n\nfrom horovod.runner.elastic import constants\n\nREADY = 'READY'\nSUCCESS = 'SUCCESS'\nFAILURE = 'FAILURE'\n\n\nclass WorkerStateRegistry(object):\n    def __init__(self, driver, host_manager, reset_limit=None, verbose=False):\n        self._driver = driver\n        self._host_manager = host_manager\n        self._reset_limit = reset_limit\n        self._reset_count = 0\n        self._lock = threading.Lock()\n        self._states = {}\n        self._workers = defaultdict(set)\n        self._barrier = None\n        self._rendezvous_id = 0\n        self._verbose = verbose\n        self._size = 0\n\n    def get_recorded_slots(self):\n        return self._states.keys()\n\n    def get(self, state):\n        return self._workers[state]\n\n    def count(self, state):\n        return len(self._workers[state])\n\n    def reset(self, size):\n        with self._lock:\n            logging.info('reset workers: {}'.format(size))\n            self._states.clear()\n            self._workers.clear()\n            self._barrier = threading.Barrier(parties=size, action=self._action)\n            self._rendezvous_id += 1\n            self._size = size\n\n    def size(self):\n        return self._size\n\n    def last_rendezvous(self):\n        return self._rendezvous_id\n\n    def record_ready(self, host, slot):\n        return self._record_state(host, slot, READY)\n\n    def record_success(self, host, slot):\n        return self._record_state(host, slot, SUCCESS)\n\n    def record_failure(self, host, slot):\n        return self._record_state(host, slot, FAILURE)\n\n    def _record_state(self, host, slot, state):\n        if self._driver.finished():\n            logging.info('driver finished, ignoring registration: {}[{}] = {}'.format(host, slot, state))\n            return self._rendezvous_id\n\n        if self._host_manager.is_blacklisted(host):\n            logging.warning('host registers state %s but is already blacklisted, ignoring: %s', state, host)\n            return self._rendezvous_id\n\n        key = (host, slot)\n        with self._lock:\n            if key in self._states:\n                if state == FAILURE:\n                    # Worker originally recorded itself as READY, but the worker failed while waiting at the barrier. As\n                    # such, we need to update the state to FAILURE, and we don't want two threads coming from the same\n                    # worker at the barrier.\n                    #\n                    # In order to ensure that the new failing thread can record results in cases of total job failure,\n                    # we also need to block this thread by waiting on the barrier. This requires us to reset the barrier,\n                    # as otherwise this worker will be double-counted (once for the READY thread and once for FAILURE),\n                    # which would cause the barrier to complete too early.\n                    logging.info('key exists, reset barrier: {}[{}] = {} -> {}'\n                                 .format(host, slot, self._states[key], state))\n                    self._barrier.reset()\n                else:\n                    logging.error('key exists and new state %s not FAILURE, '\n                                  'ignoring (current state is %s)', state, self._states[key])\n\n            if key not in self._states or state == FAILURE:\n                logging.info('record state: {}[{}] = {}'.format(host, slot, state))\n                self._states[key] = state\n                self._workers[state].add(key)\n\n            rendezvous_id = self._rendezvous_id\n\n        rendezvous_id = self._wait(key, state, rendezvous_id)\n        return rendezvous_id\n\n    def _wait(self, key, state, rendezvous_id):\n        while True:\n            try:\n                self._barrier.wait()\n                return rendezvous_id\n            except threading.BrokenBarrierError:\n                if self._barrier.broken:\n                    # Timeout or other non-recoverable error, so exit\n                    raise\n\n                # Barrier has been reset\n                with self._lock:\n                    # Check to make sure the reset was not caused by a change of state for this key\n                    rendezvous_id = self._rendezvous_id\n                    saved_state = self._states.get(key, state)\n                    if saved_state != state:\n                        # This worker changed its state, so do not attempt to wait again to avoid double-counting\n                        raise RuntimeError('State {} overridden by {}'.format(state, saved_state))\n\n    def _action(self):\n        self._on_workers_recorded()\n\n    def _on_workers_recorded(self):\n        logging.info('all {} workers recorded'.format(self.size()))\n\n        # Check for success state, if any process succeeded, shutdown all other processes\n        if self.count(SUCCESS) > 0:\n            logging.info('success count == {} -> stop running'.format(self.count(SUCCESS)))\n            self._driver.stop()\n            return\n\n        # Check that all processes failed, indicating that processing should stop\n        if self.count(FAILURE) == self._size:\n            logging.error('failure count == {} -> stop running'.format(self._size))\n            self._driver.stop()\n            return\n\n        # Check for failures, and add them to the blacklisted hosts list\n        failures = self.get(FAILURE)\n        for host, slot in failures:\n            self._host_manager.blacklist(host)\n\n        # If every active host is blacklisted, then treat this as job failure\n        if all([self._host_manager.is_blacklisted(host) for host, slot in self.get_recorded_slots()]):\n            logging.error('blacklisted slots count == {} -> stop running'.format(self._size))\n            self._driver.stop()\n            return\n\n        # Check that we have already reset the maximum number of allowed times\n        if self._reset_limit is not None and self._reset_count >= self._reset_limit:\n            logging.error('reset count {} has exceeded limit {} -> stop running'\n                          .format(self._reset_count, self._reset_limit))\n            self._driver.stop(error_message=constants.RESET_LIMIT_EXCEEDED_MESSAGE.format(self._reset_limit))\n            return\n\n        try:\n            self._reset_count += 1\n            self._driver.resume()\n        except Exception:\n            logging.exception('failed to activate new hosts -> stop running')\n            self._driver.stop()\n", "path": "horovod/runner/elastic/registration.py"}], "after_files": [{"content": "# Copyright 2020 Uber Technologies, Inc. All Rights Reserved.\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\nimport logging\nimport threading\n\nfrom collections import defaultdict\n\nfrom horovod.runner.elastic import constants\n\nREADY = 'READY'\nSUCCESS = 'SUCCESS'\nFAILURE = 'FAILURE'\n\n\nclass WorkerStateRegistry(object):\n    def __init__(self, driver, host_manager, reset_limit=None, verbose=False):\n        self._driver = driver\n        self._host_manager = host_manager\n        self._reset_limit = reset_limit\n        self._reset_count = 0\n        self._lock = threading.Lock()\n        self._states = {}\n        self._workers = defaultdict(set)\n        self._barrier = None\n        self._rendezvous_id = 0\n        self._verbose = verbose\n        self._size = 0\n        self._action_event = threading.Event()\n        self._action_event.set()\n\n    def get_recorded_slots(self):\n        return self._states.keys()\n\n    def get(self, state):\n        return self._workers[state]\n\n    def count(self, state):\n        return len(self._workers[state])\n\n    def reset(self, size):\n        with self._lock:\n            logging.info('reset workers: {}'.format(size))\n            self._states.clear()\n            self._workers.clear()\n            self._barrier = threading.Barrier(parties=size, action=self._action)\n            self._rendezvous_id += 1\n            self._size = size\n\n    def size(self):\n        return self._size\n\n    def last_rendezvous(self):\n        return self._rendezvous_id\n\n    def record_ready(self, host, slot):\n        return self._record_state(host, slot, READY)\n\n    def record_success(self, host, slot):\n        return self._record_state(host, slot, SUCCESS)\n\n    def record_failure(self, host, slot):\n        return self._record_state(host, slot, FAILURE)\n\n    def _record_state(self, host, slot, state):\n        if self._driver.finished():\n            logging.info('driver finished, ignoring registration: {}[{}] = {}'.format(host, slot, state))\n            return self._rendezvous_id\n\n        if self._host_manager.is_blacklisted(host):\n            logging.warning('host registers state %s but is already blacklisted, ignoring: %s', state, host)\n            return self._rendezvous_id\n\n        # we should wait for _action finished if a _record_state called when _action is running\n        self._action_event.wait()\n\n        key = (host, slot)\n        with self._lock:\n            if key in self._states:\n                if state == FAILURE:\n                    # Worker originally recorded itself as READY, but the worker failed while waiting at the barrier. As\n                    # such, we need to update the state to FAILURE, and we don't want two threads coming from the same\n                    # worker at the barrier.\n                    #\n                    # In order to ensure that the new failing thread can record results in cases of total job failure,\n                    # we also need to block this thread by waiting on the barrier. This requires us to reset the barrier,\n                    # as otherwise this worker will be double-counted (once for the READY thread and once for FAILURE),\n                    # which would cause the barrier to complete too early.\n                    logging.info('key exists, reset barrier: {}[{}] = {} -> {}'\n                                 .format(host, slot, self._states[key], state))\n                    self._barrier.reset()\n                else:\n                    logging.error('key exists and new state %s not FAILURE, '\n                                  'ignoring (current state is %s)', state, self._states[key])\n\n            if key not in self._states or state == FAILURE:\n                logging.info('record state: {}[{}] = {}'.format(host, slot, state))\n                self._states[key] = state\n                self._workers[state].add(key)\n\n            rendezvous_id = self._rendezvous_id\n\n        rendezvous_id = self._wait(key, state, rendezvous_id)\n        return rendezvous_id\n\n    def _wait(self, key, state, rendezvous_id):\n        while True:\n            try:\n                self._barrier.wait()\n                return rendezvous_id\n            except threading.BrokenBarrierError:\n                if self._barrier.broken:\n                    # Timeout or other non-recoverable error, so exit\n                    raise\n\n                # Barrier has been reset\n                with self._lock:\n                    # Check to make sure the reset was not caused by a change of state for this key\n                    rendezvous_id = self._rendezvous_id\n                    saved_state = self._states.get(key, state)\n                    if saved_state != state:\n                        # This worker changed its state, so do not attempt to wait again to avoid double-counting\n                        raise RuntimeError('State {} overridden by {}'.format(state, saved_state))\n\n    def _action(self):\n        self._action_event.clear()\n        self._on_workers_recorded()\n        self._action_event.set()\n\n    def _on_workers_recorded(self):\n        logging.info('all {} workers recorded'.format(self.size()))\n\n        # Check for success state, if any process succeeded, shutdown all other processes\n        if self.count(SUCCESS) > 0:\n            logging.info('success count == {} -> stop running'.format(self.count(SUCCESS)))\n            self._driver.stop()\n            return\n\n        # Check that all processes failed, indicating that processing should stop\n        if self.count(FAILURE) == self._size:\n            logging.error('failure count == {} -> stop running'.format(self._size))\n            self._driver.stop()\n            return\n\n        # Check for failures, and add them to the blacklisted hosts list\n        failures = self.get(FAILURE)\n        for host, slot in failures:\n            self._host_manager.blacklist(host)\n\n        # If every active host is blacklisted, then treat this as job failure\n        if all([self._host_manager.is_blacklisted(host) for host, slot in self.get_recorded_slots()]):\n            logging.error('blacklisted slots count == {} -> stop running'.format(self._size))\n            self._driver.stop()\n            return\n\n        # Check that we have already reset the maximum number of allowed times\n        if self._reset_limit is not None and self._reset_count >= self._reset_limit:\n            logging.error('reset count {} has exceeded limit {} -> stop running'\n                          .format(self._reset_count, self._reset_limit))\n            self._driver.stop(error_message=constants.RESET_LIMIT_EXCEEDED_MESSAGE.format(self._reset_limit))\n            return\n\n        try:\n            self._reset_count += 1\n            self._driver.resume()\n        except Exception:\n            logging.exception('failed to activate new hosts -> stop running')\n            self._driver.stop()\n", "path": "horovod/runner/elastic/registration.py"}]}