Merge branch 'main' into limit_concurrency

Author: seetadev

libp2p/__init__.py

@@ -84,6 +84,8 @@
 # Multiplexer options
 MUXER_YAMUX = "YAMUX"
 MUXER_MPLEX = "MPLEX"
+DEFAULT_NEGOTIATE_TIMEOUT = 5
+
 
 
 def set_default_muxer(muxer_name: Literal["YAMUX", "MPLEX"]) -> None:
@@ -249,6 +251,7 @@ def new_host(
     muxer_preference: Literal["YAMUX", "MPLEX"] | None = None,
     listen_addrs: Sequence[multiaddr.Multiaddr] | None = None,
     enable_mDNS: bool = False,
+    negotiate_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
 ) -> IHost:
     """
     Create a new libp2p host based on the given parameters.
@@ -274,6 +277,6 @@ def new_host(
 
     if disc_opt is not None:
         return RoutedHost(swarm, disc_opt, enable_mDNS)
-    return BasicHost(swarm, enable_mDNS)
+    return BasicHost(network=swarm,enable_mDNS=enable_mDNS , negotitate_timeout=negotiate_timeout)
 
 __version__ = __version("libp2p")

libp2p/host/basic_host.py

@@ -71,6 +71,7 @@
 
 
 logger = logging.getLogger("libp2p.network.basic_host")
+DEFAULT_NEGOTIATE_TIMEOUT = 5
 
 
 class BasicHost(IHost):
@@ -92,10 +93,12 @@ def __init__(
         network: INetworkService,
         enable_mDNS: bool = False,
         default_protocols: Optional["OrderedDict[TProtocol, StreamHandlerFn]"] = None,
+        negotitate_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
     ) -> None:
         self._network = network
         self._network.set_stream_handler(self._swarm_stream_handler)
         self.peerstore = self._network.peerstore
+        self.negotiate_timeout = negotitate_timeout
         # Protocol muxing
         default_protocols = default_protocols or get_default_protocols(self)
         self.multiselect = Multiselect(dict(default_protocols.items()))
@@ -189,7 +192,10 @@ def set_stream_handler(
         self.multiselect.add_handler(protocol_id, stream_handler)
 
     async def new_stream(
-        self, peer_id: ID, protocol_ids: Sequence[TProtocol]
+        self,
+        peer_id: ID,
+        protocol_ids: Sequence[TProtocol],
+        negotitate_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
     ) -> INetStream:
         """
         :param peer_id: peer_id that host is connecting
@@ -201,7 +207,9 @@ async def new_stream(
         # Perform protocol muxing to determine protocol to use
         try:
             selected_protocol = await self.multiselect_client.select_one_of(
-                list(protocol_ids), MultiselectCommunicator(net_stream)
+                list(protocol_ids),
+                MultiselectCommunicator(net_stream),
+                negotitate_timeout,
             )
         except MultiselectClientError as error:
             logger.debug("fail to open a stream to peer %s, error=%s", peer_id, error)
@@ -211,7 +219,12 @@ async def new_stream(
         net_stream.set_protocol(selected_protocol)
         return net_stream
 
-    async def send_command(self, peer_id: ID, command: str) -> list[str]:
+    async def send_command(
+        self,
+        peer_id: ID,
+        command: str,
+        response_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
+    ) -> list[str]:
         """
         Send a multistream-select command to the specified peer and return
         the response.
@@ -225,7 +238,7 @@ async def send_command(self, peer_id: ID, command: str) -> list[str]:
 
         try:
             response = await self.multiselect_client.query_multistream_command(
-                MultiselectCommunicator(new_stream), command
+                MultiselectCommunicator(new_stream), command, response_timeout
             )
         except MultiselectClientError as error:
             logger.debug("fail to open a stream to peer %s, error=%s", peer_id, error)
@@ -264,7 +277,7 @@ async def _swarm_stream_handler(self, net_stream: INetStream) -> None:
         # Perform protocol muxing to determine protocol to use
         try:
             protocol, handler = await self.multiselect.negotiate(
-                MultiselectCommunicator(net_stream)
+                MultiselectCommunicator(net_stream), self.negotiate_timeout
             )
         except MultiselectError as error:
             peer_id = net_stream.muxed_conn.peer_id

libp2p/protocol_muxer/multiselect.py

@@ -1,3 +1,5 @@
+import trio
+
 from libp2p.abc import (
     IMultiselectCommunicator,
     IMultiselectMuxer,
@@ -14,6 +16,7 @@
 
 MULTISELECT_PROTOCOL_ID = "/multistream/1.0.0"
 PROTOCOL_NOT_FOUND_MSG = "na"
+DEFAULT_NEGOTIATE_TIMEOUT = 5
 
 
 class Multiselect(IMultiselectMuxer):
@@ -47,47 +50,56 @@ def add_handler(
 
     # FIXME: Make TProtocol Optional[TProtocol] to keep types consistent
     async def negotiate(
-        self, communicator: IMultiselectCommunicator
+        self,
+        communicator: IMultiselectCommunicator,
+        negotiate_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
     ) -> tuple[TProtocol, StreamHandlerFn | None]:
         """
         Negotiate performs protocol selection.
 
         :param stream: stream to negotiate on
+        :param negotiate_timeout: timeout for negotiation
         :return: selected protocol name, handler function
         :raise MultiselectError: raised when negotiation failed
         """
-        await self.handshake(communicator)
-
-        while True:
-            try:
-                command = await communicator.read()
-            except MultiselectCommunicatorError as error:
-                raise MultiselectError() from error
-
-            if command == "ls":
-                supported_protocols = [p for p in self.handlers.keys() if p is not None]
-                response = "\n".join(supported_protocols) + "\n"
-
-                try:
-                    await communicator.write(response)
-                except MultiselectCommunicatorError as error:
-                    raise MultiselectError() from error
-
-            else:
-                protocol = TProtocol(command)
-                if protocol in self.handlers:
+        try:
+            with trio.fail_after(negotiate_timeout):
+                await self.handshake(communicator)
+
+                while True:
                     try:
-                        await communicator.write(protocol)
+                        command = await communicator.read()
                     except MultiselectCommunicatorError as error:
                         raise MultiselectError() from error
 
-                    return protocol, self.handlers[protocol]
-                try:
-                    await communicator.write(PROTOCOL_NOT_FOUND_MSG)
-                except MultiselectCommunicatorError as error:
-                    raise MultiselectError() from error
-
-        raise MultiselectError("Negotiation failed: no matching protocol")
+                    if command == "ls":
+                        supported_protocols = [
+                            p for p in self.handlers.keys() if p is not None
+                        ]
+                        response = "\n".join(supported_protocols) + "\n"
+
+                        try:
+                            await communicator.write(response)
+                        except MultiselectCommunicatorError as error:
+                            raise MultiselectError() from error
+
+                    else:
+                        protocol = TProtocol(command)
+                        if protocol in self.handlers:
+                            try:
+                                await communicator.write(protocol)
+                            except MultiselectCommunicatorError as error:
+                                raise MultiselectError() from error
+
+                            return protocol, self.handlers[protocol]
+                        try:
+                            await communicator.write(PROTOCOL_NOT_FOUND_MSG)
+                        except MultiselectCommunicatorError as error:
+                            raise MultiselectError() from error
+
+                raise MultiselectError("Negotiation failed: no matching protocol")
+        except trio.TooSlowError:
+            raise MultiselectError("handshake read timeout")
 
     async def handshake(self, communicator: IMultiselectCommunicator) -> None:
         """

libp2p/protocol_muxer/multiselect_client.py

@@ -2,6 +2,8 @@
     Sequence,
 )
 
+import trio
+
 from libp2p.abc import (
     IMultiselectClient,
     IMultiselectCommunicator,
@@ -17,6 +19,7 @@
 
 MULTISELECT_PROTOCOL_ID = "/multistream/1.0.0"
 PROTOCOL_NOT_FOUND_MSG = "na"
+DEFAULT_NEGOTIATE_TIMEOUT = 5
 
 
 class MultiselectClient(IMultiselectClient):
@@ -40,14 +43,18 @@ async def handshake(self, communicator: IMultiselectCommunicator) -> None:
 
         try:
             handshake_contents = await communicator.read()
+
         except MultiselectCommunicatorError as error:
             raise MultiselectClientError() from error
 
         if not is_valid_handshake(handshake_contents):
             raise MultiselectClientError("multiselect protocol ID mismatch")
 
     async def select_one_of(
-        self, protocols: Sequence[TProtocol], communicator: IMultiselectCommunicator
+        self,
+        protocols: Sequence[TProtocol],
+        communicator: IMultiselectCommunicator,
+        negotitate_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
     ) -> TProtocol:
         """
         For each protocol, send message to multiselect selecting protocol and
@@ -56,49 +63,65 @@ async def select_one_of(
 
         :param protocol: protocol to select
         :param communicator: communicator to use to communicate with counterparty
+        :param negotiate_timeout: timeout for negotiation
         :return: selected protocol
         :raise MultiselectClientError: raised when protocol negotiation failed
         """
-        await self.handshake(communicator)
-
-        for protocol in protocols:
-            try:
-                selected_protocol = await self.try_select(communicator, protocol)
-                return selected_protocol
-            except MultiselectClientError:
-                pass
-
-        raise MultiselectClientError("protocols not supported")
+        try:
+            with trio.fail_after(negotitate_timeout):
+                await self.handshake(communicator)
+
+                for protocol in protocols:
+                    try:
+                        selected_protocol = await self.try_select(
+                            communicator, protocol
+                        )
+                        return selected_protocol
+                    except MultiselectClientError:
+                        pass
+
+                raise MultiselectClientError("protocols not supported")
+        except trio.TooSlowError:
+            raise MultiselectClientError("response timed out")
 
     async def query_multistream_command(
-        self, communicator: IMultiselectCommunicator, command: str
+        self,
+        communicator: IMultiselectCommunicator,
+        command: str,
+        response_timeout: int = DEFAULT_NEGOTIATE_TIMEOUT,
     ) -> list[str]:
         """
         Send a multistream-select command over the given communicator and return
         parsed response.
 
         :param communicator: communicator to use to communicate with counterparty
         :param command: supported multistream-select command(e.g., ls)
+        :param negotiate_timeout: timeout for negotiation
         :raise MultiselectClientError: If the communicator fails to process data.
         :return: list of strings representing the response from peer.
         """
-        await self.handshake(communicator)
+        try:
+            with trio.fail_after(response_timeout):
+                await self.handshake(communicator)
 
-        if command == "ls":
-            try:
-                await communicator.write("ls")
-            except MultiselectCommunicatorError as error:
-                raise MultiselectClientError() from error
-        else:
-            raise ValueError("Command not supported")
+                if command == "ls":
+                    try:
+                        await communicator.write("ls")
+                    except MultiselectCommunicatorError as error:
+                        raise MultiselectClientError() from error
+                else:
+                    raise ValueError("Command not supported")
 
-        try:
-            response = await communicator.read()
-            response_list = response.strip().splitlines()
-        except MultiselectCommunicatorError as error:
-            raise MultiselectClientError() from error
+                try:
+                    response = await communicator.read()
+                    response_list = response.strip().splitlines()
+
+                except MultiselectCommunicatorError as error:
+                    raise MultiselectClientError() from error
 
-        return response_list
+                return response_list
+        except trio.TooSlowError:
+            raise MultiselectClientError("command response timed out")
 
     async def try_select(
         self, communicator: IMultiselectCommunicator, protocol: TProtocol
@@ -118,6 +141,7 @@ async def try_select(
 
         try:
             response = await communicator.read()
+
         except MultiselectCommunicatorError as error:
             raise MultiselectClientError() from error
 

libp2p/stream_muxer/muxer_multistream.py

@@ -31,9 +31,6 @@
     Yamux,
 )
 
-# FIXME: add negotiate timeout to `MuxerMultistream`
-DEFAULT_NEGOTIATE_TIMEOUT = 60
-
 
 class MuxerMultistream:
     """

newsfragments/696.bugfix.rst

@@ -0,0 +1,4 @@
+Add timeout wrappers in:
+1. multiselect.py: `negotiate` function
+2. multiselect_client.py: `select_one_of` , `query_multistream_command` functions
+to prevent indefinite hangs when a remote peer does not respond.