Refactor so that runtime_hash retains previous behavior

runtime_hash is almost identical as before this PR. It is used to determine if setup_commands need to run
file_mounts_contents_hash is an additional hash of the file_mounts content that is used to detect when only file syncing has to occur.

Note: runtime_hash value will have changed from before the PR because we hash the hash of the contents of the file_mounts as a performance optimization

python/ray/autoscaler/autoscaler.py

@@ -271,7 +271,8 @@ def reload_config(self, errors_fatal=False):
                      new_config["worker_setup_commands"],
                      new_config["worker_start_ray_commands"],
                  ],
-                 use_cached_contents_hash=not file_mounts_sync_continuously,
+                 generate_file_mounts_contents_hash=
+                 file_mounts_sync_continuously,
              )
             self.config = new_config
             self.launch_hash = new_launch_hash
@@ -328,8 +329,9 @@ def files_up_to_date(self, node_id):
         applied_file_mounts_contents_hash = node_tags.get(
             TAG_RAY_FILE_MOUNTS_CONTENTS)
         if (applied_config_hash != self.runtime_hash
-                or applied_file_mounts_contents_hash !=
-                self.file_mounts_contents_hash):
+                or (self.file_mounts_contents_hash is not None
+                    and self.file_mounts_contents_hash !=
+                    applied_file_mounts_contents_hash)):
             logger.info("StandardAutoscaler: "
                         "{}: Runtime state is ({},{}), want ({},{})".format(
                             node_id, applied_config_hash,
@@ -405,7 +407,6 @@ def spawn_updater(self, node_id, init_commands, ray_start_commands):
             ray_start_commands=with_head_node_ip(ray_start_commands),
             runtime_hash=self.runtime_hash,
             file_mounts_contents_hash=self.file_mounts_contents_hash,
-            run_setup_commands_on_file_mounts_change=False,
             process_runner=self.process_runner,
             use_internal_ip=True,
             docker_config=self.config.get("docker"))

python/ray/autoscaler/updater.py

@@ -33,7 +33,6 @@ def __init__(self,
                  ray_start_commands,
                  runtime_hash,
                  file_mounts_contents_hash,
-                 run_setup_commands_on_file_mounts_change=True,
                  process_runner=subprocess,
                  use_internal_ip=False,
                  docker_config=None):
@@ -58,10 +57,6 @@ def __init__(self,
         self.ray_start_commands = ray_start_commands
         self.runtime_hash = runtime_hash
         self.file_mounts_contents_hash = file_mounts_contents_hash
-        # Whether setup commands should be run when file_mounts contents
-        # have changed but otherwise the cluster config has not changed.
-        self.run_setup_commands_on_file_mounts_change = \
-            run_setup_commands_on_file_mounts_change
         self.auth_config = auth_config
 
     def run(self):
@@ -142,15 +137,12 @@ def do_update(self):
 
         node_tags = self.provider.node_tags(self.node_id)
         logger.debug("Node tags: {}".format(str(node_tags)))
-        # file_mounts_content_hash will only change whenever the user restarts
-        # or updates their cluster with `get_or_create_head_node` unless
-        # continuous mode is on. In that case, the file_mounts_content_hash
-        # will be recalculated at a regular interval while the cluster is
-        # running.
-        if node_tags.get(
-                TAG_RAY_RUNTIME_CONFIG) == self.runtime_hash and node_tags.get(
-                    TAG_RAY_FILE_MOUNTS_CONTENTS
-                ) == self.file_mounts_contents_hash:
+        # runtime_hash will only change whenever the user restarts
+        # or updates their cluster with `get_or_create_head_node`
+        if node_tags.get(TAG_RAY_RUNTIME_CONFIG) == self.runtime_hash and (
+                self.file_mounts_contents_hash is None
+                or node_tags.get(TAG_RAY_FILE_MOUNTS_CONTENTS) ==
+                self.file_mounts_contents_hash):
             logger.info(self.log_prefix +
                         "{} already up-to-date, skip to ray start".format(
                             self.node_id))
@@ -159,12 +151,10 @@ def do_update(self):
                 self.node_id, {TAG_RAY_NODE_STATUS: STATUS_SYNCING_FILES})
             self.sync_file_mounts(self.rsync_up)
 
-            # Skip setup commands when only file_mounts contents change unless
-            # run_setup_commands_on_file_mounts_change is True.
-            if node_tags.get(TAG_RAY_RUNTIME_CONFIG) != self.runtime_hash or (
-                    node_tags.get(TAG_RAY_FILE_MOUNTS_CONTENTS) !=
-                    self.file_mounts_contents_hash
-                    and self.run_setup_commands_on_file_mounts_change):
+            # Only run setup commands if runtime_hash has changed because
+            # we don't want to run setup_commands every time the head node
+            # file_mounts folders have changed.
+            if node_tags.get(TAG_RAY_RUNTIME_CONFIG) != self.runtime_hash:
                 # Run init commands
                 self.provider.set_node_tags(
                     self.node_id, {TAG_RAY_NODE_STATUS: STATUS_SETTING_UP})

python/ray/autoscaler/util.py

@@ -101,8 +101,20 @@ def hash_launch_conf(node_conf, auth):
 _hash_cache = {}
 
 
-def hash_runtime_conf(file_mounts, extra_objs, use_cached_contents_hash=True):
-    config_hasher = hashlib.sha1()
+def hash_runtime_conf(file_mounts,
+                      extra_objs,
+                      generate_file_mounts_contents_hash=False):
+    """Returns two hashes, a runtime hash and file_mounts_content hash.
+
+    The runtime hash is used to determine if the configuration or file_mounts
+    contents have changed. It is used at launch time (ray up) to determine if
+    a restart is needed.
+
+    The file_mounts_content hash is used to determine if the file_mounts
+    contents have changed. It is used at monitor time to determine if
+    additional file syncing is needed.
+    """
+    runtime_hasher = hashlib.sha1()
     contents_hasher = hashlib.sha1()
 
     def add_content_hashes(path):
@@ -128,12 +140,20 @@ def add_hash_of_file(fpath):
     conf_str = (json.dumps(file_mounts, sort_keys=True).encode("utf-8") +
                 json.dumps(extra_objs, sort_keys=True).encode("utf-8"))
 
-    # Only hash the files once, unless use_cached_contents_hash is false.
-    if not use_cached_contents_hash or conf_str not in _hash_cache:
-        config_hasher.update(conf_str)
+    # Only generate a contents hash if generate_contents_hash is true or
+    # if we need to generate the runtime_hash
+    if conf_str not in _hash_cache or generate_file_mounts_contents_hash:
         for local_path in sorted(file_mounts.values()):
             add_content_hashes(local_path)
-        _hash_cache[conf_str] = (config_hasher.hexdigest(),
-                                 contents_hasher.hexdigest())
+        contents_hash = contents_hasher.hexdigest()
 
-    return _hash_cache[conf_str]
+        # Generate a new runtime_hash if its not cached
+        if conf_str not in _hash_cache:
+            runtime_hasher.update(conf_str)
+            runtime_hasher.update(contents_hash)
+            _hash_cache[conf_str] = runtime_hasher.hexdigest()
+
+    else:
+        contents_hash = None
+
+    return (_hash_cache[conf_str], contents_hash)

python/ray/tests/test_autoscaler.py

@@ -1112,6 +1112,7 @@ def testContinuousFileMounts(self):
             2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
 
         for i in [0, 1]:
+            runner.assert_has_call("172.0.0.{}".format(i), "setup_cmd")
             runner.assert_has_call(
                 "172.0.0.{}".format(i),
                 "{}/ ubuntu@172.0.0.{}:/home/test-folder/".format(
@@ -1130,6 +1131,7 @@ def testContinuousFileMounts(self):
             2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
 
         for i in [0, 1]:
+            runner.assert_not_has_call("172.0.0.{}".format(i), "setup_cmd")
             runner.assert_has_call(
                 "172.0.0.{}".format(i),
                 "{}/ ubuntu@172.0.0.{}:/home/test-folder/".format(
@@ -1159,6 +1161,7 @@ def testFileMountsNonContinuous(self):
             2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
 
         for i in [0, 1]:
+            runner.assert_has_call("172.0.0.{}".format(i), "setup_cmd")
             runner.assert_has_call(
                 "172.0.0.{}".format(i),
                 "{}/ ubuntu@172.0.0.{}:/home/test-folder/".format(
@@ -1176,11 +1179,35 @@ def testFileMountsNonContinuous(self):
             2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
 
         for i in [0, 1]:
+            runner.assert_not_has_call("172.0.0.{}".format(i), "setup_cmd")
             runner.assert_not_has_call(
                 "172.0.0.{}".format(i),
                 "{}/ ubuntu@172.0.0.{}:/home/test-folder/".format(
                     file_mount_dir, i))
 
+        # Simulate a second `ray up` call
+        autoscaler = StandardAutoscaler(
+            config_path,
+            lm,
+            max_failures=0,
+            process_runner=runner,
+            update_interval_s=0)
+
+        autoscaler.update()
+        self.waitForNodes(2)
+        self.provider.finish_starting_nodes()
+        autoscaler.update()
+        self.waitForNodes(
+            2, tag_filters={TAG_RAY_NODE_STATUS: STATUS_UP_TO_DATE})
+
+        for i in [0, 1]:
+            runner.assert_has_call("172.0.0.{}".format(i), "setup_cmd")
+            runner.assert_has_call(
+                "172.0.0.{}".format(i),
+                "{}/ ubuntu@172.0.0.{}:/home/test-folder/".format(
+                    file_mount_dir, i))
+
+
 
 if __name__ == "__main__":
     import sys