class_docstr: A handler for check-ins for a given run.
  
  Uses filesystem to store the check-ins. Handles resumed `Run`s.
  Background thread is used to adjust the rate of check-ins written
  to the filesystem in order to avoid overloading.
  
  Calling `report_successful_finish()` is required to mark the run as
  finished, otherwise the run will be marked as failed.
class_docstr_tok: ["A","handler","for","check-ins","for","a","given","run",".","Uses","filesystem","to","store","the","check-ins",".","Handles","resumed","`","Run","`","s",".","Background","thread","is","used","to","adjust","the","rate","of","check-ins","written","to","the","filesystem","in","order","to","avoid","overloading",".","Calling","`","report_successful_finish","(",")","`","is","required","to","mark","the","run","as","finished",",","otherwise","the","run","will","be","marked","as","failed","."]
docstr: Increments the check-in counter and returns the new check-in for the
  provided flag name. The indices are shared across all the flags.
  This is an (almost) non-blocking call.
code: def _increment(
      self,
      *,
      expect_next_in: int = 0,
      flag_name: str,
  ) -> CheckIn:
      """
      Increments the check-in counter and returns the new check-in for the
      provided flag name. The indices are shared across all the flags.
      This is an (almost) non-blocking call.
      """
      idx = self.counter + 1
      self.counter = idx
  
      logger.debug(f"incrementing {flag_name} idx -> {idx}")
  
      check_in = CheckIn(
          idx=idx, expect_next_in=expect_next_in, flag_name=flag_name
      )
  
      self.last_check_ins[flag_name] = check_in
      # * If there was no such check-in with the same flag name, we will
      #   schedule it to flush immediately.
      # * We want to make sure that more strict `expect_next_in`s are respected
      #   so we need to reschedule the check-in for the next flush.
      #   If we find much sooner expiry date, we will take it immediately.
      # * Notice that `Immediately` does not mean that the call is blocking.
      was_scheduled = self.timed_tasks.get(flag_name)
      if was_scheduled is None:
          # Schedule to flush ASAP
          timed_task = TimedTask(when=0, flag_name=flag_name)
          self._schedule(timed_task)
          logger.debug(
              f"scheduled {timed_task} ASAP because no physical check-in was found"
          )
      else:
          if was_scheduled.when > check_in.expiry_date:
              # Schedule to flush ASAP
              timed_task = TimedTask(when=0, flag_name=flag_name)
              self._schedule(timed_task)
              # We need to remove the old task from the queue but the queue
              # implementation does not allow to update the priority of the
              # existing items, so we need to remove the item and re-insert it
              # with the new priority. This is done by setting `overwritten`
              # flag to `True` and inserting the new item with the same
              # `flag_name`. The writer thread will just ignore such items.
              was_scheduled.overwritten = True
              logger.debug(
                  f"scheduled {timed_task} because it newer is stricter than {was_scheduled}"
              )
  
      return check_in
  
imports: import math
  import time
  import queue
  import threading
  from collections import defaultdict
  from dataclasses import dataclass, field
  from pathlib import Path
  from typing import ClassVar, Dict, Optional, Tuple, Set, Union, TYPE_CHECKING
  from cachetools import LRUCache
  from aim.sdk.reporter.file_manager import FileManager
  import logging
label_desc: Use image node_id 4 for calling the RunStatusReporter obj's underlying member method code with example usage: obj._increment() and returns: check_in