fn apply_snap(&mut self, region_id: u64, abort: Arc<AtomicUsize>) -> Result<()> {
        info!("begin apply snap data"; "region_id" => region_id);
        fail_point!("region_apply_snap", |_| { Ok(()) });
        check_abort(&abort)?;
        let region_key = keys::region_state_key(region_id);
        let mut region_state: RegionLocalState =
            match box_try!(self.engines.kv.get_msg_cf(CF_RAFT, &region_key)) {
                Some(state) => state,
                None => {
                    return Err(box_err!(
                        "failed to get region_state from {}",
                        hex::encode_upper(&region_key)
                    ));
                }
            };

        // clear up origin data.
        let region = region_state.get_region().clone();
        let start_key = keys::enc_start_key(&region);
        let end_key = keys::enc_end_key(&region);
        check_abort(&abort)?;
        self.cleanup_overlap_ranges(&start_key, &end_key);
        box_try!(self
            .engines
            .kv
            .delete_all_in_range(&start_key, &end_key, self.use_delete_range));
        check_abort(&abort)?;
        fail_point!("apply_snap_cleanup_range");

        let state_key = keys::apply_state_key(region_id);
        let apply_state: RaftApplyState =
            match box_try!(self.engines.kv.get_msg_cf(CF_RAFT, &state_key)) {
                Some(state) => state,
                None => {
                    return Err(box_err!(
                        "failed to get raftstate from {}",
                        hex::encode_upper(&state_key)
                    ));
                }
            };
        let term = apply_state.get_truncated_state().get_term();
        let idx = apply_state.get_truncated_state().get_index();
        let snap_key = SnapKey::new(region_id, term, idx);
        self.mgr.register(snap_key.clone(), SnapEntry::Applying);
        defer!({
            self.mgr.deregister(&snap_key, &SnapEntry::Applying);
        });
        let mut s = box_try!(self.mgr.get_snapshot_for_applying_to_engine(&snap_key));
        if !s.exists() {
            return Err(box_err!("missing snapshot file {}", s.path()));
        }
        check_abort(&abort)?;
        let timer = Instant::now();
        let options = ApplyOptions {
            db: self.engines.kv.clone(),
            region,
            abort: Arc::clone(&abort),
            write_batch_size: self.batch_size,
            coprocessor_host: self.coprocessor_host.clone(),
        };
        s.apply(options)?;

        let mut wb = self.engines.kv.write_batch();
        region_state.set_state(PeerState::Normal);
        box_try!(wb.put_msg_cf(CF_RAFT, &region_key, &region_state));
        box_try!(wb.delete_cf(CF_RAFT, &keys::snapshot_raft_state_key(region_id)));
        self.engines.kv.write(&wb).unwrap_or_else(|e| {
            panic!("{} failed to save apply_snap result: {:?}", region_id, e);
        });
        info!(
            "apply new data";
            "region_id" => region_id,
            "time_takes" => ?timer.elapsed(),
        );
        Ok(())
    }
/// Applies snapshot data of the Region.
Applies snapshot data of the Region.
Applies snapshot data of the Region .
fn apply_snap(&mut self, region_id: u64, abort: Arc<AtomicUsize>) -> Result<()> {
        info!("begin apply snap data"; "region_id" => region_id);
        fail_point!("region_apply_snap", |_| { Ok(()) });
        check_abort(&abort)?;
        let region_key = keys::region_state_key(region_id);
        let mut region_state: RegionLocalState =
            match box_try!(self.engines.kv.get_msg_cf(CF_RAFT, &region_key)) {
                Some(state) => state,
                None => {
                    return Err(box_err!(
                        "failed to get region_state from {}",
                        hex::encode_upper(&region_key)
                    ));
                }
            };
        // clear up origin data.
        let region = region_state.get_region().clone();
        let start_key = keys::enc_start_key(&region);
        let end_key = keys::enc_end_key(&region);
        check_abort(&abort)?;
        self.cleanup_overlap_ranges(&start_key, &end_key);
        box_try!(self
            .engines
            .kv
            .delete_all_in_range(&start_key, &end_key, self.use_delete_range));
        check_abort(&abort)?;
        fail_point!("apply_snap_cleanup_range");
        let state_key = keys::apply_state_key(region_id);
        let apply_state: RaftApplyState =
            match box_try!(self.engines.kv.get_msg_cf(CF_RAFT, &state_key)) {
                Some(state) => state,
                None => {
                    return Err(box_err!(
                        "failed to get raftstate from {}",
                        hex::encode_upper(&state_key)
                    ));
                }
            };
        let term = apply_state.get_truncated_state().get_term();
        let idx = apply_state.get_truncated_state().get_index();
        let snap_key = SnapKey::new(region_id, term, idx);
        self.mgr.register(snap_key.clone(), SnapEntry::Applying);
        defer!({
            self.mgr.deregister(&snap_key, &SnapEntry::Applying);
        });
        let mut s = box_try!(self.mgr.get_snapshot_for_applying_to_engine(&snap_key));
        if !s.exists() {
            return Err(box_err!("missing snapshot file {}", s.path()));
        }
        check_abort(&abort)?;
        let timer = Instant::now();
        let options = ApplyOptions {
            db: self.engines.kv.clone(),
            region,
            abort: Arc::clone(&abort),
            write_batch_size: self.batch_size,
            coprocessor_host: self.coprocessor_host.clone(),
        };
        s.apply(options)?;
        let mut wb = self.engines.kv.write_batch();
        region_state.set_state(PeerState::Normal);
        box_try!(wb.put_msg_cf(CF_RAFT, &region_key, &region_state));
        box_try!(wb.delete_cf(CF_RAFT, &keys::snapshot_raft_state_key(region_id)));
        self.engines.kv.write(&wb).unwrap_or_else(|e| {
            panic!("{} failed to save apply_snap result: {:?}", region_id, e);
        });
        info!(
            "apply new data";
            "region_id" => region_id,
            "time_takes" => ?timer.elapsed(),
        );
        Ok(())
    }
fn apply_snap ( & mut self , region_id : u64 , abort : Arc < AtomicUsize > ) -> Result < ( ) > { info ! ( " " " " region_id ) ; fail_point ! ( " " { Ok ( ( ) ) } ) ; check_abort ( & abort ) ? ; let region_key = keys :: region_state_key ( region_id ) ; let mut region_state : RegionLocalState = match box_try ! ( self engines kv get_msg_cf ( CF_RAFT region_key ) ) { Some ( state ) => state , None => { return Err ( box_err ! ( " " hex encode_upper ( region_key ) ) ) ; } } ; // clear up origin data. let region = region_state . get_region ( ) . clone ( ) ; let start_key = keys :: enc_start_key ( & region ) ; let end_key = keys :: enc_end_key ( & region ) ; check_abort ( & abort ) ? ; self . cleanup_overlap_ranges ( & start_key , & end_key ) ; box_try ! ( self engines kv delete_all_in_range ( start_key end_key self use_delete_range ) ) ; check_abort ( & abort ) ? ; fail_point ! ( " " ) ; let state_key = keys :: apply_state_key ( region_id ) ; let apply_state : RaftApplyState = match box_try ! ( self engines kv get_msg_cf ( CF_RAFT state_key ) ) { Some ( state ) => state , None => { return Err ( box_err ! ( " " hex encode_upper ( state_key ) ) ) ; } } ; let term = apply_state . get_truncated_state ( ) . get_term ( ) ; let idx = apply_state . get_truncated_state ( ) . get_index ( ) ; let snap_key = SnapKey :: new ( region_id , term , idx ) ; self . mgr . register ( snap_key . clone ( ) , SnapEntry :: Applying ) ; defer ! ( { self mgr deregister ( snap_key SnapEntry Applying ) } ) ; let mut s = box_try ! ( self mgr get_snapshot_for_applying_to_engine ( snap_key ) ) ; if ! s . exists ( ) { return Err ( box_err ! ( " " s path ( ) ) ) ; } check_abort ( & abort ) ? ; let timer = Instant :: now ( ) ; let options = ApplyOptions { db : self . engines . kv . clone ( ) , region , abort : Arc :: clone ( & abort ) , write_batch_size : self . batch_size , coprocessor_host : self . coprocessor_host . clone ( ) , } ; s . apply ( options ) ? ; let mut wb = self . engines . kv . write_batch ( ) ; region_state . set_state ( PeerState :: Normal ) ; box_try ! ( wb put_msg_cf ( CF_RAFT region_key region_state ) ) ; box_try ! ( wb delete_cf ( CF_RAFT keys snapshot_raft_state_key ( region_id ) ) ) ; self . engines . kv . write ( & wb ) . unwrap_or_else ( | e | { panic ! ( " " region_id e ) ; } ) ; info ! ( " " " " region_id " " timer elapsed ( ) ) ; Ok ( ( ) ) }
Applies snapshot data of the Region.
Applies snapshot data of the Region .
&mut self
region_id
u64
abort
Arc<AtomicUsize>