code: def run_init(
      self,
      *,
      as_head: bool,
      file_mounts: Dict[str, str],
      sync_run_yet: bool,
  ):
      BOOTSTRAP_MOUNTS = [
          "~/ray_bootstrap_config.yaml",
          "~/ray_bootstrap_key.pem",
      ]
  
      specific_image = self.docker_config.get(
          f"{'head' if as_head else 'worker'}_image",
          self.docker_config.get("image"),
      )
  
      self._check_docker_installed()
      if self.docker_config.get("pull_before_run", True):
          assert specific_image, (
              "Image must be included in config if "
              + "pull_before_run is specified"
          )
          self.run(
              "{} pull {}".format(self.docker_cmd, specific_image),
              run_env="host",
          )
      else:
          self.run(
              f"{self.docker_cmd} image inspect {specific_image} "
              "1> /dev/null  2>&1 || "
              f"{self.docker_cmd} pull {specific_image}"
          )
  
      # Bootstrap files cannot be bind mounted because docker opens the
      # underlying inode. When the file is switched, docker becomes outdated.
      cleaned_bind_mounts = file_mounts.copy()
      for mnt in BOOTSTRAP_MOUNTS:
          cleaned_bind_mounts.pop(mnt, None)
  
      docker_run_executed = False
  
      container_running = self._check_container_status()
      requires_re_init = False
      if container_running:
          requires_re_init = self._check_if_container_restart_is_needed(
              specific_image, cleaned_bind_mounts
          )
          if requires_re_init:
              self.run(
                  f"{self.docker_cmd} stop {self.container_name}",
                  run_env="host",
              )
  
      if (not container_running) or requires_re_init:
          if not sync_run_yet:
              # Do not start the actual image as we need to run file_sync
              # first to ensure that all folders are created with the
              # correct ownership. Docker will create the folders with
              # `root` as the owner.
              return True
          # Get home directory
          image_env = (
              self.ssh_command_runner.run(
                  f"{self.docker_cmd} "
                  + "inspect -f '{{json .Config.Env}}' "
                  + specific_image,
                  with_output=True,
              )
              .decode()
              .strip()
          )
          home_directory = "/root"
          try:
              for env_var in json.loads(image_env):
                  if env_var.startswith("HOME="):
                      home_directory = env_var.split("HOME=")[1]
                      break
          except json.JSONDecodeError as e:
              cli_logger.error(
                  "Unable to deserialize `image_env` to Python object. "
                  f"The `image_env` is:\n{image_env}"
              )
              raise e
  
          user_docker_run_options = self.docker_config.get(
              "run_options", []
          ) + self.docker_config.get(
              f"{'head' if as_head else 'worker'}_run_options", []
          )
          start_command = docker_start_cmds(
              self.ssh_command_runner.ssh_user,
              specific_image,
              cleaned_bind_mounts,
              self.container_name,
              self._configure_runtime(
                  self._auto_configure_shm(user_docker_run_options)
              ),
              self.ssh_command_runner.cluster_name,
              home_directory,
              self.docker_cmd,
          )
          self.run(start_command, run_env="host")
          docker_run_executed = True
  
      # Explicitly copy in ray bootstrap files.
      for mount in BOOTSTRAP_MOUNTS:
          if mount in file_mounts:
              if not sync_run_yet:
                  # NOTE(ilr) This rsync is needed because when starting from
                  #  a stopped instance,  /tmp may be deleted and `run_init`
                  # is called before the first `file_sync` happens
                  self.run_rsync_up(file_mounts[mount], mount)
              self.ssh_command_runner.run(
                  "rsync -e '{cmd} exec -i' -avz {src} {container}:{dst}".format(
                      cmd=self.docker_cmd,
                      src=os.path.join(
                          self._get_docker_host_mount_location(
                              self.ssh_command_runner.cluster_name
                          ),
                          mount,
                      ),
                      container=self.container_name,
                      dst=self._docker_expand_user(mount),
                  )
              )
              try:
                  # Check if the current user has read permission.
                  # If they do not, try to change ownership!
                  self.run(
                      f"cat {mount} >/dev/null 2>&1 || "
                      f"sudo chown $(id -u):$(id -g) {mount}"
                  )
              except Exception:
                  lsl_string = (
                      self.run(f"ls -l {mount}", with_output=True)
                      .decode("utf-8")
                      .strip()
                  )
                  # The string is of format <Permission> <Links>
                  # <Owner> <Group> <Size> <Date> <Name>
                  permissions = lsl_string.split(" ")[0]
                  owner = lsl_string.split(" ")[2]
                  group = lsl_string.split(" ")[3]
                  current_user = (
                      self.run("whoami", with_output=True)
                      .decode("utf-8")
                      .strip()
                  )
                  cli_logger.warning(
                      f"File ({mount}) is owned by user:{owner} and group:"
                      f"{group} with permissions ({permissions}). The "
                      f"current user ({current_user}) does not have "
                      "permission to read these files, and Ray may not be "
                      "able to autoscale. This can be resolved by "
                      "installing `sudo` in your container, or adding a "
                      f"command like 'chown {current_user} {mount}' to "
                      "your `setup_commands`."
                  )
      self.initialized = True
      return docker_run_executed
  
imports: import hashlib
  import json
  import logging
  import os
  import subprocess
  import sys
  import time
  from getpass import getuser
  from shlex import quote
  from typing import Dict, List
  import click
  from ray.autoscaler._private.cli_logger import cf, cli_logger
  from ray.autoscaler._private.constants import AUTOSCALER_NODE_SSH_INTERVAL_S, AUTOSCALER_NODE_START_WAIT_S, DEFAULT_OBJECT_STORE_MAX_MEMORY_BYTES, DEFAULT_OBJECT_STORE_MEMORY_PROPORTION
  from ray.autoscaler._private.docker import check_bind_mounts_cmd, check_docker_image, check_docker_running_cmd, docker_start_cmds, with_docker_exec
  from ray.autoscaler._private.log_timer import LogTimer
  from ray.autoscaler._private.subprocess_output_util import ProcessRunnerError, is_output_redirected, run_cmd_redirected
  from ray.autoscaler.command_runner import CommandRunnerInterface
label_desc: Use image node_id 10 for calling the DockerCommandRunner obj's underlying member method code with example usage: obj.run_init() and returns: docker_run_executed, True