Method _sync_node_queue

rft/trainer/zmq.py:118–127

节点队列同步线程

(self)

            logger.info(f"[ Global GID: {self.current_gid} | SyncPool Size: {len(self.sync_pool)} | {self.total_ack} acked / {self.recv_count} total ] Current {self.send_count} sent, {self.ack_advantages} ack. Speed {(self.recv_count-last_count)/interval:.2f}/s.")
    
    def _sync_node_queue(self):
        """节点队列同步线程"""
        while True:
            time.sleep(3)
            with self.sync_lock:
                self.sync_sender.send_multipart([
                    b"SYNC_NODE_QUEUE_LENGTHS",
                    pickle.dumps(self.node_queue_lengths)
                ])
            logger.debug(f"Sync node queue lengths {len(self.node_queue_lengths)}")
    
    def start(self):
        """启动同步管理器"""