FEAT: Add CONNECT_TIMEOUT option

python/xoscar/backends/communication/socket.py

@@ -29,7 +29,7 @@
 from urllib.parse import urlparse
 
 from ..._utils import to_binary
-from ...constants import XOSCAR_UNIX_SOCKET_DIR
+from ...constants import XOSCAR_CONNECT_TIMEOUT, XOSCAR_UNIX_SOCKET_DIR
 from ...serialization import AioDeserializer, AioSerializer, deserialize
 from ...utils import classproperty, implements, is_py_312, is_v6_ip
 from .base import Channel, ChannelType, Client, Server
@@ -291,7 +291,13 @@ async def connect(
     ) -> "Client":
         host, port_str = dest_address.rsplit(":", 1)
         port = int(port_str)
-        (reader, writer) = await asyncio.open_connection(host=host, port=port, **kwargs)
+        config = kwargs.get("config", {})
+        connect_timeout = config.get("connect_timeout", XOSCAR_CONNECT_TIMEOUT)
+        fut = asyncio.open_connection(host=host, port=port)
+        try:
+            reader, writer = await asyncio.wait_for(fut, timeout=connect_timeout)
+        except asyncio.TimeoutError:
+            raise ConnectionError("connect timeout")
         channel = SocketChannel(
             reader, writer, local_address=local_address, dest_address=dest_address
         )

python/xoscar/backends/indigen/tests/test_pool.py

@@ -886,7 +886,7 @@ async def test_server_closed():
 
         # check if error raised normally when subprocess killed
         task = asyncio.create_task(actor_ref.sleep(10))
-        await asyncio.sleep(0)
+        await asyncio.sleep(0.1)
 
         # kill subprocess 1
         process = list(pool._sub_processes.values())[0]

python/xoscar/constants.py

@@ -19,3 +19,5 @@
 
 # unix socket.
 XOSCAR_UNIX_SOCKET_DIR = XOSCAR_TEMP_DIR / "socket"
+
+XOSCAR_CONNECT_TIMEOUT = 8