Method sync_handler

rft/trainer/zmq.py:478–579

处理同步信号并更新任务状态

(self)

                        self.ready_queue.put(chunk_data)
    
    def sync_handler(self):
        """处理同步信号并更新任务状态"""
        while True:
            parts = self.sync_signal.recv_multipart()
            if len(parts) != 2:
                logger.error(f"Invalid message parts: {len(parts)}, parts: {parts}")
                continue
            
            topic, d = parts
            latest_gid = pickle.loads(d)
            
            while self.local_gid <= latest_gid:
                logger.debug(f"Sync advantages for group {self.local_gid}")
                self.advantage_syncer.send_pyobj(SyncAdvantagesRequest(self.local_gid))
                task_status = self.advantage_syncer.recv_pyobj()
                task_status.sort(key=lambda x: x.advantage, reverse=True)
                
                next_new_tasks = []
                valid_next_task_completions = []
                scores = []
                
                for status in task_status:
                    scores.append(status.score)
                    if (status.score >= float(os.environ.get("MULTITURN_SAMPLE_THRESHOLD", DEFAULT_THRESHOLD)) and 
                        status.advantage >= task_status[self.mt_max_beam_width % len(task_status)].advantage):
                        # 考虑将下一轮任务添加到任务队列
                        valid_next_task_completions.append(status.completion_id)
                
                # 检查是否可能有进一步的任务
                for status in task_status:
                    if status.completion_id in self.cached_tasks:
                        d = self.cached_tasks[status.completion_id]
                        if (d.data.get("next_id", None) is not None and 
                            d.status.completion_id in valid_next_task_completions):
                            # 考虑将下一轮任务添加到任务队列
                            next_new_tasks.append(d)
                
                if next_new_tasks:
                    # 发送到全局任务分发循环
                    data = []
                    for item in next_new_tasks:
                        data.append({
                            "id": item.data["id"],
                            "gid": self.local_gid,
                            "next_id": item.data["next_id"],
                            "completion_id": item.status.completion_id,
                            "completion": item.data["completion"],
                        })
                    
                    # 计算发送数据的正确长度
                    valid_completions_counts = len(valid_next_task_completions)
                    ratio = len(task_status) // valid_completions_counts
                    data = data * ratio
                    
                    if (rem := len(task_status) % valid_completions_counts) != 0:
                        # 仅对一个节点附加提醒
                        # 选择包含最小字典顺序UUID的节点并添加提醒
                        small_completion_id = sorted(valid_next_task_completions)[0]