fn init(&mut self) -> Result<Vec<SenderFsmPair<EK, ER>>> {
        // Scan region meta to get saved regions.
        let start_key = keys::REGION_META_MIN_KEY;
        let end_key = keys::REGION_META_MAX_KEY;
        let kv_engine = self.engines.kv.clone();
        let store_id = self.store.get_id();
        let mut total_count = 0;
        let mut tombstone_count = 0;
        let mut applying_count = 0;
        let mut region_peers = vec![];

        let t = Instant::now();
        let mut kv_wb = self.engines.kv.write_batch();
        let mut raft_wb = self.engines.raft.log_batch(4 * 1024);
        let mut applying_regions = vec![];
        let mut merging_count = 0;
        let mut meta = self.store_meta.lock().unwrap();
        let mut replication_state = self.global_replication_state.lock().unwrap();
        kv_engine.scan_cf(CF_RAFT, start_key, end_key, false, |key, value| {
            let (region_id, suffix) = box_try!(keys::decode_region_meta_key(key));
            if suffix != keys::REGION_STATE_SUFFIX {
                return Ok(true);
            }

            total_count += 1;

            let mut local_state = RegionLocalState::default();
            local_state.merge_from_bytes(value)?;

            let region = local_state.get_region();
            if local_state.get_state() == PeerState::Tombstone {
                tombstone_count += 1;
                debug!("region is tombstone"; "region" => ?region, "store_id" => store_id);
                self.clear_stale_meta(&mut kv_wb, &mut raft_wb, &local_state);
                return Ok(true);
            }
            if local_state.get_state() == PeerState::Applying {
                // in case of restart happen when we just write region state to Applying,
                // but not write raft_local_state to raft rocksdb in time.
                box_try!(peer_storage::recover_from_applying_state(
                    &self.engines,
                    &mut raft_wb,
                    region_id
                ));
                applying_count += 1;
                applying_regions.push(region.clone());
                return Ok(true);
            }

            let (tx, mut peer) = box_try!(PeerFsm::create(
                store_id,
                &self.cfg.value(),
                self.region_scheduler.clone(),
                self.engines.clone(),
                region,
            ));
            peer.peer.init_replication_mode(&mut *replication_state);
            if local_state.get_state() == PeerState::Merging {
                info!("region is merging"; "region" => ?region, "store_id" => store_id);
                merging_count += 1;
                peer.set_pending_merge_state(local_state.get_merge_state().to_owned());
            }
            meta.region_ranges.insert(enc_end_key(region), region_id);
            meta.regions.insert(region_id, region.clone());
            // No need to check duplicated here, because we use region id as the key
            // in DB.
            region_peers.push((tx, peer));
            self.coprocessor_host.on_region_changed(
                region,
                RegionChangeEvent::Create,
                StateRole::Follower,
            );
            Ok(true)
        })?;

        if !kv_wb.is_empty() {
            self.engines.kv.write(&kv_wb).unwrap();
            self.engines.kv.sync_wal().unwrap();
        }
        if !raft_wb.is_empty() {
            self.engines.raft.consume(&mut raft_wb, true).unwrap();
        }

        // schedule applying snapshot after raft writebatch were written.
        for region in applying_regions {
            info!("region is applying snapshot"; "region" => ?region, "store_id" => store_id);
            let (tx, mut peer) = PeerFsm::create(
                store_id,
                &self.cfg.value(),
                self.region_scheduler.clone(),
                self.engines.clone(),
                &region,
            )?;
            peer.peer.init_replication_mode(&mut *replication_state);
            peer.schedule_applying_snapshot();
            meta.region_ranges
                .insert(enc_end_key(&region), region.get_id());
            meta.regions.insert(region.get_id(), region);
            region_peers.push((tx, peer));
        }

        info!(
            "start store";
            "store_id" => store_id,
            "region_count" => total_count,
            "tombstone_count" => tombstone_count,
            "applying_count" =>  applying_count,
            "merge_count" => merging_count,
            "takes" => ?t.elapsed(),
        );

        self.clear_stale_data(&meta)?;

        Ok(region_peers)
    }
/// Initialize this store. It scans the db engine, loads all regions
/// and their peers from it, and schedules snapshot worker if necessary.
/// WARN: This store should not be used before initialized.
Initialize this store. It scans the db engine, loads all regions
and their peers from it, and schedules snapshot worker if necessary.
WARN: This store should not be used before initialized.
Initialize this store . It scans the db engine loads all regions and their peers from it and schedules snapshot worker if necessary . WARN : This store should not be used before initialized .
fn init(&mut self) -> Result<Vec<SenderFsmPair<EK, ER>>> {
        // Scan region meta to get saved regions.
        let start_key = keys::REGION_META_MIN_KEY;
        let end_key = keys::REGION_META_MAX_KEY;
        let kv_engine = self.engines.kv.clone();
        let store_id = self.store.get_id();
        let mut total_count = 0;
        let mut tombstone_count = 0;
        let mut applying_count = 0;
        let mut region_peers = vec![];
        let t = Instant::now();
        let mut kv_wb = self.engines.kv.write_batch();
        let mut raft_wb = self.engines.raft.log_batch(4 * 1024);
        let mut applying_regions = vec![];
        let mut merging_count = 0;
        let mut meta = self.store_meta.lock().unwrap();
        let mut replication_state = self.global_replication_state.lock().unwrap();
        kv_engine.scan_cf(CF_RAFT, start_key, end_key, false, |key, value| {
            let (region_id, suffix) = box_try!(keys::decode_region_meta_key(key));
            if suffix != keys::REGION_STATE_SUFFIX {
                return Ok(true);
            }
            total_count += 1;
            let mut local_state = RegionLocalState::default();
            local_state.merge_from_bytes(value)?;
            let region = local_state.get_region();
            if local_state.get_state() == PeerState::Tombstone {
                tombstone_count += 1;
                debug!("region is tombstone"; "region" => ?region, "store_id" => store_id);
                self.clear_stale_meta(&mut kv_wb, &mut raft_wb, &local_state);
                return Ok(true);
            }
            if local_state.get_state() == PeerState::Applying {
                // in case of restart happen when we just write region state to Applying,
                // but not write raft_local_state to raft rocksdb in time.
                box_try!(peer_storage::recover_from_applying_state(
                    &self.engines,
                    &mut raft_wb,
                    region_id
                ));
                applying_count += 1;
                applying_regions.push(region.clone());
                return Ok(true);
            }
            let (tx, mut peer) = box_try!(PeerFsm::create(
                store_id,
                &self.cfg.value(),
                self.region_scheduler.clone(),
                self.engines.clone(),
                region,
            ));
            peer.peer.init_replication_mode(&mut *replication_state);
            if local_state.get_state() == PeerState::Merging {
                info!("region is merging"; "region" => ?region, "store_id" => store_id);
                merging_count += 1;
                peer.set_pending_merge_state(local_state.get_merge_state().to_owned());
            }
            meta.region_ranges.insert(enc_end_key(region), region_id);
            meta.regions.insert(region_id, region.clone());
            // No need to check duplicated here, because we use region id as the key
            // in DB.
            region_peers.push((tx, peer));
            self.coprocessor_host.on_region_changed(
                region,
                RegionChangeEvent::Create,
                StateRole::Follower,
            );
            Ok(true)
        })?;
        if !kv_wb.is_empty() {
            self.engines.kv.write(&kv_wb).unwrap();
            self.engines.kv.sync_wal().unwrap();
        }
        if !raft_wb.is_empty() {
            self.engines.raft.consume(&mut raft_wb, true).unwrap();
        }
        // schedule applying snapshot after raft writebatch were written.
        for region in applying_regions {
            info!("region is applying snapshot"; "region" => ?region, "store_id" => store_id);
            let (tx, mut peer) = PeerFsm::create(
                store_id,
                &self.cfg.value(),
                self.region_scheduler.clone(),
                self.engines.clone(),
                &region,
            )?;
            peer.peer.init_replication_mode(&mut *replication_state);
            peer.schedule_applying_snapshot();
            meta.region_ranges
                .insert(enc_end_key(&region), region.get_id());
            meta.regions.insert(region.get_id(), region);
            region_peers.push((tx, peer));
        }
        info!(
            "start store";
            "store_id" => store_id,
            "region_count" => total_count,
            "tombstone_count" => tombstone_count,
            "applying_count" =>  applying_count,
            "merge_count" => merging_count,
            "takes" => ?t.elapsed(),
        );
        self.clear_stale_data(&meta)?;
        Ok(region_peers)
    }
fn init ( & mut self ) -> Result < Vec < SenderFsmPair < EK , ER > > > { // Scan region meta to get saved regions. let start_key = keys :: REGION_META_MIN_KEY ; let end_key = keys :: REGION_META_MAX_KEY ; let kv_engine = self . engines . kv . clone ( ) ; let store_id = self . store . get_id ( ) ; let mut total_count = 0 ; let mut tombstone_count = 0 ; let mut applying_count = 0 ; let mut region_peers = vec ! [ ] ; let t = Instant :: now ( ) ; let mut kv_wb = self . engines . kv . write_batch ( ) ; let mut raft_wb = self . engines . raft . log_batch ( 4 * 1024 ) ; let mut applying_regions = vec ! [ ] ; let mut merging_count = 0 ; let mut meta = self . store_meta . lock ( ) . unwrap ( ) ; let mut replication_state = self . global_replication_state . lock ( ) . unwrap ( ) ; kv_engine . scan_cf ( CF_RAFT , start_key , end_key , false , | key , value | { let ( region_id , suffix ) = box_try ! ( keys decode_region_meta_key ( key ) ) ; if suffix != keys :: REGION_STATE_SUFFIX { return Ok ( true ) ; } total_count += 1 ; let mut local_state = RegionLocalState :: default ( ) ; local_state . merge_from_bytes ( value ) ? ; let region = local_state . get_region ( ) ; if local_state . get_state ( ) == PeerState :: Tombstone { tombstone_count += 1 ; debug ! ( " " " " region " " store_id ) ; self . clear_stale_meta ( & mut kv_wb , & mut raft_wb , & local_state ) ; return Ok ( true ) ; } if local_state . get_state ( ) == PeerState :: Applying { // in case of restart happen when we just write region state to Applying, // but not write raft_local_state to raft rocksdb in time. box_try ! ( peer_storage recover_from_applying_state ( self engines mut raft_wb region_id ) ) ; applying_count += 1 ; applying_regions . push ( region . clone ( ) ) ; return Ok ( true ) ; } let ( tx , mut peer ) = box_try ! ( PeerFsm create ( store_id self cfg value ( ) self region_scheduler clone ( ) self engines clone ( ) region ) ) ; peer . peer . init_replication_mode ( & mut * replication_state ) ; if local_state . get_state ( ) == PeerState :: Merging { info ! ( " " " " region " " store_id ) ; merging_count += 1 ; peer . set_pending_merge_state ( local_state . get_merge_state ( ) . to_owned ( ) ) ; } meta . region_ranges . insert ( enc_end_key ( region ) , region_id ) ; meta . regions . insert ( region_id , region . clone ( ) ) ; // No need to check duplicated here, because we use region id as the key // in DB. region_peers . push ( ( tx , peer ) ) ; self . coprocessor_host . on_region_changed ( region , RegionChangeEvent :: Create , StateRole :: Follower , ) ; Ok ( true ) } ) ? ; if ! kv_wb . is_empty ( ) { self . engines . kv . write ( & kv_wb ) . unwrap ( ) ; self . engines . kv . sync_wal ( ) . unwrap ( ) ; } if ! raft_wb . is_empty ( ) { self . engines . raft . consume ( & mut raft_wb , true ) . unwrap ( ) ; } // schedule applying snapshot after raft writebatch were written. for region in applying_regions { info ! ( " " " " region " " store_id ) ; let ( tx , mut peer ) = PeerFsm :: create ( store_id , & self . cfg . value ( ) , self . region_scheduler . clone ( ) , self . engines . clone ( ) , & region , ) ? ; peer . peer . init_replication_mode ( & mut * replication_state ) ; peer . schedule_applying_snapshot ( ) ; meta . region_ranges . insert ( enc_end_key ( & region ) , region . get_id ( ) ) ; meta . regions . insert ( region . get_id ( ) , region ) ; region_peers . push ( ( tx , peer ) ) ; } info ! ( " " " " store_id " " total_count " " tombstone_count " " applying_count " " merging_count " " t elapsed ( ) ) ; self . clear_stale_data ( & meta ) ? ; Ok ( region_peers ) }
Initialize this store.
Initialize this store .
&mut self