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

qdrant 采用raft 协议

1.共识服务初始化

    // mpsc 是一个 多生产者单消费者的通道
    // propose_sender 会将操作的状态变化 发送到propose_receiver 
    // propose_receiver 会将变化发送给其他的节点
    let (propose_sender, propose_receiver) = std::sync::mpsc::channel();
    // 如果开启了集群 会开启共识服务
    let propose_operation_sender = if settings.cluster.enabled {
        Some(OperationSender::new(propose_sender))
    } else {
        None
    };

    // 将发送服务 关联到集合的操作服务上
    // 这里设置钩子函数，对操作拦截操作后将变化发送到其他的集群节点
    let toc = TableOfContent::new(
        &settings.storage,
        search_runtime,
        update_runtime,
        general_runtime,
        optimizer_resource_budget,
        channel_service.clone(),
        persistent_consensus_state.this_peer_id(),
        propose_operation_sender.clone(),
    );
    // 设置钩子函数
        pub fn new(
        storage_config: &StorageConfig,
        search_runtime: Runtime,
        update_runtime: Runtime,
        general_runtime: Runtime,
        optimizer_resource_budget: ResourceBudget,
        channel_service: ChannelService,
        this_peer_id: PeerId,
        consensus_proposal_sender: Option<OperationSender>,
    ) -> Self {
    
            let collection = general_runtime.block_on(Collection::load(
                collection_name.clone(),
                this_peer_id,
                &collection_path,
                &collection_snapshots_path,
                storage_config
                    .to_shared_storage_config(is_distributed)
                    .into(),
                channel_service.clone(),
                // 钩子
                Self::change_peer_from_state_callback(
                    consensus_proposal_sender.clone(),
                    collection_name.clone(),
                    ReplicaState::Dead,
                ),
                 // 钩子
                Self::request_shard_transfer_callback(
                    consensus_proposal_sender.clone(),
                    collection_name.clone(),
                ),
                 // 钩子
                Self::abort_shard_transfer_callback(
                    consensus_proposal_sender.clone(),
                    collection_name.clone(),
                ),

            ));

    }


    fn request_shard_transfer_callback(
        proposal_sender: Option<OperationSender>,
        collection_name: String,
    ) -> RequestShardTransfer {
        Arc::new(move |shard_transfer| {
            if let Some(proposal_sender) = &proposal_sender {
                let collection_name = collection_name.clone();
                let to_peer = shard_transfer.to;
                let operation =
                    ConsensusOperations::start_transfer(collection_name.clone(), shard_transfer);
                // 生产者 产生消息。 propose_sender会消费对应的消息 
                if let Err(send_error) = proposal_sender.send(operation) {

                }
            } else {

            }
        })
    }
   

2.共识服务生成提议

将消息发送到集群的节点，生成提议。

// 共识服务
let handle = Consensus::run(
            &slog_logger,
            consensus_state.clone(),
            bootstrap,
            args.uri.map(|uri| uri.to_string()),
            settings.clone(),
            channel_service,
            propose_receiver,
            tonic_telemetry_collector,
            toc_arc.clone(),
            runtime_handle.clone(),
            args.reinit,
        )
        .expect("Can't initialize consensus");


        // message_sender 发送提议到其他节点
        let (mut consensus, message_sender) = Self::new(
            logger,
            state_ref.clone(),
            bootstrap_peer,
            uri,
            p2p_port,
            config,
            tls_client_config,
            channel_service,
            runtime.clone(),
            reinit,
        )?;
        let message_sender_moved = message_sender.clone();
        thread::Builder::new()
            .name("forward-proposals".to_string())
            .spawn(move || {
                // 收到本地操作数据，发送到message_sender用来发送提议
                while let Ok(entry) = propose_receiver.recv() {
                    if message_sender_moved
                        .blocking_send(Message::FromClient(entry))
                        .is_err()
                    {
                        log::error!("Can not forward new entry to consensus as it was stopped.");
                        break;
                    }
                }
            })?;