code: def run_and_wait(
      self,
      cmd_to_run,
      env_vars,
      working_dir: Optional[str] = None,
      timeout: int = 120,
      upload_path: Optional[str] = None,
      pip: Optional[List[str]] = None,
  ) -> Tuple[int, float]:
      self._run_job(
          cmd_to_run,
          env_vars,
          working_dir=working_dir,
          upload_path=upload_path,
          pip=pip,
      )
      return self._wait_job(timeout)
  
imports: import os
  import time
  import subprocess
  import tempfile
  from collections import deque
  from contextlib import contextmanager
  from typing import Any, Dict, Optional, Tuple, List
  from anyscale.sdk.anyscale_client.models import CreateProductionJob, HaJobStates
  from ray_release.anyscale_util import LAST_LOGS_LENGTH, get_cluster_name
  from ray_release.cluster_manager.cluster_manager import ClusterManager
  from ray_release.exception import CommandTimeout, JobStartupTimeout, JobStartupFailed
  from ray_release.logger import logger
  from ray_release.signal_handling import register_handler, unregister_handler
  from ray_release.util import ANYSCALE_HOST, ERROR_LOG_PATTERNS, exponential_backoff_retry, anyscale_job_url, format_link
label_desc: Use image node_id 14 for calling the AnyscaleJobManager obj's underlying member method code with example usage: obj.run_and_wait(cmd_to_run, env_vars, working_dir, timeout, upload_path, pip) and returns: self