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

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 }}

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

frontend/src/pages/peeraddress/PeerAddress.js

@@ -98,6 +98,7 @@ export default function PeerAddressPage() {
   };
 
   const handleConnectPeerError = (err) => {
+    setWaitingForConnectedPeer(false);
     alert(`Connect peer failure: ${err}`);
   };
 

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

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.
     """

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)

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)

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,
+        )

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()