[autoscaler] Revert to double-spawning updater threads (#5903)

* [autoscaler] Revert to double-spawning threads

* Use log prefix

* add comment

python/ray/autoscaler/autoscaler.py

@@ -553,11 +553,22 @@ class StandardAutoscaler(object):
             nodes = self.workers()
             self.log_info_string(nodes, target_workers)
 
-        # Update nodes with out-of-date files
+        # Update nodes with out-of-date files.
+        # TODO(edoakes): Spawning these threads directly seems to cause
+        # problems. They should at a minimum be spawned as daemon threads.
+        # See https://github.com/ray-project/ray/pull/5903 for more info.
+        T = []
         for node_id, commands, ray_start in (self.should_update(node_id)
                                              for node_id in nodes):
             if node_id is not None:
-                self.spawn_updater(node_id, commands, ray_start)
+                T.append(
+                    threading.Thread(
+                        target=self.spawn_updater,
+                        args=(node_id, commands, ray_start)))
+        for t in T:
+            t.start()
+        for t in T:
+            t.join()
 
         # Attempt to recover unhealthy nodes
         for node_id in nodes:

python/ray/autoscaler/updater.py

@@ -430,9 +430,9 @@ class NodeUpdater(object):
 
         node_tags = self.provider.node_tags(self.node_id)
         if node_tags.get(TAG_RAY_RUNTIME_CONFIG) == self.runtime_hash:
-            logger.info(
-                "NodeUpdater: {} already up-to-date, skip to ray start".format(
-                    self.node_id))
+            logger.info(self.log_prefix +
+                        "{} already up-to-date, skip to ray start".format(
+                            self.node_id))
         else:
             self.provider.set_node_tags(
                 self.node_id, {TAG_RAY_NODE_STATUS: STATUS_SYNCING_FILES})