Qdrant 集群下的共识服务(下)

qdrant集群 同步来自集群的各种元信息 包括集群的状态，心跳，raft的日志信息。

后台的共识任务

    // 在集群实例启动后加载后台的共识任务
    thread::Builder::new()
    .name("consensus".to_string())
    .spawn(move || {
        // 共识任务的入口
        if let Err(err) = consensus.start() {
        } else {
        }
    })?;

    // 循环处理集群各种不同类型的信息，收到信息落地
    pub fn start(&mut self) -> anyhow::Result<()> {
        let tick_period = Duration::from_millis(self.config.tick_period_ms);
        let mut previous_tick = Instant::now();
        let mut idle_cycles = 0_usize;

        loop {
            //  处理来自Raft的消息
            let raft_messages = self.advance_node(tick_period)?;

            // 计算下一次循环的时间
            let elapsed_ticks = previous_tick.elapsed().div_duration_f32(tick_period) as u32;

            previous_tick += tick_period * elapsed_ticks;

            // 对时间间隔调整
            let report_ticks = if raft_messages > 0 {
                let max_elapsed_ticks =
                    cmp::max(1, self.raft_config.election_tick.saturating_sub(5));

                cmp::min(elapsed_ticks, max_elapsed_ticks as u32)
            } else {
                elapsed_ticks
            };

            for _ in 0..report_ticks {
                self.node.tick();
            }
          
        }
    }
    
    fn advance_node(&mut self, tick_period: Duration) -> anyhow::Result<usize> {

        const RAFT_BATCH_SIZE: usize = 128;
        let hard_timeout_at = Instant::now() + tick_period;
        let consecutive_message_timeout = tick_period / 10;

        let mut timeout_at = hard_timeout_at;

        let mut events = 0;
        let mut raft_messages = 0;

        loop {
            // 从leader 收到的消息
            let Ok(message) = self.recv_update(timeout_at) else {
                break;
            };
            // 处理各种不同的消息类型
            let is_raft_message = matches!(message, Message::FromPeer(_));

            if let Err(err) = self.advance_node_impl(message) {
                log::warn!("{err}");
                continue;
            }

            timeout_at = cmp::min(
                hard_timeout_at,
                Instant::now() + consecutive_message_timeout,
            );

            events += 1;
            raft_messages += usize::from(is_raft_message);

            if events >= RAFT_BATCH_SIZE || is_conf_change {
                break;
            }
        }

        Ok(raft_messages)
    }


    fn advance_node_impl(&mut self, message: Message) -> anyhow::Result<()> {
        match message {
            Message::FromClient(operation) => {
                let data =
                    serde_cbor::to_vec(&operation).context("failed to serialize operation")?;
                // 将消息落地写到日志来保证数据一致性
                self.node
                    .propose(vec![], data)
                    .context("failed to propose entry")?;
            }
        }

        Ok(())
    }