Master will not add any worker if the current rendezvous hosts become empty after starting training. (#2508)

* Master will not add any worker if the rendezvous hosts become empty

* Don't add a new worker host if the next group is empty

* Fix the error message

* Split a long function

* Fix a typo

Author: workingloong

elasticai_api/common/base_controller.py

@@ -63,6 +63,8 @@ def init_horovod_if_needed(self):
                 time.sleep(RETRY_ALLREDUCE_INTERVAL_SECS)
             else:
                 break
+        if rank_response.rank_id < 0:
+            raise ValueError("Invalid rank {}".format(rank_response.rank_id))
 
         # If the rendezvous from master is unequal to self._rendezvous_id,
         # the worker should rebuild the communication because the master
@@ -73,19 +75,22 @@ def init_horovod_if_needed(self):
                     rank_response.rank_id, rank_response.world_size
                 )
             )
-            os.environ[HorovodEnv.RENDEZVOUS_PORT] = str(
-                rank_response.rendezvous_port
-            )
-            os.environ[HorovodEnv.RANK] = str(rank_response.rank_id)
-            os.environ[HorovodEnv.SIZE] = str(rank_response.world_size)
-            # Not using Horovod elastic feature in init, but need it for
-            # allreduce to call allreduce op when size=1.
-            os.environ[HorovodEnv.ELASTIC] = str(0)
-            hvd.shutdown()
-            hvd.init()
-            os.environ[HorovodEnv.ELASTIC] = str(1)
-            self._rendezvous_id = rank_response.rendezvous_id
-            self.need_broadcast = True
+            self._restart_hvd(rank_response)
+
+    def _restart_hvd(self, rank_response):
+        os.environ[HorovodEnv.RENDEZVOUS_PORT] = str(
+            rank_response.rendezvous_port
+        )
+        os.environ[HorovodEnv.RANK] = str(rank_response.rank_id)
+        os.environ[HorovodEnv.SIZE] = str(rank_response.world_size)
+        # Not using Horovod elastic feature in init, but need it for
+        # allreduce to call allreduce op when size=1.
+        os.environ[HorovodEnv.ELASTIC] = str(0)
+        hvd.shutdown()
+        hvd.init()
+        os.environ[HorovodEnv.ELASTIC] = str(1)
+        self._rendezvous_id = rank_response.rendezvous_id
+        self.need_broadcast = True
 
     def _set_horovod_env(self):
         master_addr_port = os.getenv(WorkerEnv.MASTER_ADDR, None)
@@ -103,7 +108,7 @@ def notify_training_loop_status(self, status):
 
 class AllReduceController(object):
     """The controller initializes Horovod and calls the function with forward
-    and backward computation using a mini-batch of data. If Horovod raise an
+    and backward computation using a mini-batch of data. If Horovod raises an
     exception about AllReduce, Allgather and Broadcast, the controller will
     catch the exception and re-initialize Horovod. Then, it will broadcast
     the variables and retry to call those functions.

elasticdl/python/master/rendezvous_server.py

@@ -145,6 +145,10 @@ def add_worker(self, worker_host):
                     self._next_rendezvous_hosts = copy.deepcopy(
                         self._cur_rendezvous_hosts
                     )
+                # Master will not add any worker if the current rendezvous
+                # hosts become empty after starting training.
+                if self._rendezvous_id > 0 and not self._next_rendezvous_hosts:
+                    return
                 if worker_host not in self._next_rendezvous_hosts:
                     self._next_rendezvous_hosts.append(worker_host)