Repository: horovod/horovod
Pull request: Added reset limit for elastic mode to prevent introducing excessive variance
Fixes #1947. 

Problem statement:
Add reset limit when running in elastic mode
The user should be able to optionally specify a hard limit on the number of reset events that can occur, with an optional sliding window.

For example:

```
horovodrun --reset-limit 3
```

Or:

```
horovodrun --reset-limit 3 --reset-limit-window-duration 60
```

In the latter case, we will only fail of there are 3 reset events in a minute (60 seconds), while in the former, we will fail if we ever see 3 reset events over the lifetime of the job.

Hints from issue discussion:

Commit message: Added reset limit for Spark

Signed-off-by: Travis Addair <taddair@uber.com>

Patch to review:
diff --git a/horovod/run/elastic/constants.py b/horovod/run/elastic/constants.py
new file mode 100644
index 0000000000..2fc5479784
--- /dev/null
+++ b/horovod/run/elastic/constants.py
@@ -0,0 +1,16 @@
+# Copyright 2020 Uber Technologies, Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+
+RESET_LIMIT_EXCEEDED_MESSAGE = 'Horovod was shutdown because number of resets during elastic job exceeded limit: {}'
diff --git a/horovod/run/elastic/driver.py b/horovod/run/elastic/driver.py
index 8cb14944d8..26ea79e155 100644
--- a/horovod/run/elastic/driver.py
+++ b/horovod/run/elastic/driver.py
@@ -36,27 +36,37 @@ def _epoch_time_s():
 
 
 class Results(object):
+    def __init__(self, error_message, worker_results):
+        self.error_message = error_message
+        self.worker_results = worker_results
+
+
+class ResultsRecorder(object):
     def __init__(self):
-        self._results = {}
+        self._error_message = None
+        self._worker_results = {}
         self._worker_threads = queue.Queue()
 
     def expect(self, worker_thread):
         self._worker_threads.put(worker_thread)
 
+    def set_error_message(self, error_message):
+        self._error_message = error_message
+
     def add_result(self, key, value):
-        if key in self._results:
+        if key in self._worker_results:
             return
-        self._results[key] = value
+        self._worker_results[key] = value
 
     def get_results(self):
         while not self._worker_threads.empty():
             worker_thread = self._worker_threads.get()
             worker_thread.join()
-        return self._results
+        return Results(self._error_message, self._worker_results)
 
 
 class ElasticDriver(object):
-    def __init__(self, rendezvous, discovery, min_np, max_np, timeout=None, verbose=0):
+    def __init__(self, rendezvous, discovery, min_np, max_np, timeout=None, reset_limit=None, verbose=0):
         self._rendezvous = rendezvous
         self._host_manager = HostManager(discovery)
         self._min_np = min_np
@@ -73,8 +83,8 @@ def __init__(self, rendezvous, discovery, min_np, max_np, timeout=None, verbose=
         self._create_worker_fn = None
         self._worker_clients = {}
 
-        self._worker_registry = WorkerStateRegistry(self, self._host_manager)
-        self._results = Results()
+        self._worker_registry = WorkerStateRegistry(self, self._host_manager, reset_limit=reset_limit)
+        self._results = ResultsRecorder()
         self._shutdown = threading.Event()
 
         self._discovery_thread = threading.Thread(target=self._discover_hosts)
@@ -88,7 +98,8 @@ def start(self, np, create_worker_fn):
     def resume(self):
         self._activate_workers(self._min_np)
 
-    def stop(self):
+    def stop(self, error_message=None):
+        self._results.set_error_message(error_message)
         self._shutdown.set()
         self._discovery_thread.join()
 
diff --git a/horovod/run/elastic/registration.py b/horovod/run/elastic/registration.py
index 15e4191cd4..12dff1bd07 100644
--- a/horovod/run/elastic/registration.py
+++ b/horovod/run/elastic/registration.py
@@ -18,15 +18,19 @@
 
 from collections import defaultdict
 
+from horovod.run.elastic import constants
+
 READY = 'READY'
 SUCCESS = 'SUCCESS'
 FAILURE = 'FAILURE'
 
 
 class WorkerStateRegistry(object):
-    def __init__(self, driver, host_manager, verbose=False):
+    def __init__(self, driver, host_manager, reset_limit=None, verbose=False):
         self._driver = driver
         self._host_manager = host_manager
+        self._reset_limit = reset_limit
+        self._reset_count = 0
         self._lock = threading.Lock()
         self._states = {}
         self._workers = defaultdict(set)
@@ -100,7 +104,8 @@ def _record_state(self, host, slot, state):
                 logging.info('record state: {}[{}] = {}'.format(host, slot, state))
                 self._states[key] = state
                 self._workers[state].add(key)
-                rendezvous_id = self._rendezvous_id
+
+            rendezvous_id = self._rendezvous_id
 
         rendezvous_id = self._wait(key, state, rendezvous_id)
         return rendezvous_id
@@ -130,6 +135,13 @@ def _action(self):
     def _on_workers_recorded(self):
         logging.info('all {} workers recorded'.format(self.size()))
 
+        # Check that we have already reset the maximum number of allowed times
+        if self._reset_limit is not None and self._reset_count > self._reset_limit:
+            logging.error('reset count {} has exceeded limit {} -> stop running'
+                          .format(self._reset_count, self._reset_limit))
+            self._driver.stop(error_message=constants.RESET_LIMIT_EXCEEDED_MESSAGE.format(self._reset_limit))
+            return
+
         # Check for success state, if any process succeeded, shutdown all other processes
         if self.count(SUCCESS) > 0:
             logging.info('success count == {} -> stop running'.format(self.count(SUCCESS)))
@@ -154,6 +166,7 @@ def _on_workers_recorded(self):
             return
 
         try:
+            self._reset_count += 1
             self._driver.resume()
         except Exception:
             logging.exception('failed to activate new hosts -> stop running')
diff --git a/horovod/run/elastic/settings.py b/horovod/run/elastic/settings.py
index d9c263e5f2..b909b7396c 100644
--- a/horovod/run/elastic/settings.py
+++ b/horovod/run/elastic/settings.py
@@ -17,7 +17,7 @@
 
 
 class ElasticSettings(BaseSettings):
-    def __init__(self, discovery, min_np, max_np, elastic_timeout, **kwargs):
+    def __init__(self, discovery, min_np, max_np, elastic_timeout, reset_limit, **kwargs):
         """
         :param discovery: object used to detect and manage available hosts
         :type discovery: horovod.run.elastic.discovery.HostDiscovery
@@ -27,12 +27,15 @@ def __init__(self, discovery, min_np, max_np, elastic_timeout, **kwargs):
         :type max_np: int
         :param elastic_timeout: timeout for elastic initialisation after re-scaling in seconds
         :type elastic_timeout: int
+        :param reset_limit: maximum number of resets after which the job is terminated
+        :type reset_limit: int
         """
         super(ElasticSettings, self).__init__(elastic=True, **kwargs)
         self.discovery = discovery
         self.min_np = min_np
         self.max_np = max_np
         self.elastic_timeout = elastic_timeout
+        self.reset_limit = reset_limit
 
     # we do not serialize the discovery instance
     # it is not needed on the worker and might not be serializable
diff --git a/horovod/run/gloo_run.py b/horovod/run/gloo_run.py
index 251503104c..c726cae48f 100644
--- a/horovod/run/gloo_run.py
+++ b/horovod/run/gloo_run.py
@@ -281,6 +281,7 @@ def launch_gloo_elastic(command, exec_command, settings, env, get_common_interfa
     driver = ElasticDriver(rendezvous, settings.discovery,
                            settings.min_np, settings.max_np,
                            timeout=settings.elastic_timeout,
+                           reset_limit=settings.reset_limit,
                            verbose=settings.verbose)
 
     handler = create_rendezvous_handler(driver)
@@ -299,7 +300,10 @@ def launch_gloo_elastic(command, exec_command, settings, env, get_common_interfa
     driver.stop()
     rendezvous.stop()
 
-    for name, value in sorted(res.items(), key=lambda item: item[1][1]):
+    if res.error_message is not None:
+        raise RuntimeError(res.error_message)
+
+    for name, value in sorted(res.worker_results.items(), key=lambda item: item[1][1]):
         exit_code, timestamp = value
         if exit_code != 0:
             raise RuntimeError('Horovod detected that one or more processes exited with non-zero '
diff --git a/horovod/run/runner.py b/horovod/run/runner.py
index d2318d5a89..8da77f5b92 100644
--- a/horovod/run/runner.py
+++ b/horovod/run/runner.py
@@ -363,6 +363,9 @@ def parse_args():
                                     'The default value is 600 seconds. Alternatively, '
                                     'The environment variable HOROVOD_ELASTIC_TIMEOUT '
                                     'can also be used to.')
+    group_elastic.add_argument('--reset-limit', action='store', dest='reset_limit', type=int,
+                               help='Maximum number of times that the training job can scale up or down '
+                                    'the number of workers after which the job is terminated. (default: None)')
 
     group_timeline = parser.add_argument_group('timeline arguments')
     group_timeline.add_argument('--timeline-filename', action=make_override_action(override_args),
@@ -519,6 +522,7 @@ def __init__(self):
         self.max_np = None
         self.slots = None
         self.elastic_timeout = None
+        self.reset_limit = None
 
         # timeline arguments
         self.timeline_filename = None
@@ -702,15 +706,16 @@ def _run_elastic(args):
                                     'may need to increase the --start-timeout '
                                     'parameter if you have too many servers.')
     settings = elastic_settings.ElasticSettings(discovery=discover_hosts,
-                                                num_proc=args.np,
                                                 min_np=args.min_np or args.np,
                                                 max_np=args.max_np,
+                                                elastic_timeout=args.elastic_timeout,
+                                                reset_limit=args.reset_limit,
+                                                num_proc=args.np,
                                                 verbose=2 if args.verbose else 0,
                                                 ssh_port=args.ssh_port,
                                                 extra_mpi_args=args.mpi_args,
                                                 key=secret.make_secret_key(),
                                                 start_timeout=tmout,
-                                                elastic_timeout=args.elastic_timeout,
                                                 output_filename=args.output_filename,
                                                 run_func_mode=args.run_func is not None,
                                                 nics=args.nics)
@@ -830,6 +835,7 @@ def run(
         min_np=None,
         max_np=None,
         slots=None,
+        reset_limit=None,
         hosts=None,
         hostfile=None,
         start_timeout=None,
@@ -858,6 +864,7 @@ def run(
     :param slots: Number of slots for processes per host. Normally 1 slot per GPU per host.
                   If slots are provided by the output of the host discovery script, then that
                   value will override this parameter.
+    :param reset_limit: Maximum number of reset events after which the job is terminated.
 
     :param hosts: List of host names and the number of available slots
                   for running processes on each, of the form: <hostname>:<slots>
@@ -913,6 +920,7 @@ def wrapped_func():
     hargs.min_np = min_np
     hargs.max_np = max_np
     hargs.slots = slots
+    hargs.reset_limit = reset_limit
     hargs.hosts = hosts
     hargs.hostfile = hostfile
     hargs.start_timeout = start_timeout
diff --git a/horovod/spark/runner.py b/horovod/spark/runner.py
index a023474994..e67a122774 100644
--- a/horovod/spark/runner.py
+++ b/horovod/spark/runner.py
@@ -292,7 +292,7 @@ def run(fn, args=(), kwargs={}, num_proc=None, start_timeout=None,
 
 
 def run_elastic(fn, args=(), kwargs={}, num_proc=None, min_np=None, max_np=None,
-                start_timeout=None, elastic_timeout=None, env=None, verbose=1, nics=None):
+                start_timeout=None, elastic_timeout=None, reset_limit=None, env=None, verbose=1, nics=None):
     """
     Runs Elastic Horovod in Spark.  Runs `num_proc` processes executing `fn` using the same amount of Spark tasks.
 
@@ -307,6 +307,7 @@ def run_elastic(fn, args=(), kwargs={}, num_proc=None, min_np=None, max_np=None,
         elastic_timeout: Timeout for elastic initialisation after re-scaling the cluster.
                        If not set, falls back to `HOROVOD_ELASTIC_TIMEOUT` environment variable value.
                        If it is not set as well, defaults to 600 seconds.
+        reset_limit: Maximum number of resets after which the job is terminated.
         env: Environment dictionary to use in Horovod run.  Defaults to `os.environ`.
         verbose: Debug output verbosity (0-2). Defaults to 1.
         nics: List of NICs for tcp network communication.
@@ -361,14 +362,15 @@ def run_elastic(fn, args=(), kwargs={}, num_proc=None, min_np=None, max_np=None,
                                     'process runs in a Spark task. You may need to increase the '
                                     'start_timeout parameter to a larger value if your Spark resources '
                                     'are allocated on-demand.')
-    settings = hvd_elastic_settings.ElasticSettings(num_proc=num_proc,
+    settings = hvd_elastic_settings.ElasticSettings(discovery=discovery,
                                                     min_np=min_np,
                                                     max_np=max_np,
-                                                    discovery=discovery,
+                                                    elastic_timeout=elastic_timeout,
+                                                    reset_limit=reset_limit,
+                                                    num_proc=num_proc,
                                                     verbose=verbose,
                                                     key=key,
                                                     start_timeout=tmout,
-                                                    elastic_timeout=elastic_timeout,
                                                     nics=nics,
                                                     run_func_mode=True)
 
diff --git a/test/integration/elastic_common.py b/test/integration/elastic_common.py
index ef29149dca..35d10e8d40 100644
--- a/test/integration/elastic_common.py
+++ b/test/integration/elastic_common.py
@@ -22,6 +22,7 @@
 import pytest
 
 from horovod.run.common.util import config_parser
+from horovod.run.elastic import constants
 from horovod.run.runner import parse_args, _run_elastic
 
 sys.path.append(os.path.join(os.path.dirname(__file__), os.pardir))
@@ -69,7 +70,7 @@ def __init__(self, training_script, *args, **kwargs):
         super(BaseElasticTests, self).__init__(*args, **kwargs)
 
     def _run(self, discovery_schedule=None, exit_schedule=None, exit_mode='exception',
-             np=2, min_np=2, max_np=4, hosts=None):
+             np=2, min_np=2, max_np=4, hosts=None, reset_limit=None):
         if not discovery_schedule and not hosts:
             raise ValueError('at least one of discovery schedule or hosts must be given')
 
@@ -86,6 +87,9 @@ def _run(self, discovery_schedule=None, exit_schedule=None, exit_mode='exception
                     command_args += ['--host-discovery-script', discovery_script,
                                      '--max-np', str(max_np)]
 
+                if reset_limit is not None:
+                    command_args += ['--reset-limit', str(reset_limit)]
+
                 command_args += ['python', self._training_script, '--logfile', logfile]
                 if discovery_schedule:
                     command_args += ['--discovery-schedule', json.dumps(discovery_schedule)]
@@ -235,3 +239,22 @@ def test_min_hosts_timeout(self, mock_get_min_start_hosts):
         message = 'Horovod detected that one or more processes exited with non-zero status'
         with pytest.raises(RuntimeError, match=message):
             self._run(discovery_schedule, exit_schedule=exit_schedule, np=4, min_np=4)
+
+    @mock.patch('horovod.run.elastic.driver.ELASTIC_TIMEOUT_SECS', 1)
+    @mock.patch('horovod.run.elastic.driver.DISCOVER_HOSTS_FREQUENCY_SECS', 0.01)
+    @mock.patch('horovod.run.gloo_run._get_min_start_hosts', return_value=1)
+    def test_reset_limit(self, mock_get_min_start_hosts):
+        discovery_schedule = [
+            (0, ['localhost:{}'.format(2)]),
+            (1, ['localhost:{}'.format(2), '127.0.0.1:{}'.format(2)]),
+            (None, ['127.0.0.1:{}'.format(2)]),
+        ]
+
+        # Job should fail with reset_limit=1
+        message = constants.RESET_LIMIT_EXCEEDED_MESSAGE.format(1)
+        with pytest.raises(RuntimeError, match=message):
+            self._run(discovery_schedule, np=2, min_np=2, max_np=4, reset_limit=1)
+
+        # Job should succeed with reset_limit=2
+        results = self._run(discovery_schedule, np=2, min_np=2, max_np=4, reset_limit=2)
+        assert len(results) == 3
diff --git a/test/test_elastic_driver.py b/test/test_elastic_driver.py
index 9cd86c3417..4ec1608060 100644
--- a/test/test_elastic_driver.py
+++ b/test/test_elastic_driver.py
@@ -69,7 +69,7 @@ def exec_command(slot_info, events):
             return 0, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 4
@@ -100,7 +100,7 @@ def exec_command(slot_info, events):
             return 0, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 2
@@ -146,7 +146,7 @@ def exec_command(slot_info, events):
             return 0, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 4
@@ -212,7 +212,7 @@ def exec_command(slot_info, events):
             return 1, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 4
@@ -236,7 +236,7 @@ def exec_command(slot_info, events):
             return 1, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 4
@@ -271,7 +271,7 @@ def exec_command(slot_info, events):
             return 0, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 2
@@ -348,7 +348,7 @@ def exec_command(slot_info, events):
             return 0, time.time()
 
         driver.start(np=2, create_worker_fn=exec_command)
-        res = driver.get_results()
+        res = driver.get_results().worker_results
         driver.stop()
 
         assert len(res) == 2


Code review comments:
diff_hunk: @@ -130,6 +135,13 @@ def _action(self):
     def _on_workers_recorded(self):
         logging.info('all {} workers recorded'.format(self.size()))
 
+        # Check that we have already reset the maximum number of allowed times
path: horovod/run/elastic/registration.py
text: @user1:
I think we can reset once too often and then fail here even if all workers succeed if we check the limit here. We should check it right before we increment the counter further down, and then for `≥`.

@author:
Yes, good point.  Fixed.

diff_hunk: @@ -235,3 +239,22 @@ def test_min_hosts_timeout(self, mock_get_min_start_hosts):
         message = 'Horovod detected that one or more processes exited with non-zero status'
         with pytest.raises(RuntimeError, match=message):
             self._run(discovery_schedule, exit_schedule=exit_schedule, np=4, min_np=4)
+
+    @mock.patch('horovod.run.elastic.driver.ELASTIC_TIMEOUT_SECS', 1)
+    @mock.patch('horovod.run.elastic.driver.DISCOVER_HOSTS_FREQUENCY_SECS', 0.01)
+    @mock.patch('horovod.run.gloo_run._get_min_start_hosts', return_value=1)
+    def test_reset_limit(self, mock_get_min_start_hosts):
+        discovery_schedule = [
+            (0, ['localhost:{}'.format(2)]),
path: test/integration/elastic_common.py
text: @user1:
why the `format(2)`?

@author:
Copy-paste ;)