From 8daeee2717dd4d37b8a41061fd526ac3f127e191 Mon Sep 17 00:00:00 2001 From: Jonathan Zernik Date: Tue, 26 Oct 2021 15:27:06 -0700 Subject: [PATCH] Return immediately from download when zero connected peers (#1733) --- frontend/src/pages/squeak/Squeak.js | 8 +++++ .../src/pages/squeakaddress/SqueakAddress.js | 4 +++ proto/squeak_admin.proto | 6 ++++ squeaknode/admin/messages.py | 2 ++ .../admin/squeak_admin_server_handler.py | 8 +++-- squeaknode/core/download_result.py | 3 +- squeaknode/network/network_manager.py | 10 +++++- squeaknode/node/active_download_manager.py | 32 ++++++++++++++----- squeaknode/node/squeak_controller.py | 4 +-- tests/node/test_active_download.py | 3 +- 10 files changed, 65 insertions(+), 15 deletions(-) diff --git a/frontend/src/pages/squeak/Squeak.js b/frontend/src/pages/squeak/Squeak.js index 4428cd60..dda493d1 100644 --- a/frontend/src/pages/squeak/Squeak.js +++ b/frontend/src/pages/squeak/Squeak.js @@ -97,6 +97,10 @@ export default function SqueakPage() { setWaitingForDownloadAncestors(true); downloadSqueakRequest(hash, (response) => { setWaitingForDownloadAncestors(false); + if (response.getDownloadResult().getNumberPeers() === 0) { + alert("Unable to download because zero connected peers."); + return; + } setAncestorSqueaks(null); // Temporary fix until component unmounts correcyly getAncestorSqueaks(hash); }); @@ -107,6 +111,10 @@ export default function SqueakPage() { setWaitingForDownloadReplies(true); downloadRepliesRequest(hash, (response) => { setWaitingForDownloadReplies(false); + if (response.getDownloadResult().getNumberPeers() === 0) { + alert("Unable to download because zero connected peers."); + return; + } setReplySqueaks(null); // Temporary fix until component unmounts correcyly getReplySqueaks(hash, SQUEAKS_PER_PAGE, null); }); diff --git a/frontend/src/pages/squeakaddress/SqueakAddress.js b/frontend/src/pages/squeakaddress/SqueakAddress.js index 25fe953f..ba362d97 100644 --- a/frontend/src/pages/squeakaddress/SqueakAddress.js +++ b/frontend/src/pages/squeakaddress/SqueakAddress.js @@ -88,6 +88,10 @@ export default function SqueakAddressPage() { setWaitingForDownload(true); downloadAddressSqueaksRequest(address, (response) => { setWaitingForDownload(false); + if (response.getDownloadResult().getNumberPeers() === 0) { + alert("Unable to download because zero connected peers."); + return; + } setSqueaks([]); getSqueaks(address, SQUEAKS_PER_PAGE, null); }); diff --git a/proto/squeak_admin.proto b/proto/squeak_admin.proto index eec159c7..82b52003 100644 --- a/proto/squeak_admin.proto +++ b/proto/squeak_admin.proto @@ -843,6 +843,12 @@ message DownloadResult { /// Number of squeaks requested int32 number_requested = 2; + + /// Number of peers downloaded from + int32 number_peers = 3; + + /// Download time in milliseconds. + int32 elapsed_time_ms = 4; } message DownloadSqueaksReply { diff --git a/squeaknode/admin/messages.py b/squeaknode/admin/messages.py index f16ec040..967621f4 100644 --- a/squeaknode/admin/messages.py +++ b/squeaknode/admin/messages.py @@ -263,6 +263,8 @@ def download_result_to_message(download_result: DownloadResult) -> squeak_admin_ return squeak_admin_pb2.DownloadResult( number_downloaded=download_result.number_downloaded, number_requested=download_result.number_requested, + number_peers=download_result.number_peers, + elapsed_time_ms=download_result.elapsed_time_ms, ) diff --git a/squeaknode/admin/squeak_admin_server_handler.py b/squeaknode/admin/squeak_admin_server_handler.py index 6345579c..5c6ec7a1 100644 --- a/squeaknode/admin/squeak_admin_server_handler.py +++ b/squeaknode/admin/squeak_admin_server_handler.py @@ -648,8 +648,12 @@ class SqueakAdminServerHandler(object): squeak_hash = bytes.fromhex(squeak_hash_str) logger.info( "Handle download replies for hash: {}".format(squeak_hash_str)) - self.squeak_controller.download_replies(squeak_hash) - return squeak_admin_pb2.DownloadRepliesReply() + download_result = self.squeak_controller.download_replies(squeak_hash) + logger.info("Download result: {}".format(download_result)) + download_result_msg = download_result_to_message(download_result) + return squeak_admin_pb2.DownloadRepliesReply( + download_result=download_result_msg, + ) def handle_download_address_squeaks(self, request): squeak_address = request.address diff --git a/squeaknode/core/download_result.py b/squeaknode/core/download_result.py index 1f25bf18..125f2d35 100644 --- a/squeaknode/core/download_result.py +++ b/squeaknode/core/download_result.py @@ -26,4 +26,5 @@ class DownloadResult(NamedTuple): """Represents a payment made by a buyer.""" number_downloaded: int number_requested: int - request_time_s: int + elapsed_time_ms: int + number_peers: int diff --git a/squeaknode/network/network_manager.py b/squeaknode/network/network_manager.py index 99249a1f..0dc1c213 100644 --- a/squeaknode/network/network_manager.py +++ b/squeaknode/network/network_manager.py @@ -100,14 +100,22 @@ class NetworkManager(object): def get_connected_peers(self) -> List[Peer]: return self.connection_manager.peers - def broadcast_msg(self, msg: MsgSerializable) -> None: + def broadcast_msg(self, msg: MsgSerializable) -> int: + """Send a message to all connected peers. + + Returns: + int: the number of peers message was sent to. + """ + count = 0 for peer in self.connection_manager.peers: try: peer.send_msg(msg) + count += 1 except Exception: logger.exception("Failed to send msg to peer: {}".format( peer, )) + return count def update_local_subscriptions(self, locator: CSqueakLocator) -> None: for peer in self.connection_manager.peers: diff --git a/squeaknode/node/active_download_manager.py b/squeaknode/node/active_download_manager.py index bbdedb98..de33feeb 100644 --- a/squeaknode/node/active_download_manager.py +++ b/squeaknode/node/active_download_manager.py @@ -21,6 +21,7 @@ # SOFTWARE. import logging import threading +import time import uuid from abc import ABC from abc import abstractmethod @@ -32,6 +33,7 @@ from typing import Optional from squeak.core import CSqueak from squeak.messages import msg_getdata from squeak.messages import msg_getsqueaks +from squeak.messages import MsgSerializable from squeak.net import CInterested from squeak.net import CInv from squeak.net import CSqueakLocator @@ -53,14 +55,23 @@ class ActiveDownload(ABC): self.count = 0 self._lock = threading.Lock() self.stopped = threading.Event() + self.num_peers = 0 + self.start_time_ms: Optional[int] = None @abstractmethod def is_interested(self, squeak: CSqueak) -> bool: """Return True if the given squeak matches the download interest.""" @abstractmethod + def get_download_msg(self) -> MsgSerializable: + """Get the message to send to peers to get download response.""" + def initiate_download(self, broadcast_fn) -> None: - """Broadcast a message to peers to get data.""" + self.start_time_ms = int(time.time() * 1000) + msg = self.get_download_msg() + self.num_peers = broadcast_fn(msg) + if self.num_peers == 0: + self.mark_complete() def increment(self) -> None: with self._lock: @@ -74,6 +85,12 @@ class ActiveDownload(ABC): def cancel(self): self.stopped.set() + def get_elapsed_time_ms(self): + if self.start_time_ms is None: + return 0 + end_time_ms = int(time.time() * 1000) + return end_time_ms - self.start_time_ms + def wait_for_complete(self, timeout_s: int) -> None: self.stopped.wait(timeout=timeout_s) @@ -81,7 +98,8 @@ class ActiveDownload(ABC): return DownloadResult( number_downloaded=self.count, number_requested=self.limit, - request_time_s=-1, + elapsed_time_ms=self.get_elapsed_time_ms(), + number_peers=self.num_peers, ) @@ -94,14 +112,13 @@ class InterestDownload(ActiveDownload): def is_interested(self, squeak: CSqueak) -> bool: return squeak_matches_interest(squeak, self.interest) - def initiate_download(self, broadcast_fn) -> None: + def get_download_msg(self) -> MsgSerializable: locator = CSqueakLocator( vInterested=[self.interest], ) - getsqueaks_msg = msg_getsqueaks( + return msg_getsqueaks( locator=locator, ) - broadcast_fn(getsqueaks_msg) class HashDownload(ActiveDownload): @@ -113,14 +130,13 @@ class HashDownload(ActiveDownload): def is_interested(self, squeak: CSqueak) -> bool: return self.squeak_hash == get_hash(squeak) - def initiate_download(self, broadcast_fn) -> None: + def get_download_msg(self) -> MsgSerializable: invs = [ CInv(type=1, hash=self.squeak_hash) ] - getdata_msg = msg_getdata( + return msg_getdata( inv=invs, ) - broadcast_fn(getdata_msg) class ActiveDownloadManager: diff --git a/squeaknode/node/squeak_controller.py b/squeaknode/node/squeak_controller.py index 2912bc54..ec09bbaa 100644 --- a/squeaknode/node/squeak_controller.py +++ b/squeaknode/node/squeak_controller.py @@ -737,8 +737,8 @@ class SqueakController: ) return self.active_download_manager.download_interest(10, interest) - def broadcast_msg(self, msg: MsgSerializable) -> None: - self.network_manager.broadcast_msg(msg) + def broadcast_msg(self, msg: MsgSerializable) -> int: + return self.network_manager.broadcast_msg(msg) def disconnect_peer(self, peer_address: PeerAddress) -> None: logger.info("Disconnect to peer: {}".format( diff --git a/tests/node/test_active_download.py b/tests/node/test_active_download.py index d501462f..22f672ab 100644 --- a/tests/node/test_active_download.py +++ b/tests/node/test_active_download.py @@ -126,5 +126,6 @@ def test_download_hash_get_result(download_hash, squeak): assert download_result == DownloadResult( number_downloaded=1, number_requested=1, - request_time_s=-1, + elapsed_time_ms=0, + number_peers=0, )