def run()

in integration_test/soak_test/cmd/launcher/log_generator.py [0:0]

• 15 lines of code
• 8 McCabe index (conditional complexity)

  def run(self, count: int):
    """Generate logs.

    The high level idea here is that we sit in a loop waiting until a log entry
    can be printed, and then print it. The benefits of this approach are:
    * No skew overtime as the log times are calculated against the
      start_time.
    * Precise detection of overruns. The code knows exactly how far behind
      schedule it is, since it knows when every log should be printed.
    * Can allow more complex logging patterns like dump logs every
      minute instead of second, ramping up log generation, or periodical log
      throughputs.

    Args:
      count: how many logs to produce. If count <= 0, produce logs forever.
    """
    # last_checked_time is used to optimize away time.time calls.
    last_checked_time = 0
    for n in itertools.count(start=1):
      expected_nth_log_time = self.log_distribution.get_nth_time(n)
      # If last_checked_time >= expected_nth_log_time,
      # there is no need for any waiting logic.
      if expected_nth_log_time > last_checked_time:
        while expected_nth_log_time > time.time():
          # Sleep until approximately 5ms before the next entry should be
          # generated.
          time.sleep(max(expected_nth_log_time - time.time() - 0.05, 0))
        last_checked_time = time.time()
        if last_checked_time - expected_nth_log_time > 1:
          # We only print delays after checking the wall clock
          # and seeing we are more than 1 second behind.
          logger.error(
              'Detected overruns. Log Generator %s seconds behind schedule.',
              last_checked_time - expected_nth_log_time)
      self.log_writer.write_log(self.log_distribution.get_nth_log(n))
      if count == n:
        break