Merge #13512: [qa] mininode: Expose connection state through is_connected

fa1eac9cdb [qa] mininode: Expose connection state through is_connected (MarcoFalke)

Pull request description:

  This gets rid of some non-type safe string comparisons and access to members that are implementation details of `class P2PConnection(asyncore.dispatcher)`. Such refactoring is required to replace the deprecated asyncore with something more sane.

  Changes:
  * Get rid of non-enum member `state` and replace is with bool `connected`
  * Get rid of confusing argument `pushbuf` and literally just push to the buffer at the call site

Tree-SHA512: 09074c7e5ed251a2e0509ef205ab82f89887c1e1fa1cc6efc1db60d196eb2403788a4987df8809fd06d80ef652e614c5d3c3fdef70096fc5815102243388288d

test/functional/feature_assumevalid.py

@@ -68,12 +68,12 @@ class AssumeValidTest(BitcoinTestFramework):
     def send_blocks_until_disconnected(self, p2p_conn):
         """Keep sending blocks to the node until we're disconnected."""
         for i in range(len(self.blocks)):
-            if p2p_conn.state != "connected":
+            if not p2p_conn.is_connected:
                 break
             try:
                 p2p_conn.send_message(msg_block(self.blocks[i]))
             except IOError as e:
-                assert str(e) == 'Not connected, no pushbuf'
+                assert not p2p_conn.is_connected
                 break
 
     def assert_blockchain_height(self, node, height):

test/functional/p2p_compactblocks.py

@@ -87,7 +87,7 @@ class TestP2PConn(P2PInterface):
         This is used when we want to send a message into the node that we expect
         will get us disconnected, eg an invalid block."""
         self.send_message(message)
-        wait_until(lambda: self.state != "connected", timeout=timeout, lock=mininode_lock)
+        wait_until(lambda: not self.is_connected, timeout=timeout, lock=mininode_lock)
 
 class CompactBlocksTest(BitcoinTestFramework):
     def set_test_params(self):

test/functional/p2p_leak.py

@@ -118,11 +118,11 @@ class P2PLeakTest(BitcoinTestFramework):
         time.sleep(5)
 
         #This node should have been banned
-        assert no_version_bannode.state != "connected"
+        assert not no_version_bannode.is_connected
 
         # These nodes should have been disconnected
-        assert unsupported_service_bit5_node.state != "connected"
+        assert not unsupported_service_bit5_node.is_connected
-        assert unsupported_service_bit7_node.state != "connected"
+        assert not unsupported_service_bit7_node.is_connected
 
         self.nodes[0].disconnect_p2ps()
 

test/functional/p2p_timeouts.py

@@ -47,9 +47,9 @@ class TimeoutsTest(BitcoinTestFramework):
 
         sleep(1)
 
-        assert no_verack_node.connected
+        assert no_verack_node.is_connected
-        assert no_version_node.connected
+        assert no_version_node.is_connected
-        assert no_send_node.connected
+        assert no_send_node.is_connected
 
         no_verack_node.send_message(msg_ping())
         no_version_node.send_message(msg_ping())
@@ -58,18 +58,18 @@ class TimeoutsTest(BitcoinTestFramework):
 
         assert "version" in no_verack_node.last_message
 
-        assert no_verack_node.connected
+        assert no_verack_node.is_connected
-        assert no_version_node.connected
+        assert no_version_node.is_connected
-        assert no_send_node.connected
+        assert no_send_node.is_connected
 
         no_verack_node.send_message(msg_ping())
         no_version_node.send_message(msg_ping())
 
         sleep(31)
 
-        assert not no_verack_node.connected
+        assert not no_verack_node.is_connected
-        assert not no_version_node.connected
+        assert not no_version_node.is_connected
-        assert not no_send_node.connected
+        assert not no_send_node.is_connected
 
 if __name__ == '__main__':
     TimeoutsTest().main()

test/functional/test_framework/mininode.py

@@ -77,6 +77,12 @@ class P2PConnection(asyncore.dispatcher):
 
         super().__init__(map=mininode_socket_map)
 
+        self._conn_open = False
+
+    @property
+    def is_connected(self):
+        return self._conn_open
+
     def peer_connect(self, dstaddr, dstport, net="regtest"):
         self.dstaddr = dstaddr
         self.dstport = dstport
@@ -84,7 +90,7 @@ class P2PConnection(asyncore.dispatcher):
         self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
         self.sendbuf = b""
         self.recvbuf = b""
-        self.state = "connecting"
+        self._asyncore_pre_connection = True
         self.network = net
         self.disconnect = False
 
@@ -97,22 +103,23 @@ class P2PConnection(asyncore.dispatcher):
 
     def peer_disconnect(self):
         # Connection could have already been closed by other end.
-        if self.state == "connected":
+        if self.is_connected:
-            self.disconnect_node()
+            self.disconnect = True  # Signal asyncore to disconnect
 
     # Connection and disconnection methods
 
     def handle_connect(self):
         """asyncore callback when a connection is opened."""
-        if self.state != "connected":
+        if not self.is_connected:
             logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport))
-            self.state = "connected"
+            self._conn_open = True
+            self._asyncore_pre_connection = False
             self.on_open()
 
     def handle_close(self):
         """asyncore callback when a connection is closed."""
         logger.debug("Closing connection to: %s:%d" % (self.dstaddr, self.dstport))
-        self.state = "closed"
+        self._conn_open = False
         self.recvbuf = b""
         self.sendbuf = b""
         try:
@@ -121,13 +128,6 @@ class P2PConnection(asyncore.dispatcher):
             pass
         self.on_close()
 
-    def disconnect_node(self):
-        """Disconnect the p2p connection.
-
-        Called by the test logic thread. Causes the p2p connection
-        to be disconnected on the next iteration of the asyncore loop."""
-        self.disconnect = True
-
     # Socket read methods
 
     def handle_read(self):
@@ -182,9 +182,8 @@ class P2PConnection(asyncore.dispatcher):
     def writable(self):
         """asyncore method to determine whether the handle_write() callback should be called on the next loop."""
         with mininode_lock:
-            pre_connection = self.state == "connecting"
             length = len(self.sendbuf)
-        return (length > 0 or pre_connection)
+        return length > 0 or self._asyncore_pre_connection
 
     def handle_write(self):
         """asyncore callback when data should be written to the socket."""
@@ -192,7 +191,7 @@ class P2PConnection(asyncore.dispatcher):
             # asyncore does not expose socket connection, only the first read/write
             # event, thus we must check connection manually here to know when we
             # actually connect
-            if self.state == "connecting":
+            if self._asyncore_pre_connection:
                 self.handle_connect()
             if not self.writable():
                 return
@@ -204,14 +203,29 @@ class P2PConnection(asyncore.dispatcher):
                 return
             self.sendbuf = self.sendbuf[sent:]
 
-    def send_message(self, message, pushbuf=False):
+    def send_message(self, message):
         """Send a P2P message over the socket.
 
         This method takes a P2P payload, builds the P2P header and adds
         the message to the send buffer to be sent over the socket."""
-        if self.state != "connected" and not pushbuf:
+        if not self.is_connected:
-            raise IOError('Not connected, no pushbuf')
+            raise IOError('Not connected')
         self._log_message("send", message)
+        tmsg = self._build_message(message)
+        with mininode_lock:
+            if len(self.sendbuf) == 0:
+                try:
+                    sent = self.send(tmsg)
+                    self.sendbuf = tmsg[sent:]
+                except BlockingIOError:
+                    self.sendbuf = tmsg
+            else:
+                self.sendbuf += tmsg
+
+    # Class utility methods
+
+    def _build_message(self, message):
+        """Build a serialized P2P message"""
         command = message.command
         data = message.serialize()
         tmsg = MAGIC_BYTES[self.network]
@@ -222,17 +236,7 @@ class P2PConnection(asyncore.dispatcher):
         h = sha256(th)
         tmsg += h[:4]
         tmsg += data
-        with mininode_lock:
+        return tmsg
-            if (len(self.sendbuf) == 0 and not pushbuf):
-                try:
-                    sent = self.send(tmsg)
-                    self.sendbuf = tmsg[sent:]
-                except BlockingIOError:
-                    self.sendbuf = tmsg
-            else:
-                self.sendbuf += tmsg
-
-    # Class utility methods
 
     def _log_message(self, direction, msg):
         """Logs a message being sent or received over the connection."""
@@ -280,7 +284,7 @@ class P2PInterface(P2PConnection):
             vt.addrTo.port = self.dstport
             vt.addrFrom.ip = "0.0.0.0"
             vt.addrFrom.port = 0
-            self.send_message(vt, True)
+            self.sendbuf = self._build_message(vt)  # Will be sent right after handle_connect
 
     # Message receiving methods
 
@@ -348,7 +352,7 @@ class P2PInterface(P2PConnection):
     # Connection helper methods
 
     def wait_for_disconnect(self, timeout=60):
-        test_function = lambda: self.state != "connected"
+        test_function = lambda: not self.is_connected
         wait_until(test_function, timeout=timeout, lock=mininode_lock)
 
     # Message receiving helper methods