libp2p_ping/
protocol.rs

1// Copyright 2018 Parity Technologies (UK) Ltd.
2//
3// Permission is hereby granted, free of charge, to any person obtaining a
4// copy of this software and associated documentation files (the "Software"),
5// to deal in the Software without restriction, including without limitation
6// the rights to use, copy, modify, merge, publish, distribute, sublicense,
7// and/or sell copies of the Software, and to permit persons to whom the
8// Software is furnished to do so, subject to the following conditions:
9//
10// The above copyright notice and this permission notice shall be included in
11// all copies or substantial portions of the Software.
12//
13// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
14// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
19// DEALINGS IN THE SOFTWARE.
20
21use futures::prelude::*;
22use libp2p_swarm::StreamProtocol;
23use rand::{distributions, prelude::*};
24use std::{io, time::Duration};
25use web_time::Instant;
26
27pub const PROTOCOL_NAME: StreamProtocol = StreamProtocol::new("/ipfs/ping/1.0.0");
28
29/// The `Ping` protocol upgrade.
30///
31/// The ping protocol sends 32 bytes of random data in configurable
32/// intervals over a single outbound substream, expecting to receive
33/// the same bytes as a response. At the same time, incoming pings
34/// on inbound substreams are answered by sending back the received bytes.
35///
36/// At most a single inbound and outbound substream is kept open at
37/// any time. In case of a ping timeout or another error on a substream, the
38/// substream is dropped.
39///
40/// Successful pings report the round-trip time.
41///
42/// > **Note**: The round-trip time of a ping may be subject to delays induced
43/// >           by the underlying transport, e.g. in the case of TCP there is
44/// >           Nagle's algorithm, delayed acks and similar configuration options
45/// >           which can affect latencies especially on otherwise low-volume
46/// >           connections.
47
48const PING_SIZE: usize = 32;
49
50/// Sends a ping and waits for the pong.
51pub(crate) async fn send_ping<S>(mut stream: S) -> io::Result<(S, Duration)>
52where
53    S: AsyncRead + AsyncWrite + Unpin,
54{
55    let payload: [u8; PING_SIZE] = thread_rng().sample(distributions::Standard);
56    stream.write_all(&payload).await?;
57    stream.flush().await?;
58    let started = Instant::now();
59    let mut recv_payload = [0u8; PING_SIZE];
60    stream.read_exact(&mut recv_payload).await?;
61    if recv_payload == payload {
62        Ok((stream, started.elapsed()))
63    } else {
64        Err(io::Error::new(
65            io::ErrorKind::InvalidData,
66            "Ping payload mismatch",
67        ))
68    }
69}
70
71/// Waits for a ping and sends a pong.
72pub(crate) async fn recv_ping<S>(mut stream: S) -> io::Result<S>
73where
74    S: AsyncRead + AsyncWrite + Unpin,
75{
76    let mut payload = [0u8; PING_SIZE];
77    stream.read_exact(&mut payload).await?;
78    stream.write_all(&payload).await?;
79    stream.flush().await?;
80    Ok(stream)
81}
82
83#[cfg(test)]
84mod tests {
85    use super::*;
86    use futures::StreamExt;
87    use libp2p_core::{
88        multiaddr::multiaddr,
89        transport::{memory::MemoryTransport, DialOpts, ListenerId, PortUse, Transport},
90        Endpoint,
91    };
92
93    #[test]
94    fn ping_pong() {
95        let mem_addr = multiaddr![Memory(thread_rng().gen::<u64>())];
96        let mut transport = MemoryTransport::new().boxed();
97        transport.listen_on(ListenerId::next(), mem_addr).unwrap();
98
99        let listener_addr = transport
100            .select_next_some()
101            .now_or_never()
102            .and_then(|ev| ev.into_new_address())
103            .expect("MemoryTransport not listening on an address!");
104
105        async_std::task::spawn(async move {
106            let transport_event = transport.next().await.unwrap();
107            let (listener_upgrade, _) = transport_event.into_incoming().unwrap();
108            let conn = listener_upgrade.await.unwrap();
109            recv_ping(conn).await.unwrap();
110        });
111
112        async_std::task::block_on(async move {
113            let c = MemoryTransport::new()
114                .dial(
115                    listener_addr,
116                    DialOpts {
117                        role: Endpoint::Dialer,
118                        port_use: PortUse::Reuse,
119                    },
120                )
121                .unwrap()
122                .await
123                .unwrap();
124            let (_, rtt) = send_ping(c).await.unwrap();
125            assert!(rtt > Duration::from_secs(0));
126        });
127    }
128}
libp2p_ping/protocol.rs

libp2p_ping/
protocol.rs
