Method start

rft/trainer/zmq.py:602–632

启动所有线程并运行主循环

(self)

                            f"create time: {oldest_task.status.created_time}")
    
    def start(self):
        """启动所有线程并运行主循环"""
        # 启动所有线程
        threads = [
            threading.Thread(target=self.reprocess, daemon=True),
            threading.Thread(target=self.provider, daemon=True),
            threading.Thread(target=self.reporter, daemon=True),
            threading.Thread(target=self.serve_stealing, daemon=True),
            threading.Thread(target=self.work_stealing, daemon=True),
            threading.Thread(target=self.sync_handler, daemon=True),
            threading.Thread(target=self.monitor, daemon=True)
        ]
        
        for thread in threads:
            thread.start()
        
        # 主循环
        while True:
            tac: TaskAndContent = self.balance_collect.recv_pyobj()
            self.cached_tasks[tac.status.completion_id] = tac
            self.balance_collect.send_string("Received")
            
            # 发送任务状态到全局
            self.result_sender.send_pyobj(tac.status)
            self.result_sender.recv_string()
            
            if len(self.cached_tasks) >= self.max_cache_size:
                logger.warning(f"Too many cached tasks. [ Local GID: {self.local_gid} | "
                               f"Cached: {len(self.cached_tasks)}, Reprocessing: {self.valid_tasks.qsize()} | "
                               f"Queued: {self.ready_queue.qsize()} ]")
                time.sleep(3)  # 缓存过多