Method monitor

rft/trainer/zmq.py:581–600

检查缓存中是否有超时的任务

(self)

                self.local_gid += 1
    
    def monitor(self):
        """检查缓存中是否有超时的任务"""
        interval = int(os.environ.get("LOCAL_MONITOR_INTERVAL", "300"))
        while True:
            time.sleep(interval)
            oldest_task = None
            
            for k, v in list(self.cached_tasks.items()):
                if oldest_task is None or v.status.created_time < oldest_task.status.created_time:
                    oldest_task = v
                
                if (datetime.datetime.now() - v.status.created_time).total_seconds() > self.timeout:
                    logger.warning(f"Task {v.status.task_id}, completion {k} is out of time.")
                    # del self.cached_tasks[k]
            
            if oldest_task is not None:
                logger.info(f"[ Local GID: {self.local_gid} | Cached: {len(self.cached_tasks)}, "
                            f"Reprocessing: {self.valid_tasks.qsize()} | Queued: {self.ready_queue.qsize()} ] "
                            f"The oldest task id {oldest_task.status.task_id}, "
                            f"create time: {oldest_task.status.created_time}")
    
    def start(self):
        """启动所有线程并运行主循环"""