fn read_index<T: Transport, C>(
        &mut self,
        poll_ctx: &mut PollContext<EK, ER, T, C>,
        req: RaftCmdRequest,
        mut err_resp: RaftCmdResponse,
        cb: Callback<EK::Snapshot>,
    ) -> bool {
        if let Err(e) = self.pre_read_index() {
            debug!(
                "prevents unsafe read index";
                "region_id" => self.region_id,
                "peer_id" => self.peer.get_id(),
                "err" => ?e,
            );
            poll_ctx.raft_metrics.propose.unsafe_read_index += 1;
            cmd_resp::bind_error(&mut err_resp, e);
            cb.invoke_with_response(err_resp);
            self.should_wake_up = true;
            return false;
        }

        let renew_lease_time = monotonic_raw_now();
        if self.is_leader() {
            match self.inspect_lease() {
                // Here combine the new read request with the previous one even if the lease expired is
                // ok because in this case, the previous read index must be sent out with a valid
                // lease instead of a suspect lease. So there must no pending transfer-leader proposals
                // before or after the previous read index, and the lease can be renewed when get
                // heartbeat responses.
                LeaseState::Valid | LeaseState::Expired => {
                    let committed_index = self.get_store().committed_index();
                    if let Some(read) = self.pending_reads.back_mut() {
                        let max_lease = poll_ctx.cfg.raft_store_max_leader_lease();
                        if read.renew_lease_time + max_lease > renew_lease_time {
                            read.push_command(req, cb, committed_index);
                            return false;
                        }
                    }
                }
                // If the current lease is suspect, new read requests can't be appended into
                // `pending_reads` because if the leader is transferred, the latest read could
                // be dirty.
                _ => {}
            }
        }

        // When a replica cannot detect any leader, `MsgReadIndex` will be dropped, which would
        // cause a long time waiting for a read response. Then we should return an error directly
        // in this situation.
        if !self.is_leader() && self.leader_id() == INVALID_ID {
            cmd_resp::bind_error(
                &mut err_resp,
                box_err!("{} can not read index due to no leader", self.tag),
            );
            poll_ctx.raft_metrics.invalid_proposal.read_index_no_leader += 1;
            // The leader may be hibernated, send a message for trying to awaken the leader.
            if poll_ctx.cfg.hibernate_regions
                && (self.bcast_wake_up_time.is_none()
                    || self.bcast_wake_up_time.as_ref().unwrap().elapsed()
                        >= Duration::from_millis(MIN_BCAST_WAKE_UP_INTERVAL))
            {
                self.bcast_wake_up_message(poll_ctx);
                self.bcast_wake_up_time = Some(UtilInstant::now_coarse());
            }
            self.should_wake_up = true;
            cb.invoke_with_response(err_resp);
            return false;
        }

        // Should we call pre_propose here?
        let last_pending_read_count = self.raft_group.raft.pending_read_count();
        let last_ready_read_count = self.raft_group.raft.ready_read_count();

        poll_ctx.raft_metrics.propose.read_index += 1;

        self.bcast_wake_up_time = None;
        let id = Uuid::new_v4();
        self.raft_group.read_index(id.as_bytes().to_vec());

        let pending_read_count = self.raft_group.raft.pending_read_count();
        let ready_read_count = self.raft_group.raft.ready_read_count();

        if pending_read_count == last_pending_read_count
            && ready_read_count == last_ready_read_count
            && self.is_leader()
        {
            // The message gets dropped silently, can't be handled anymore.
            apply::notify_stale_req(self.term(), cb);
            return false;
        }

        let read = ReadIndexRequest::with_command(id, req, cb, renew_lease_time);
        self.pending_reads.push_back(read, self.is_leader());
        self.should_wake_up = true;

        debug!(
            "request to get a read index";
            "request_id" => ?id,
            "region_id" => self.region_id,
            "peer_id" => self.peer.get_id(),
            "is_leader" => self.is_leader(),
        );

        // TimeoutNow has been sent out, so we need to propose explicitly to
        // update leader lease.
        if self.leader_lease.inspect(Some(renew_lease_time)) == LeaseState::Suspect {
            let req = RaftCmdRequest::default();
            if let Ok(Either::Left(index)) = self.propose_normal(poll_ctx, req) {
                let p = Proposal {
                    is_conf_change: false,
                    index,
                    term: self.term(),
                    cb: Callback::None,
                    renew_lease_time: Some(renew_lease_time),
                };
                self.post_propose(poll_ctx, p);
            }
        }

        true
    }
// Returns a boolean to indicate whether the `read` is proposed or not.
// For these cases it won't be proposed:
// 1. The region is in merging or splitting;
// 2. The message is stale and dropped by the Raft group internally;
// 3. There is already a read request proposed in the current lease;
Returns a boolean to indicate whether the `read` is proposed or not.
For these cases it won't be proposed:
1. The region is in merging or splitting;
2. The message is stale and dropped by the Raft group internally;
3. There is already a read request proposed in the current lease.
Returns a boolean to indicate whether the ` read ` is proposed or not . For these cases it won ' t be proposed : 1 . The region is in merging or splitting ; 2 . The message is stale and dropped by the Raft group internally ; 3 . There is already a read request proposed in the current lease .
fn read_index<T: Transport, C>(
        &mut self,
        poll_ctx: &mut PollContext<EK, ER, T, C>,
        req: RaftCmdRequest,
        mut err_resp: RaftCmdResponse,
        cb: Callback<EK::Snapshot>,
    ) -> bool {
        if let Err(e) = self.pre_read_index() {
            debug!(
                "prevents unsafe read index";
                "region_id" => self.region_id,
                "peer_id" => self.peer.get_id(),
                "err" => ?e,
            );
            poll_ctx.raft_metrics.propose.unsafe_read_index += 1;
            cmd_resp::bind_error(&mut err_resp, e);
            cb.invoke_with_response(err_resp);
            self.should_wake_up = true;
            return false;
        }
        let renew_lease_time = monotonic_raw_now();
        if self.is_leader() {
            match self.inspect_lease() {
                // Here combine the new read request with the previous one even if the lease expired is
                // ok because in this case, the previous read index must be sent out with a valid
                // lease instead of a suspect lease. So there must no pending transfer-leader proposals
                // before or after the previous read index, and the lease can be renewed when get
                // heartbeat responses.
                LeaseState::Valid | LeaseState::Expired => {
                    let committed_index = self.get_store().committed_index();
                    if let Some(read) = self.pending_reads.back_mut() {
                        let max_lease = poll_ctx.cfg.raft_store_max_leader_lease();
                        if read.renew_lease_time + max_lease > renew_lease_time {
                            read.push_command(req, cb, committed_index);
                            return false;
                        }
                    }
                }
                // If the current lease is suspect, new read requests can't be appended into
                // `pending_reads` because if the leader is transferred, the latest read could
                // be dirty.
                _ => {}
            }
        }
        // When a replica cannot detect any leader, `MsgReadIndex` will be dropped, which would
        // cause a long time waiting for a read response. Then we should return an error directly
        // in this situation.
        if !self.is_leader() && self.leader_id() == INVALID_ID {
            cmd_resp::bind_error(
                &mut err_resp,
                box_err!("{} can not read index due to no leader", self.tag),
            );
            poll_ctx.raft_metrics.invalid_proposal.read_index_no_leader += 1;
            // The leader may be hibernated, send a message for trying to awaken the leader.
            if poll_ctx.cfg.hibernate_regions
                && (self.bcast_wake_up_time.is_none()
                    || self.bcast_wake_up_time.as_ref().unwrap().elapsed()
                        >= Duration::from_millis(MIN_BCAST_WAKE_UP_INTERVAL))
            {
                self.bcast_wake_up_message(poll_ctx);
                self.bcast_wake_up_time = Some(UtilInstant::now_coarse());
            }
            self.should_wake_up = true;
            cb.invoke_with_response(err_resp);
            return false;
        }
        // Should we call pre_propose here?
        let last_pending_read_count = self.raft_group.raft.pending_read_count();
        let last_ready_read_count = self.raft_group.raft.ready_read_count();
        poll_ctx.raft_metrics.propose.read_index += 1;
        self.bcast_wake_up_time = None;
        let id = Uuid::new_v4();
        self.raft_group.read_index(id.as_bytes().to_vec());
        let pending_read_count = self.raft_group.raft.pending_read_count();
        let ready_read_count = self.raft_group.raft.ready_read_count();
        if pending_read_count == last_pending_read_count
            && ready_read_count == last_ready_read_count
            && self.is_leader()
        {
            // The message gets dropped silently, can't be handled anymore.
            apply::notify_stale_req(self.term(), cb);
            return false;
        }
        let read = ReadIndexRequest::with_command(id, req, cb, renew_lease_time);
        self.pending_reads.push_back(read, self.is_leader());
        self.should_wake_up = true;
        debug!(
            "request to get a read index";
            "request_id" => ?id,
            "region_id" => self.region_id,
            "peer_id" => self.peer.get_id(),
            "is_leader" => self.is_leader(),
        );
        // TimeoutNow has been sent out, so we need to propose explicitly to
        // update leader lease.
        if self.leader_lease.inspect(Some(renew_lease_time)) == LeaseState::Suspect {
            let req = RaftCmdRequest::default();
            if let Ok(Either::Left(index)) = self.propose_normal(poll_ctx, req) {
                let p = Proposal {
                    is_conf_change: false,
                    index,
                    term: self.term(),
                    cb: Callback::None,
                    renew_lease_time: Some(renew_lease_time),
                };
                self.post_propose(poll_ctx, p);
            }
        }
        true
    }
fn read_index < T : Transport , C > ( & mut self , poll_ctx : & mut PollContext < EK , ER , T , C > , req : RaftCmdRequest , mut err_resp : RaftCmdResponse , cb : Callback < EK :: Snapshot > , ) -> bool { if let Err ( e ) = self . pre_read_index ( ) { debug ! ( " " " " self region_id " " self peer get_id ( ) " " e ) ; poll_ctx . raft_metrics . propose . unsafe_read_index += 1 ; cmd_resp :: bind_error ( & mut err_resp , e ) ; cb . invoke_with_response ( err_resp ) ; self . should_wake_up = true ; return false ; } let renew_lease_time = monotonic_raw_now ( ) ; if self . is_leader ( ) { match self . inspect_lease ( ) { // Here combine the new read request with the previous one even if the lease expired is // ok because in this case, the previous read index must be sent out with a valid // lease instead of a suspect lease. So there must no pending transfer-leader proposals // before or after the previous read index, and the lease can be renewed when get // heartbeat responses. LeaseState :: Valid | LeaseState :: Expired => { let committed_index = self . get_store ( ) . committed_index ( ) ; if let Some ( read ) = self . pending_reads . back_mut ( ) { let max_lease = poll_ctx . cfg . raft_store_max_leader_lease ( ) ; if read . renew_lease_time + max_lease > renew_lease_time { read . push_command ( req , cb , committed_index ) ; return false ; } } } // If the current lease is suspect, new read requests can't be appended into // `pending_reads` because if the leader is transferred, the latest read could // be dirty. _ => { } } } // When a replica cannot detect any leader, `MsgReadIndex` will be dropped, which would // cause a long time waiting for a read response. Then we should return an error directly // in this situation. if ! self . is_leader ( ) && self . leader_id ( ) == INVALID_ID { cmd_resp :: bind_error ( & mut err_resp , box_err ! ( " " self tag ) , ) ; poll_ctx . raft_metrics . invalid_proposal . read_index_no_leader += 1 ; // The leader may be hibernated, send a message for trying to awaken the leader. if poll_ctx . cfg . hibernate_regions && ( self . bcast_wake_up_time . is_none ( ) || self . bcast_wake_up_time . as_ref ( ) . unwrap ( ) . elapsed ( ) >= Duration :: from_millis ( MIN_BCAST_WAKE_UP_INTERVAL ) ) { self . bcast_wake_up_message ( poll_ctx ) ; self . bcast_wake_up_time = Some ( UtilInstant :: now_coarse ( ) ) ; } self . should_wake_up = true ; cb . invoke_with_response ( err_resp ) ; return false ; } // Should we call pre_propose here? let last_pending_read_count = self . raft_group . raft . pending_read_count ( ) ; let last_ready_read_count = self . raft_group . raft . ready_read_count ( ) ; poll_ctx . raft_metrics . propose . read_index += 1 ; self . bcast_wake_up_time = None ; let id = Uuid :: new_v4 ( ) ; self . raft_group . read_index ( id . as_bytes ( ) . to_vec ( ) ) ; let pending_read_count = self . raft_group . raft . pending_read_count ( ) ; let ready_read_count = self . raft_group . raft . ready_read_count ( ) ; if pending_read_count == last_pending_read_count && ready_read_count == last_ready_read_count && self . is_leader ( ) { // The message gets dropped silently, can't be handled anymore. apply :: notify_stale_req ( self . term ( ) , cb ) ; return false ; } let read = ReadIndexRequest :: with_command ( id , req , cb , renew_lease_time ) ; self . pending_reads . push_back ( read , self . is_leader ( ) ) ; self . should_wake_up = true ; debug ! ( " " " " id " " self region_id " " self peer get_id ( ) " " self is_leader ( ) ) ; // TimeoutNow has been sent out, so we need to propose explicitly to // update leader lease. if self . leader_lease . inspect ( Some ( renew_lease_time ) ) == LeaseState :: Suspect { let req = RaftCmdRequest :: default ( ) ; if let Ok ( Either :: Left ( index ) ) = self . propose_normal ( poll_ctx , req ) { let p = Proposal { is_conf_change : false , index , term : self . term ( ) , cb : Callback :: None , renew_lease_time : Some ( renew_lease_time ) , } ; self . post_propose ( poll_ctx , p ) ; } } true }
Returns a boolean to indicate whether the `read` is proposed or not.
Returns a boolean to indicate whether the ` read ` is proposed or not .
&mut self
poll_ctx
&mut PollContext<EK, ER, T, C>
req
RaftCmdRequest
err_resp
RaftCmdResponse
cb
Callback<EK::Snapshot>