fn send_snap(
    env: Arc<Environment>,
    mgr: SnapManager,
    security_mgr: Arc<SecurityManager>,
    cfg: &Config,
    addr: &str,
    msg: RaftMessage,
) -> Result<impl Future<Item = SendStat, Error = Error>> {
    assert!(msg.get_message().has_snapshot());
    let timer = Instant::now();

    let send_timer = SEND_SNAP_HISTOGRAM.start_coarse_timer();

    let key = {
        let snap = msg.get_message().get_snapshot();
        SnapKey::from_snap(snap)?
    };

    mgr.register(key.clone(), SnapEntry::Sending);
    let deregister = {
        let (mgr, key) = (mgr.clone(), key.clone());
        DeferContext::new(move || mgr.deregister(&key, &SnapEntry::Sending))
    };

    let s = box_try!(mgr.get_snapshot_for_sending(&key));
    if !s.exists() {
        return Err(box_err!("missing snap file: {:?}", s.path()));
    }
    let total_size = s.total_size()?;

    let chunks = {
        let mut first_chunk = SnapshotChunk::default();
        first_chunk.set_message(msg);

        SnapChunk {
            first: Some(first_chunk),
            snap: s,
            remain_bytes: total_size as usize,
        }
    };

    let cb = ChannelBuilder::new(env)
        .stream_initial_window_size(cfg.grpc_stream_initial_window_size.0 as i32)
        .keepalive_time(cfg.grpc_keepalive_time.0)
        .keepalive_timeout(cfg.grpc_keepalive_timeout.0)
        .default_compression_algorithm(cfg.grpc_compression_algorithm());

    let channel = security_mgr.connect(cb, addr);
    let client = TikvClient::new(channel);
    let (sink, receiver) = client.snapshot()?;

    let send = chunks.forward(sink).map_err(Error::from);
    let send = send
        .and_then(|(s, _)| receiver.map_err(Error::from).map(|_| s))
        .then(move |result| {
            send_timer.observe_duration();
            drop(deregister);
            drop(client);
            result.map(|s| {
                fail_point!("snapshot_delete_after_send");
                s.snap.delete();
                // TODO: improve it after rustc resolves the bug.
                // Call `info` in the closure directly will cause rustc
                // panic with `Cannot create local mono-item for DefId`.
                SendStat {
                    key,
                    total_size,
                    elapsed: timer.elapsed(),
                }
            })
        });
    Ok(send)
}
/// Send the snapshot to specified address.
///
/// It will first send the normal raft snapshot message and then send the snapshot file.
Send the snapshot to specified address.
It will first send the normal raft snapshot message and then send the snapshot file.
Send the snapshot to specified address . It will first send the normal raft snapshot message and then send the snapshot file .
fn send_snap(
    env: Arc<Environment>,
    mgr: SnapManager,
    security_mgr: Arc<SecurityManager>,
    cfg: &Config,
    addr: &str,
    msg: RaftMessage,
) -> Result<impl Future<Item = SendStat, Error = Error>> {
    assert!(msg.get_message().has_snapshot());
    let timer = Instant::now();
    let send_timer = SEND_SNAP_HISTOGRAM.start_coarse_timer();
    let key = {
        let snap = msg.get_message().get_snapshot();
        SnapKey::from_snap(snap)?
    };
    mgr.register(key.clone(), SnapEntry::Sending);
    let deregister = {
        let (mgr, key) = (mgr.clone(), key.clone());
        DeferContext::new(move || mgr.deregister(&key, &SnapEntry::Sending))
    };
    let s = box_try!(mgr.get_snapshot_for_sending(&key));
    if !s.exists() {
        return Err(box_err!("missing snap file: {:?}", s.path()));
    }
    let total_size = s.total_size()?;
    let chunks = {
        let mut first_chunk = SnapshotChunk::default();
        first_chunk.set_message(msg);
        SnapChunk {
            first: Some(first_chunk),
            snap: s,
            remain_bytes: total_size as usize,
        }
    };
    let cb = ChannelBuilder::new(env)
        .stream_initial_window_size(cfg.grpc_stream_initial_window_size.0 as i32)
        .keepalive_time(cfg.grpc_keepalive_time.0)
        .keepalive_timeout(cfg.grpc_keepalive_timeout.0)
        .default_compression_algorithm(cfg.grpc_compression_algorithm());
    let channel = security_mgr.connect(cb, addr);
    let client = TikvClient::new(channel);
    let (sink, receiver) = client.snapshot()?;
    let send = chunks.forward(sink).map_err(Error::from);
    let send = send
        .and_then(|(s, _)| receiver.map_err(Error::from).map(|_| s))
        .then(move |result| {
            send_timer.observe_duration();
            drop(deregister);
            drop(client);
            result.map(|s| {
                fail_point!("snapshot_delete_after_send");
                s.snap.delete();
                // TODO: improve it after rustc resolves the bug.
                // Call `info` in the closure directly will cause rustc
                // panic with `Cannot create local mono-item for DefId`.
                SendStat {
                    key,
                    total_size,
                    elapsed: timer.elapsed(),
                }
            })
        });
    Ok(send)
}
fn send_snap ( env : Arc < Environment > , mgr : SnapManager , security_mgr : Arc < SecurityManager > , cfg : & Config , addr : & str , msg : RaftMessage , ) -> Result < impl Future < Item = SendStat , Error = Error > > { assert ! ( msg get_message ( ) has_snapshot ( ) ) ; let timer = Instant :: now ( ) ; let send_timer = SEND_SNAP_HISTOGRAM . start_coarse_timer ( ) ; let key = { let snap = msg . get_message ( ) . get_snapshot ( ) ; SnapKey :: from_snap ( snap ) ? } ; mgr . register ( key . clone ( ) , SnapEntry :: Sending ) ; let deregister = { let ( mgr , key ) = ( mgr . clone ( ) , key . clone ( ) ) ; DeferContext :: new ( move | | mgr . deregister ( & key , & SnapEntry :: Sending ) ) } ; let s = box_try ! ( mgr get_snapshot_for_sending ( key ) ) ; if ! s . exists ( ) { return Err ( box_err ! ( " " s path ( ) ) ) ; } let total_size = s . total_size ( ) ? ; let chunks = { let mut first_chunk = SnapshotChunk :: default ( ) ; first_chunk . set_message ( msg ) ; SnapChunk { first : Some ( first_chunk ) , snap : s , remain_bytes : total_size as usize , } } ; let cb = ChannelBuilder :: new ( env ) . stream_initial_window_size ( cfg . grpc_stream_initial_window_size . 0 as i32 ) . keepalive_time ( cfg . grpc_keepalive_time . 0 ) . keepalive_timeout ( cfg . grpc_keepalive_timeout . 0 ) . default_compression_algorithm ( cfg . grpc_compression_algorithm ( ) ) ; let channel = security_mgr . connect ( cb , addr ) ; let client = TikvClient :: new ( channel ) ; let ( sink , receiver ) = client . snapshot ( ) ? ; let send = chunks . forward ( sink ) . map_err ( Error :: from ) ; let send = send . and_then ( | ( s , _ ) | receiver . map_err ( Error :: from ) . map ( | _ | s ) ) . then ( move | result | { send_timer . observe_duration ( ) ; drop ( deregister ) ; drop ( client ) ; result . map ( | s | { fail_point ! ( " " ) ; s . snap . delete ( ) ; // TODO: improve it after rustc resolves the bug. // Call `info` in the closure directly will cause rustc // panic with `Cannot create local mono-item for DefId`. SendStat { key , total_size , elapsed : timer . elapsed ( ) , } } ) } ) ; Ok ( send ) }
Send the snapshot to specified address.
Send the snapshot to specified address .
env
Arc<Environment>
mgr
SnapManager
security_mgr
Arc<SecurityManager>
cfg
&Config
addr
&str
msg
RaftMessage