hunk.code_window:         runner.assert_has_call("1.2.3.4", "init_cmd")
           runner.assert_has_call("1.2.3.4", "head_setup_cmd")
           runner.assert_has_call("1.2.3.4", "start_ray_head")
           self.assertEqual(self.provider.mock_nodes[0].node_type, None)
       def testScaleUp(self):
           config_path = self.write_config(SMALL_CLUSTER)
  
hunk.labels: keep keep keep add keep keep keep
hunk.after_edit:         runner.assert_has_call("1.2.3.4", pattern="docker run")
           runner.assert_not_has_call(
               "1.2.3.4", pattern="-v /tmp/ray_tmp_mount/~/ray_bootstrap_config")
           runner.assert_has_call(
               "1.2.3.4",
               pattern="docker cp /tmp/ray_tmp_mount/~/ray_bootstrap_key.pem")
           runner.assert_has_call(
               "1.2.3.4",
               pattern="docker cp /tmp/ray_tmp_mount/~/ray_bootstrap_config.yaml")
  
hunk.file_path: python/ray/tests/test_autoscaler.py
hunk.type: add
file: import numpy as np
  import time
  
  
  class _Timer:
      """A running stat for conveniently logging the duration of a code block.
  
      Example:
          wait_timer = TimerStat()
          with wait_timer:
              ray.wait(...)
  
      Note that this class is *not* thread-safe.
      """
  
      def __init__(self, window_size=10):
          self._window_size = window_size
          self._samples = []
          self._units_processed = []
          self._start_time = None
          self._total_time = 0.0
          self.count = 0
  
      def __enter__(self):
          assert self._start_time is None, "concurrent updates not supported"
          self._start_time = time.time()
  
      def __exit__(self, exc_type, exc_value, tb):
          assert self._start_time is not None
          time_delta = time.time() - self._start_time
          self.push(time_delta)
          self._start_time = None
  
      def push(self, time_delta):
          self._samples.append(time_delta)
          if len(self._samples) > self._window_size:
              self._samples.pop(0)
          self.count += 1
          self._total_time += time_delta
  
      def push_units_processed(self, n):
          self._units_processed.append(n)
          if len(self._units_processed) > self._window_size:
              self._units_processed.pop(0)
  
      def has_units_processed(self):
          return len(self._units_processed) > 0
  
      @property
      def mean(self):
          if not self._samples:
              return 0.0
          return float(np.mean(self._samples))
  
      @property
      def mean_units_processed(self):
          if not self._units_processed:
              return 0.0
          return float(np.mean(self._units_processed))
  
      @property
      def mean_throughput(self):
          time_total = float(sum(self._samples))
          if not time_total:
              return 0.0
          return float(sum(self._units_processed)) / time_total
  