code: def launch_jobs(
      launcher: RayAWSLauncher, local_tmp_dir: str, sweep_dir: Path
  ) -> Sequence[JobReturn]:
      config = OmegaConf.to_container(
          launcher.ray_cfg.cluster, resolve=True, enum_to_str=True
      )
      sdk.create_or_update_cluster(
          config,
          **launcher.create_update_cluster,
      )
      with tempfile.TemporaryDirectory() as local_tmp_download_dir:
          assert isinstance(config, dict)
          with ray_tmp_dir(
              config, launcher.ray_cfg.run_env.name
          ) as remote_tmp_dir:
              sdk.rsync(
                  config,
                  source=os.path.join(local_tmp_dir, ""),
                  target=remote_tmp_dir,
                  down=False,
              )
  
              script_path = os.path.join(
                  os.path.dirname(__file__), "_remote_invoke.py"
              )
              remote_script_path = os.path.join(
                  remote_tmp_dir, "_remote_invoke.py"
              )
              sdk.rsync(
                  config,
                  source=script_path,
                  target=remote_script_path,
                  down=False,
              )
  
              if launcher.sync_up.source_dir:
                  sync_up_source_dir = _get_abs_code_dir(
                      launcher.sync_up.source_dir
                  )
                  target_dir = (
                      launcher.sync_up.target_dir
                      if launcher.sync_up.target_dir
                      else remote_tmp_dir
                  )
                  rsync(
                      config,
                      launcher.sync_up.include,
                      launcher.sync_up.exclude,
                      os.path.join(sync_up_source_dir, ""),
                      target_dir,
                  )
              sdk.run_on_cluster(
                  config,
                  run_env=launcher.ray_cfg.run_env.name,
                  cmd=f"python {remote_script_path} {remote_tmp_dir}",
              )
  
              sdk.rsync(
                  config,
                  source=os.path.join(
                      remote_tmp_dir, JOB_RETURN_PICKLE
                  ),
                  target=local_tmp_download_dir,
                  down=True,
              )
  
              sync_down_cfg = launcher.sync_down
  
              if (
                  sync_down_cfg.target_dir
                  or sync_down_cfg.source_dir
                  or sync_down_cfg.include
                  or sync_down_cfg.exclude
              ):
                  sync_down_source_dir: Union[str, Path] = (
                      sync_down_cfg.source_dir
                      if sync_down_cfg.source_dir
                      else sweep_dir
                  )
                  target_dir = (
                      sync_down_cfg.target_dir
                      if sync_down_cfg.target_dir
                      else sweep_dir
                  )
                  target_dir = Path(_get_abs_code_dir(target_dir))
                  target_dir.mkdir(parents=True, exist_ok=True)
  
                  rsync(
                      config,
                      launcher.sync_down.include,
                      launcher.sync_down.exclude,
                      os.path.join(sync_down_source_dir),
                      str(target_dir),
                      up=False,
                  )
                  log.info(
                      f"Syncing outputs from remote dir: {sync_down_source_dir} to local dir: {target_dir.absolute()} "
                  )
  
          if launcher.stop_cluster:
              log.info("Stopping cluster now. (stop_cluster=true)")
              if launcher.ray_cfg.cluster.provider.cache_stopped_nodes:
                  log.info(
                      "NOT deleting the cluster (provider.cache_stopped_nodes=true)"
                  )
              else:
                  log.info(
                      "Deleted the cluster (provider.cache_stopped_nodes=false)"
                  )
              sdk.teardown_cluster(
                  config,
                  **launcher.teardown_cluster,
              )
          else:
              log.warning(
                  "NOT stopping cluster, this may incur extra cost for you. (stop_cluster=false)"
              )
  
          with open(
              os.path.join(local_tmp_download_dir, JOB_RETURN_PICKLE),
              "rb",
          ) as f:
              job_returns = pickle.load(f)  # nosec
              assert isinstance(job_returns, List)
              for run in job_returns:
                  assert isinstance(run, JobReturn)
              return job_returns
  
file_path: hydra/plugins/hydra_ray_launcher/hydra_plugins/hydra_ray_launcher/_core_aws.py
imports: import logging
  import os
  import tempfile
  from pathlib import Path
  from typing import Any, List, Sequence, Union
  import cloudpickle
  from hydra.core.singleton import Singleton
  from hydra.core.utils import JobReturn, configure_log, filter_overrides, setup_globals
  from omegaconf import Node, OmegaConf, open_dict, read_write
  from ._launcher_util import JOB_RETURN_PICKLE, JOB_SPEC_PICKLE, ray_tmp_dir, rsync
  from .ray_aws_launcher import RayAWSLauncher
label_desc: Use image node_id 4 for calling a global function with example usage: launch_jobs(launcher, local_tmp_dir, sweep_dir) and returns: job_returns