class_docstr: Server that works with ``FileTimerClient``. Clients are expected to be
  running on the same host as the process that is running this server.
  Each host in the job is expected to start its own timer server locally
  and each server instance manages timers for local workers (running on
  processes on the same host).
  
  Args:
  
      file_path: str, the path of a FIFO special file to be created.
  
      max_interval: float, max interval in seconds for each watchdog loop.
  
      daemon: bool, running the watchdog thread in daemon mode or not.
                    A daemon thread will not block a process to stop.
      log_event: Callable[[Dict[str, str]], None], an optional callback for
              logging the events in JSON format.
class_docstr_tok: ["Server","that","works","with","``","FileTimerClient","``",".","Clients","are","expected","to","be","running","on","the","same","host","as","the","process","that","is","running","this","server",".","Each","host","in","the","job","is","expected","to","start","its","own","timer","server","locally","and","each","server","instance","manages","timers","for","local","workers","(","running","on","processes","on","the","same","host",")",".","Args",":","file_path",":","str",",","the","path","of","a","FIFO","special","file","to","be","created",".","max_interval",":","float",",","max","interval","in","seconds","for","each","watchdog","loop",".","daemon",":","bool",",","running","the","watchdog","thread","in","daemon","mode","or","not",".","A","daemon","thread","will","not","block","a","process","to","stop",".","log_event",":","Callable","[","[","Dict","[","str",",","str","]","]",",","None","]",",","an","optional","callback","for","logging","the","events","in","JSON","format","."]
code: def _run_watchdog(self, fd: io.TextIOWrapper) -> None:
      timer_requests = self._get_requests(fd, self._max_interval)
      self.register_timers(timer_requests)
      now = time.time()
      reaped_worker_pids = set()
      for worker_pid, expired_timers in self.get_expired_timers(
          now
      ).items():
          log.info(
              "Reaping worker_pid=[%s]. Expired timers: %s",
              worker_pid,
              self._get_scopes(expired_timers),
          )
          reaped_worker_pids.add(worker_pid)
          # In case we have multiple expired timers, we find the first timer
          # with a valid signal (>0) in the expiration time order.
          expired_timers.sort(key=lambda timer: timer.expiration_time)
          signal = 0
          expired_timer = None
          for timer in expired_timers:
              self._log_event("timer expired", timer)
              if timer.signal > 0:
                  signal = timer.signal
                  expired_timer = timer
                  break
          if signal <= 0:
              log.info(
                  "No signal specified with worker=[%s]. Do not reap it.",
                  worker_pid,
              )
              continue
          if self._reap_worker(worker_pid, signal):
              log.info(
                  "Successfully reaped worker=[%s] with signal=%s",
                  worker_pid,
                  signal,
              )
              self._log_event("kill worker process", expired_timer)
          else:
              log.error(
                  "Error reaping worker=[%s]. Will retry on next watchdog.",
                  worker_pid,
              )
      self.clear_timers(reaped_worker_pids)
  
imports: import io
  import json
  import logging
  import os
  import select
  import signal
  import sys
  import threading
  import time
  from typing import Callable, Dict, List, Optional, Set, Tuple
  from torch.distributed.elastic.timer.api import TimerClient, TimerRequest
label_desc: Use image node_id 6 for calling the FileTimerServer obj's underlying member method code with example usage: obj._run_watchdog(fd) without return types