Added a few workarounds for HTTP driver's latency issues

nvflare/fuel/f3/drivers/aio_http_driver.py

@@ -85,7 +85,8 @@ async def _async_send_frame(self, frame: BytesAlike):
             # This is to yield control. See bug: https://github.com/aaugustin/websockets/issues/865
             await asyncio.sleep(0)
         except Exception as ex:
-            log.error(f"Error sending frame for connection {self}: {secure_format_exception(ex)}")
+            log.error(f"Error sending frame for connection {self}, closing: {secure_format_exception(ex)}")
+            self.close()
 
 
 class AioHttpDriver(BaseDriver):
@@ -184,5 +185,11 @@ async def _handler(self, websocket):
     async def _read_loop(conn: WsConnection):
         while not conn.closing:
             # Reading from websocket and call receiver CB
-            frame = await conn.websocket.recv()
-            conn.process_frame(frame)
+            try:
+                frame = await conn.websocket.recv()
+                conn.process_frame(frame)
+            except ConnectionClosedOK as ex:
+                raise ex
+            except Exception as ex:
+                log.error(f"Exception {type(ex)} on connection {conn}: {ex}")
+                raise ex

nvflare/fuel/f3/streaming/byte_receiver.py

@@ -39,7 +39,7 @@
 MAX_OUT_SEQ_CHUNKS = 16
 # 1/4 of the window size
 ACK_INTERVAL = 1024 * 1024 * 4
-READ_TIMEOUT = 60
+READ_TIMEOUT = 300
 COUNTER_NAME_RECEIVED = "received"
 
 

nvflare/fuel/f3/streaming/byte_streamer.py

@@ -38,7 +38,7 @@
 
 STREAM_CHUNK_SIZE = 1024 * 1024
 STREAM_WINDOW_SIZE = 16 * STREAM_CHUNK_SIZE
-STREAM_ACK_WAIT = 10
+STREAM_ACK_WAIT = 60
 
 STREAM_TYPE_BYTE = "byte"
 STREAM_TYPE_BLOB = "blob"

nvflare/private/fed/client/fed_client_base.py

@@ -93,7 +93,7 @@ def __init__(
             cell=cell,
             client_register_interval=client_args.get("client_register_interval", 2.0),
             timeout=client_args.get("communication_timeout", 30.0),
-            maint_msg_timeout=client_args.get("maint_msg_timeout", 5.0),
+            maint_msg_timeout=client_args.get("maint_msg_timeout", 30.0),
         )
 
         self.secure_train = secure_train