From 1658ea65fdbf7fb096ae57db3b8b4af1906438ea Mon Sep 17 00:00:00 2001 From: Jonathan Zernik Date: Mon, 20 Sep 2021 20:53:44 -0700 Subject: [PATCH] Do handshake in connection class, use queue for synchronous connect peer (#1370) * Do handshake in connection class, use queue for synchronous connect peer * Update handshake timer in connection class * Initialize peer in connection manager * Remove old comments in peer handler * Remove unused local_address param from peer handler * Remove old comment for sleep in itest * Fix stop waiting indicator after peer connect fail --- .../ConnectPeerDialog/ConnectPeerDialog.js | 1 + .../ImportSigningProfileDialog.js | 1 - frontend/src/pages/peeraddress/PeerAddress.js | 1 + itests/tests/test_squeak_node.py | 1 - squeaknode/network/connection.py | 55 +++++++++++ squeaknode/network/connection_manager.py | 33 ++++++- squeaknode/network/network_manager.py | 5 +- squeaknode/network/peer_client.py | 54 +++++++++- squeaknode/network/peer_handler.py | 99 +++++-------------- 9 files changed, 167 insertions(+), 83 deletions(-) diff --git a/frontend/src/components/ConnectPeerDialog/ConnectPeerDialog.js b/frontend/src/components/ConnectPeerDialog/ConnectPeerDialog.js index 51d8e9e0..ab13d8ab 100644 --- a/frontend/src/components/ConnectPeerDialog/ConnectPeerDialog.js +++ b/frontend/src/components/ConnectPeerDialog/ConnectPeerDialog.js @@ -100,6 +100,7 @@ export default function ConnectPeerDialog({ required variant="outlined" label="Host" + autoFocus value={host} onChange={handleChangeHost} inputProps={{ maxLength: 128 }} diff --git a/frontend/src/components/ImportSigningProfileDialog/ImportSigningProfileDialog.js b/frontend/src/components/ImportSigningProfileDialog/ImportSigningProfileDialog.js index 68381d71..a2674591 100644 --- a/frontend/src/components/ImportSigningProfileDialog/ImportSigningProfileDialog.js +++ b/frontend/src/components/ImportSigningProfileDialog/ImportSigningProfileDialog.js @@ -87,7 +87,6 @@ export default function ImportSigningProfileDialog({ id="standard-textarea" label="Private Key" required - autoFocus value={privateKey} onChange={handleChangePrivateKey} fullWidth diff --git a/frontend/src/pages/peeraddress/PeerAddress.js b/frontend/src/pages/peeraddress/PeerAddress.js index 7ba6de2d..21c7aeb2 100644 --- a/frontend/src/pages/peeraddress/PeerAddress.js +++ b/frontend/src/pages/peeraddress/PeerAddress.js @@ -98,6 +98,7 @@ export default function PeerAddressPage() { }; const handleConnectPeerError = (err) => { + setWaitingForConnectedPeer(false); alert(`Connect peer failure: ${err}`); }; diff --git a/itests/tests/test_squeak_node.py b/itests/tests/test_squeak_node.py index 24c67dcf..c543173c 100644 --- a/itests/tests/test_squeak_node.py +++ b/itests/tests/test_squeak_node.py @@ -938,7 +938,6 @@ def test_connect_peer(admin_stub, other_admin_stub): "squeaknode", 18777, ): - time.sleep(2) connected_peers = get_connected_peers(admin_stub) print(connected_peers) assert len(connected_peers) == 1 diff --git a/squeaknode/network/connection.py b/squeaknode/network/connection.py index 7d685100..b8b9e46b 100644 --- a/squeaknode/network/connection.py +++ b/squeaknode/network/connection.py @@ -47,6 +47,7 @@ logger = logging.getLogger(__name__) EMPTY_HASH = b'\x00' * 32 +HANDSHAKE_TIMEOUT = 30 PING_TIMEOUT = 60 PONG_TIMEOUT = 30 @@ -58,6 +59,10 @@ class Connection(object): def __init__(self, peer: Peer, squeak_controller): self.peer = peer self.squeak_controller = squeak_controller + self.handshake_timer = HandshakeTimer( + self.shutdown, + str(self), + ) self.ping_timer = PingTimer( self.send_ping, str(self.peer), @@ -68,6 +73,21 @@ class Connection(object): str(self.peer), ) + def handshake(self): + """Do a handshake with a peer. + """ + self.handshake_timer.start_timer() + + if self.peer.outgoing: + self.peer.send_version() + self.peer.receive_version() + if not self.peer.outgoing: + self.peer.send_version() + + self.peer.set_connected() + self.handshake_timer.cancel() + logger.debug("HANDSHAKE COMPLETE-----------") + def shutdown(self): logger.debug("Peet shutting down...") self.peer.stop() @@ -305,6 +325,41 @@ class Connection(object): ) +class HandshakeTimer: + """Stop the peer if handshake is not complete before timeout. + """ + + def __init__( + self, + shutdown_fn, + peer_name, + ): + self.shutdown_fn = shutdown_fn + self.peer_name = peer_name + self.timer = None + self._lock = threading.Lock() + + def start_timer(self): + with self._lock: + self.timer = threading.Timer( + HANDSHAKE_TIMEOUT, + self.shutdown, + ) + self.timer.name = "handshake_timer_thread_{}".format( + self.peer_name) + self.timer.start() + + def cancel(self): + logger.debug("Cancelling handshake timer.") + with self._lock: + if self.timer: + self.timer.cancel() + + def shutdown(self): + logger.debug("Shutdown connection triggered by handshake timer.") + self.shutdown_fn() + + class PingTimer: """Send a ping message when the timer expires. """ diff --git a/squeaknode/network/connection_manager.py b/squeaknode/network/connection_manager.py index 6ee5fd09..18ab18bd 100644 --- a/squeaknode/network/connection_manager.py +++ b/squeaknode/network/connection_manager.py @@ -20,6 +20,8 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import logging +import queue +import socket import threading from contextlib import contextmanager from typing import Dict @@ -29,7 +31,9 @@ from typing import Optional from squeaknode.core.peer_address import PeerAddress from squeaknode.network.connection import Connection from squeaknode.network.peer import Peer +from squeaknode.network.peer_client import ConnectPeerResult from squeaknode.node.listener_subscription_client import EventListener +from squeaknode.node.squeak_controller import SqueakController MIN_PEERS = 5 @@ -44,20 +48,43 @@ class ConnectionManager(object): """Maintains connections to other peers in the network. """ - def __init__(self): + def __init__(self, local_address): self._peers: Dict[PeerAddress, Peer] = {} self.peers_lock = threading.Lock() self.peer_changed_listener = EventListener() self.single_peer_changed_listener = EventListener() self.accept_connections = True + self.local_address = local_address @contextmanager - def connect(self, peer: Peer, squeak_controller): + def connect( + self, + peer_socket: socket.socket, + address: PeerAddress, + outgoing: bool, + squeak_controller: SqueakController, + result_queue: queue.Queue, + ): try: + peer = Peer( + peer_socket, + self.local_address, + address, + outgoing, + self.single_peer_changed_listener, + ) + connection = Connection(peer, squeak_controller) + logger.debug("Doing handshake.") + connection.handshake() logger.debug("Adding peer.") self.add_peer(peer) + result_queue.put( + ConnectPeerResult.from_success(peer.remote_address)) logger.debug("Yielding connection.") - yield Connection(peer, squeak_controller) + yield connection + except Exception as e: + result_queue.put(ConnectPeerResult.from_failure(e)) + raise finally: logger.debug("Removing peer.") self.remove_peer(peer) diff --git a/squeaknode/network/network_manager.py b/squeaknode/network/network_manager.py index 0f946771..22bf6b81 100644 --- a/squeaknode/network/network_manager.py +++ b/squeaknode/network/network_manager.py @@ -57,11 +57,10 @@ class NetworkManager(object): self.peer_client = None self.tor_proxy_ip = self.config.node.tor_proxy_ip self.tor_proxy_port = self.config.node.tor_proxy_port - self.connection_manager = ConnectionManager() + self.connection_manager = ConnectionManager(self.local_address) def start(self, squeak_controller): peer_handler = PeerHandler( - self.local_address, self.connection_manager, squeak_controller, ) @@ -88,7 +87,7 @@ class NetworkManager(object): ) if self.connection_manager.has_connection(peer_address): return - self.peer_client.connect_address(peer_address) + self.peer_client.make_connection(peer_address) def disconnect_peer(self, peer_address: PeerAddress) -> None: self.connection_manager.stop_connection(peer_address) diff --git a/squeaknode/network/peer_client.py b/squeaknode/network/peer_client.py index 23ab0c1f..5513cdc1 100644 --- a/squeaknode/network/peer_client.py +++ b/squeaknode/network/peer_client.py @@ -20,7 +20,9 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import logging +import queue import socket +import threading import socks @@ -42,8 +44,22 @@ class PeerClient(object): self.tor_proxy_ip = tor_proxy_ip self.tor_proxy_port = tor_proxy_port - def connect_address(self, address: PeerAddress): + def make_connection(self, address: PeerAddress): logger.info('Making connection to {}'.format(address)) + result_queue: queue.Queue = queue.Queue() + threading.Thread( + target=self.connect_address, + args=(address, result_queue,), + ).start() + + # Wait for connect result from the queue. + connect_result = result_queue.get() + logger.info("connect_result: {}".format(connect_result)) + if connect_result.failure is not None: + raise connect_result.failure + + def connect_address(self, address: PeerAddress, result_queue: queue.Queue): + logger.info('Conecting to address: {}'.format(address)) try: peer_socket = self.get_socket() logger.info('Trying to connect socket to {}'.format(address)) @@ -53,21 +69,24 @@ class PeerClient(object): self.handle_connection( peer_socket, address, + result_queue, ) - except Exception: + except Exception as e: logger.exception('Failed to make connection to {}'.format(address)) - raise + result_queue.put(ConnectPeerResult.from_failure(e)) def handle_connection( self, peer_socket: socket.socket, peer_address: PeerAddress, + result_queue: queue.Queue, ): """Handle a newly connected peer socket.""" self.peer_handler.handle_connection( peer_socket, peer_address, outgoing=True, + result_queue=result_queue, ) def get_socket(self): @@ -76,3 +95,32 @@ class PeerClient(object): s.set_proxy(socks.SOCKS5, self.tor_proxy_ip, self.tor_proxy_port) return s return socket.socket() + + +class ConnectPeerResult(object): + """Result of a connect peer attempt. + """ + + def __init__( + self, + success: PeerAddress = None, + failure: Exception = None, + ): + self.success = success + self.failure = failure + + @classmethod + def from_success(cls, success): + return cls(success=success) + + @classmethod + def from_failure(cls, failure): + return cls(failure=failure) + + def __repr__(self): + return "ConnectPeerResult: \ + success: {} \ + failure: {}".format( + self.success, + self.failure, + ) diff --git a/squeaknode/network/peer_handler.py b/squeaknode/network/peer_handler.py index 26267b08..6eecfb2b 100644 --- a/squeaknode/network/peer_handler.py +++ b/squeaknode/network/peer_handler.py @@ -20,30 +20,26 @@ # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import logging +import queue import socket import threading +from typing import Optional from squeaknode.core.peer_address import PeerAddress -from squeaknode.network.peer import Peer logger = logging.getLogger(__name__) -HANDSHAKE_TIMEOUT = 30 - - class PeerHandler(): """Handles new peer connection. """ def __init__( self, - local_address, connection_manager, squeak_controller, ): - self.local_address = local_address self.connection_manager = connection_manager self.squeak_controller = squeak_controller @@ -52,81 +48,40 @@ class PeerHandler(): peer_socket: socket.socket, address: PeerAddress, outgoing: bool, + result_queue: Optional[queue.Queue] = None, ): """Handle a new socket connection. - This method blocks until the socket connection has stopped. + This method blocks until the peer connection is established. """ - peer = Peer( - peer_socket, - self.local_address, - address, - outgoing, - self.connection_manager.single_peer_changed_listener, - ) - - try: - self.do_handshake(peer) - except Exception: - peer.stop() - raise + # Create a dummy queue if not needed. + if result_queue is None: + result_queue = queue.Queue() threading.Thread( target=self.start_connection, - args=(peer,), - name="handle_peer_connection_thread", + args=( + peer_socket, + address, + outgoing, + result_queue, + ), ).start() - def do_handshake(self, peer: Peer): - """Do a handshake with a peer. - """ - timer = HandshakeTimer( - peer.stop, - str(self), - ) - timer.start_timer() - - if peer.outgoing: - peer.send_version() - peer.receive_version() - if not peer.outgoing: - peer.send_version() - - peer.set_connected() - logger.debug("HANDSHAKE COMPLETE-----------") - timer.stop_timer() - - def start_connection(self, peer: Peer): + def start_connection( + self, + peer_socket: socket.socket, + address: PeerAddress, + outgoing: bool, + result_queue: queue.Queue, + ): """Start a connection """ - with self.connection_manager.connect(peer, self.squeak_controller) as connection: + with self.connection_manager.connect( + peer_socket, + address, + outgoing, + self.squeak_controller, + result_queue, + ) as connection: connection.handle_connection() - - -class HandshakeTimer: - """Stop the peer if handshake is not complete before timeout. - """ - - def __init__(self, - stop_fn, - peer_name, - ): - self.stop_fn = stop_fn - self.peer_name = peer_name - self.timer = None - - def start_timer(self): - self.timer = threading.Timer( - HANDSHAKE_TIMEOUT, - self.stop_peer, - ) - self.timer.name = "handshake_timere_thread_{}".format(self.peer_name) - self.timer.start() - - def stop_timer(self): - logger.debug("Canceling handshake timer.") - self.timer.cancel() - - def stop_peer(self): - logger.info("Closing peer from handshake timer.") - self.stop_fn()