Introduce file_mounts_sync_continuously cluster option (#9544)

* Separate out file_mounts contents hashing into its own separate hash

Add an option to continuously sync file_mounts from head node to worker nodes:
monitor.py will re-sync file mounts whenver contents change but will only run setup_commands if the config also changes

* add test and default value for file_mounts_sync_continuously

* format code

* Update comments

* Add param to skip setup commands when only file_mounts content changed during monitor.py's update tick

Fixed so setup commands run when ray up is run and file_mounts content changes

* 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

* fix issue with hashing a hash

* fix bug where trying to set contents hash when it wasn't generated

* Fix lint error

Fix bug in command_runner where check_output was no longer returning the output of the command

* clear out provider between tests to get rid of flakyness

* reduce chance of race condition from node_launcher launching a node in the middle of an autoscaler.update call

python/ray/autoscaler/commands.py

@@ -373,6 +373,7 @@ def kill_node(config_file, yes, hard, override_cluster_name):
                 setup_commands=[],
                 ray_start_commands=[],
                 runtime_hash="",
+                file_mounts_contents_hash="",
                 docker_config=config.get("docker"))
 
             _exec(updater, "ray stop", False, False)
@@ -540,7 +541,8 @@ def get_or_create_head_node(config, config_file, no_restart, restart_only, yes,
             # TODO(ekl) right now we always update the head node even if the
             # hash matches.
             # We could prompt the user for what they want to do here.
-            runtime_hash = hash_runtime_conf(config["file_mounts"], config)
+            (runtime_hash, file_mounts_contents_hash) = hash_runtime_conf(
+                config["file_mounts"], config)
 
             cli_logger.old_info(
                 logger,
@@ -604,6 +606,7 @@ def get_or_create_head_node(config, config_file, no_restart, restart_only, yes,
                 setup_commands=init_commands,
                 ray_start_commands=ray_start_commands,
                 runtime_hash=runtime_hash,
+                file_mounts_contents_hash=file_mounts_contents_hash,
                 docker_config=config.get("docker"))
             updater.start()
             updater.join()
@@ -740,6 +743,7 @@ def exec_cluster(config_file: str,
             setup_commands=[],
             ray_start_commands=[],
             runtime_hash="",
+            file_mounts_contents_hash="",
             docker_config=config.get("docker"))
 
         is_docker = isinstance(updater.cmd_runner, DockerCommandRunner)
@@ -863,6 +867,7 @@ def rsync(config_file: str,
                 setup_commands=[],
                 ray_start_commands=[],
                 runtime_hash="",
+                file_mounts_contents_hash="",
                 docker_config=config.get("docker"))
             if down:
                 rsync = updater.rsync_down