From 4a800e5c56a3f9c1090cab7c0e29f2446d3b9df2 Mon Sep 17 00:00:00 2001 From: Alex Stokes Date: Tue, 24 Sep 2019 18:21:04 -0700 Subject: [PATCH 1/5] Add first-pass at `ping` protocol --- libp2p/host/ping.py | 41 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 41 insertions(+) create mode 100644 libp2p/host/ping.py diff --git a/libp2p/host/ping.py b/libp2p/host/ping.py new file mode 100644 index 00000000..a01609d3 --- /dev/null +++ b/libp2p/host/ping.py @@ -0,0 +1,41 @@ +import logging + +from libp2p.network.stream.net_stream_interface import INetStream +from libp2p.peer.id import ID + +ID = "/ipfs/ping/1.0.0" +PING_LENGTH = 32 +RESP_TIMEOUT = 60 + +logger = logging.getLogger("libp2p.host.ping") + + +async def _handle_ping(stream: INetStream, peer_id: ID) -> None: + try: + payload = await asyncio.wait_for(stream.read(PING_LENGTH), RESP_TIMEOUT) + except asyncio.TimeoutError as error: + logger.debug("Timed out waiting for ping from %s: %s", peer_id, error) + raise + # TODO: handle the other end closing the stream + except Exception as error: + logger.debug("Error while waiting to read ping for %s: %s", peer_id, error) + raise + + logger.debug("Received ping from %s with data: 0x%s", peer_id, payload.hex()) + + await stream.write(payload) + + +async def handle_ping(self, stream: INetStream) -> None: + """ + ``handle_ping`` responds to incoming ping requests until one side + errors or closes the ``stream``. + """ + peer_id = stream.muxed_conn.peer_id + + while True: + try: + await _handle_ping(stream, peer_id) + except Exception: + await stream.reset() + return From f24b488f79b0597dfde3eedd982bb5ec854d4231 Mon Sep 17 00:00:00 2001 From: Alex Stokes Date: Tue, 24 Sep 2019 19:04:18 -0700 Subject: [PATCH 2/5] handle other side closing their end of the connection during `ping` --- libp2p/host/ping.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/libp2p/host/ping.py b/libp2p/host/ping.py index a01609d3..9f43d16a 100644 --- a/libp2p/host/ping.py +++ b/libp2p/host/ping.py @@ -1,5 +1,6 @@ import logging +from libp2p.network.stream.exceptions import StreamEOF, StreamReset from libp2p.network.stream.net_stream_interface import INetStream from libp2p.peer.id import ID @@ -16,7 +17,11 @@ async def _handle_ping(stream: INetStream, peer_id: ID) -> None: except asyncio.TimeoutError as error: logger.debug("Timed out waiting for ping from %s: %s", peer_id, error) raise - # TODO: handle the other end closing the stream + except (StreamEOF, StreamReset) as error: + logger.debug( + "Other side closed while waiting for ping from %s: %s", peer_id, error + ) + raise except Exception as error: logger.debug("Error while waiting to read ping for %s: %s", peer_id, error) raise From e157c3f654cb50b6890c05f6cbfe402694fbe1b6 Mon Sep 17 00:00:00 2001 From: Alex Stokes Date: Tue, 24 Sep 2019 19:15:32 -0700 Subject: [PATCH 3/5] typing fixes --- libp2p/host/ping.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/libp2p/host/ping.py b/libp2p/host/ping.py index 9f43d16a..110a0836 100644 --- a/libp2p/host/ping.py +++ b/libp2p/host/ping.py @@ -1,8 +1,9 @@ +import asyncio import logging from libp2p.network.stream.exceptions import StreamEOF, StreamReset from libp2p.network.stream.net_stream_interface import INetStream -from libp2p.peer.id import ID +from libp2p.peer.id import ID as PeerID ID = "/ipfs/ping/1.0.0" PING_LENGTH = 32 @@ -11,7 +12,7 @@ RESP_TIMEOUT = 60 logger = logging.getLogger("libp2p.host.ping") -async def _handle_ping(stream: INetStream, peer_id: ID) -> None: +async def _handle_ping(stream: INetStream, peer_id: PeerID) -> None: try: payload = await asyncio.wait_for(stream.read(PING_LENGTH), RESP_TIMEOUT) except asyncio.TimeoutError as error: @@ -31,7 +32,7 @@ async def _handle_ping(stream: INetStream, peer_id: ID) -> None: await stream.write(payload) -async def handle_ping(self, stream: INetStream) -> None: +async def handle_ping(stream: INetStream) -> None: """ ``handle_ping`` responds to incoming ping requests until one side errors or closes the ``stream``. From 1cf239cce64aaf15650a9c124892088d655192d3 Mon Sep 17 00:00:00 2001 From: Alex Stokes Date: Thu, 24 Oct 2019 19:44:52 +0900 Subject: [PATCH 4/5] Respect a remote close during the `ping` protocol --- libp2p/host/ping.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/libp2p/host/ping.py b/libp2p/host/ping.py index 110a0836..f47e83d8 100644 --- a/libp2p/host/ping.py +++ b/libp2p/host/ping.py @@ -12,15 +12,21 @@ RESP_TIMEOUT = 60 logger = logging.getLogger("libp2p.host.ping") -async def _handle_ping(stream: INetStream, peer_id: PeerID) -> None: +async def _handle_ping(stream: INetStream, peer_id: PeerID) -> bool: + """ + Return a boolean indicating if we expect more pings from the peer at ``peer_id``. + """ try: payload = await asyncio.wait_for(stream.read(PING_LENGTH), RESP_TIMEOUT) except asyncio.TimeoutError as error: logger.debug("Timed out waiting for ping from %s: %s", peer_id, error) raise - except (StreamEOF, StreamReset) as error: + except StreamEOF: + logger.debug("Other side closed while waiting for ping from %s", peer_id) + return False + except StreamReset as error: logger.debug( - "Other side closed while waiting for ping from %s: %s", peer_id, error + "Other side reset while waiting for ping from %s: %s", peer_id, error ) raise except Exception as error: @@ -30,6 +36,7 @@ async def _handle_ping(stream: INetStream, peer_id: PeerID) -> None: logger.debug("Received ping from %s with data: 0x%s", peer_id, payload.hex()) await stream.write(payload) + return True async def handle_ping(stream: INetStream) -> None: @@ -41,7 +48,9 @@ async def handle_ping(stream: INetStream) -> None: while True: try: - await _handle_ping(stream, peer_id) + should_continue = await _handle_ping(stream, peer_id) + if not should_continue: + return except Exception: await stream.reset() return From d795a9b8542598b4d9d6a95a3d0868a5ffe89479 Mon Sep 17 00:00:00 2001 From: Alex Stokes Date: Thu, 24 Oct 2019 19:50:38 +0900 Subject: [PATCH 5/5] Add tests for `ping` protocol --- tests/host/test_ping.py | 70 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100644 tests/host/test_ping.py diff --git a/tests/host/test_ping.py b/tests/host/test_ping.py new file mode 100644 index 00000000..d6f02449 --- /dev/null +++ b/tests/host/test_ping.py @@ -0,0 +1,70 @@ +import asyncio +import secrets + +import pytest + +from libp2p.host.ping import ID, PING_LENGTH, handle_ping +from libp2p.peer.peerinfo import info_from_p2p_addr +from tests.utils import set_up_nodes_by_transport_opt + + +def _add_ping_to(host): + host.set_stream_handler(ID, handle_ping) + + +@pytest.mark.asyncio +async def test_ping_once(): + transport_opt_list = [["/ip4/127.0.0.1/tcp/0"], ["/ip4/127.0.0.1/tcp/0"]] + (host_a, host_b) = await set_up_nodes_by_transport_opt(transport_opt_list) + + # NOTE: this will fail after we add ping as a default handler + # as a forced reminder to fix this test by removing the calls to + # `_add_ping_to` + assert host_a.get_mux().handlers == {} + assert host_b.get_mux().handlers == {} + _add_ping_to(host_a) + _add_ping_to(host_b) + + addr = host_a.get_addrs()[0] + info = info_from_p2p_addr(addr) + await host_b.connect(info) + + stream = await host_b.new_stream(host_a.get_id(), (ID,)) + some_ping = secrets.token_bytes(PING_LENGTH) + await stream.write(some_ping) + some_pong = await stream.read(PING_LENGTH) + assert some_ping == some_pong + await stream.close() + + +SOME_PING_COUNT = 3 + + +@pytest.mark.asyncio +async def test_ping_several(): + transport_opt_list = [["/ip4/127.0.0.1/tcp/0"], ["/ip4/127.0.0.1/tcp/0"]] + (host_a, host_b) = await set_up_nodes_by_transport_opt(transport_opt_list) + + # NOTE: this will fail after we add ping as a default handler + # as a forced reminder to fix this test by removing the calls to + # `_add_ping_to` + assert host_a.get_mux().handlers == {} + assert host_b.get_mux().handlers == {} + _add_ping_to(host_a) + _add_ping_to(host_b) + + addr = host_a.get_addrs()[0] + info = info_from_p2p_addr(addr) + await host_b.connect(info) + + stream = await host_b.new_stream(host_a.get_id(), (ID,)) + for _ in range(SOME_PING_COUNT): + some_ping = secrets.token_bytes(PING_LENGTH) + await stream.write(some_ping) + some_pong = await stream.read(PING_LENGTH) + assert some_ping == some_pong + # NOTE: simulate some time to sleep to mirror a real + # world usage where a peer sends pings on some periodic interval + # NOTE: this interval can be `0` for this test. + await asyncio.sleep(0) + await stream.close()