Moving idle timeout to Connection to enable across all connection streams (#307)

* move idle timeout logic to connection

* more informative logs

* more informative logs

libp2p/errors.nim

@@ -20,7 +20,7 @@ macro checkFutures*[T](futs: seq[Future[T]], exclude: untyped = []): untyped =
           let exc = res.readError()
           # We still don't abort but warn
           warn "A future has failed, enable trace logging for details", error = exc.name
-          trace "Exception message", msg=exc.msg
+          trace "Exception message", msg= exc.msg, stack = getStackTrace()
   else:
     quote do:
       for res in `futs`:

libp2p/muxers/mplex/lpchannel.nim

@@ -46,8 +46,6 @@ logScope:
 type
   LPChannel* = ref object of BufferStream
     id*: uint64                   # channel id
-    timeout: Duration             # channel timeout if no activity
-    activity: bool                # reset every time data is sent or received
     name*: string                 # name of the channel (for debugging)
     conn*: Connection             # wrapped connection used to for writing
     initiator*: bool              # initiated remotely or locally flag
@@ -57,7 +55,6 @@ type
     msgCode*: MessageType         # cached in/out message code
     closeCode*: MessageType       # cached in/out close code
     resetCode*: MessageType       # cached in/out reset code
-    timerTaskFut: Future[void]    # the current timer instanse
 
 proc open*(s: LPChannel) {.async, gcsafe.}
 
@@ -79,11 +76,6 @@ template withEOFExceptions(body: untyped): untyped =
   except LPStreamIncompleteError as exc:
     trace "incomplete message", exc = exc.msg
 
-proc cleanupTimer(s: LPChannel) {.async.} =
-  ## cleanup timers
-  if not isNil(s.timerTaskFut) and not s.timerTaskFut.finished:
-    s.timerTaskFut.cancel()
-
 proc closeMessage(s: LPChannel) {.async.} =
   logScope:
     id = s.id
@@ -150,7 +142,6 @@ proc closeRemote*(s: LPChannel) {.async.} =
 
     # call to avoid leaks
     await procCall BufferStream(s).close() # close parent bufferstream
-    await s.cleanupTimer()
 
     trace "channel closed on EOF"
   except CancelledError as exc:
@@ -193,8 +184,6 @@ method reset*(s: LPChannel) {.base, async, gcsafe.} =
     s.isEof = true
     s.closedLocal = true
 
-    await s.cleanupTimer()
-
   except CancelledError as exc:
     raise exc
   except CatchableError as exc:
@@ -222,7 +211,6 @@ method close*(s: LPChannel) {.async, gcsafe.} =
       await s.closeMessage().wait(2.minutes)
       if s.atEof: # already closed by remote close parent buffer immediately
         await procCall BufferStream(s).close()
-        await s.cleanupTimer()
     except CancelledError as exc:
       await s.reset()
       raise exc
@@ -235,60 +223,16 @@ method close*(s: LPChannel) {.async, gcsafe.} =
   s.closedLocal = true
   asyncCheck closeInternal()
 
-proc timeoutMonitor(s: LPChannel) {.async.} =
-  ## monitor the channel for innactivity
-  ##
-  ## if the timeout was hit, it means that
-  ## neither incoming nor outgoing activity
-  ## has been detected and the channel will
-  ## be reset
-  ##
-
-  logScope:
-    id = s.id
-    initiator = s.initiator
-    name = s.name
-    oid = $s.oid
-    peer = $s.conn.peerInfo
-
-  try:
-    while true:
-      await sleepAsync(s.timeout)
-
-      if s.closed or s.atEof:
-        return
-
-      if s.activity:
-        s.activity = false
-        continue
-
-      break
-
-    # reset channel on innactivity timeout
-    trace "channel timed out, resetting"
-    await s.reset()
-  except CancelledError as exc:
-    raise exc
-  except CatchableError as exc:
-    trace "exception in timeout", exc = exc.msg
-
 method initStream*(s: LPChannel) =
   if s.objName.len == 0:
     s.objName = "LPChannel"
 
+  s.timeoutHandler = proc() {.async, gcsafe.} =
+    trace "idle timeout expired, resetting LPChannel"
+    await s.reset()
+
   procCall BufferStream(s).initStream()
 
-method readOnce*(s: LPChannel,
-                 pbytes: pointer,
-                 nbytes: int):
-                 Future[int] =
-  s.activity = true
-  procCall BufferStream(s).readOnce(pbytes, nbytes)
-
-method write*(s: LPChannel, msg: seq[byte]): Future[void] =
-  s.activity = true
-  procCall BufferStream(s).write(msg)
-
 proc init*(
   L: type LPChannel,
   id: uint64,
@@ -339,7 +283,6 @@ proc init*(
   when chronicles.enabledLogLevel == LogLevel.TRACE:
     chann.name = if chann.name.len > 0: chann.name else: $chann.oid
 
-  chann.timerTaskFut = chann.timeoutMonitor()
   trace "created new lpchannel"
 
   return chann

libp2p/muxers/muxer.nim

@@ -16,7 +16,7 @@ logScope:
   topics = "muxer"
 
 const
-  DefaultChanTimeout* = 1.minutes
+  DefaultChanTimeout* = 5.minutes
 
 type
   StreamHandler* = proc(conn: Connection): Future[void] {.gcsafe.}

libp2p/stream/bufferstream.nim

@@ -220,6 +220,8 @@ method readOnce*(s: BufferStream,
   var index = 0
   var size = min(nbytes, s.len)
   let output = cast[ptr UncheckedArray[byte]](pbytes)
+
+  s.activity = true # reset activity flag
   while s.len() > 0 and index < size:
     output[index] = s.popFirst()
     inc(index)
@@ -243,6 +245,7 @@ method write*(s: BufferStream, msg: seq[byte]) {.async.} =
   if isNil(s.writeHandler):
     raise newNotWritableError()
 
+  s.activity = true # reset activity flag
   await s.writeHandler(msg)
 
 # TODO: move pipe routines out

libp2p/stream/chronosstream.nim

@@ -14,18 +14,28 @@ import connection
 logScope:
   topics = "chronosstream"
 
-type ChronosStream* = ref object of Connection
+const
+  DefaultChronosStreamTimeout = 10.minutes
+
+type
+  ChronosStream* = ref object of Connection
     client: StreamTransport
 
 method initStream*(s: ChronosStream) =
   if s.objName.len == 0:
     s.objName = "ChronosStream"
 
+  s.timeoutHandler = proc() {.async, gcsafe.} =
+    trace "idle timeout expired, closing ChronosStream"
+    await s.close()
+
   procCall Connection(s).initStream()
 
-proc newChronosStream*(client: StreamTransport): ChronosStream =
+proc init*(C: type ChronosStream,
-  new result
+           client: StreamTransport,
-  result.client = client
+           timeout = DefaultChronosStreamTimeout): ChronosStream =
+  result = C(client: client,
+             timeout: timeout)
   result.initStream()
 
 template withExceptions(body: untyped) =
@@ -48,6 +58,7 @@ method readOnce*(s: ChronosStream, pbytes: pointer, nbytes: int): Future[int] {.
 
   withExceptions:
     result = await s.client.readOnce(pbytes, nbytes)
+    s.activity = true # reset activity flag
 
 method write*(s: ChronosStream, msg: seq[byte]) {.async.} =
   if s.closed:
@@ -60,6 +71,7 @@ method write*(s: ChronosStream, msg: seq[byte]) {.async.} =
     var written = 0
     while not s.client.closed and written < msg.len:
       written += await s.client.write(msg[written..<msg.len])
+      s.activity = true # reset activity flag
 
     if written < msg.len:
       raise (ref LPStreamClosedError)(msg: "Write couldn't finish writing")

libp2p/stream/connection.nim

@@ -20,12 +20,19 @@ logScope:
 
 const
   ConnectionTrackerName* = "libp2p.connection"
+  DefaultConnectionTimeout* = 1.minutes
 
 type
+  TimeoutHandler* = proc(): Future[void] {.gcsafe.}
+
   Direction* {.pure.} = enum
     None, In, Out
 
   Connection* = ref object of LPStream
+    activity*: bool                 # reset every time data is sent or received
+    timeout*: Duration              # channel timeout if no activity
+    timerTaskFut: Future[void]      # the current timer instanse
+    timeoutHandler*: TimeoutHandler # timeout handler
     peerInfo*: PeerInfo
     observedAddr*: Multiaddress
     dir*: Direction
@@ -35,6 +42,7 @@ type
     closed*: uint64
 
 proc setupConnectionTracker(): ConnectionTracker {.gcsafe.}
+proc timeoutMonitor(s: Connection) {.async, gcsafe.}
 
 proc getConnectionTracker*(): ConnectionTracker {.gcsafe.} =
   result = cast[ConnectionTracker](getTracker(ConnectionTrackerName))
@@ -58,21 +66,23 @@ proc setupConnectionTracker(): ConnectionTracker =
   result.isLeaked = leakTransport
   addTracker(ConnectionTrackerName, result)
 
-proc init*(C: type Connection,
-           peerInfo: PeerInfo,
-           dir: Direction): Connection =
-  result = C(peerInfo: peerInfo, dir: dir)
-  result.initStream()
-
 method initStream*(s: Connection) =
   if s.objName.len == 0:
     s.objName = "Connection"
 
   procCall LPStream(s).initStream()
   s.closeEvent = newAsyncEvent()
+
+  doAssert(isNil(s.timerTaskFut))
+  s.timerTaskFut = s.timeoutMonitor()
+
   inc getConnectionTracker().opened
 
 method close*(s: Connection) {.async.} =
+  ## cleanup timers
+  if not isNil(s.timerTaskFut) and not s.timerTaskFut.finished:
+    s.timerTaskFut.cancel()
+
   if not s.isClosed:
     await procCall LPStream(s).close()
     inc getConnectionTracker().closed
@@ -83,3 +93,50 @@ proc `$`*(conn: Connection): string =
 
 func hash*(p: Connection): Hash =
   cast[pointer](p).hash
+
+proc timeoutMonitor(s: Connection) {.async, gcsafe.} =
+  ## monitor the channel for innactivity
+  ##
+  ## if the timeout was hit, it means that
+  ## neither incoming nor outgoing activity
+  ## has been detected and the channel will
+  ## be reset
+  ##
+
+  logScope:
+    oid = $s.oid
+
+  try:
+    while true:
+      await sleepAsync(s.timeout)
+
+      if s.closed or s.atEof:
+        return
+
+      if s.activity:
+        s.activity = false
+        continue
+
+      break
+
+    # reset channel on innactivity timeout
+    trace "Connection timed out"
+    if not(isNil(s.timeoutHandler)):
+      await s.timeoutHandler()
+
+  except CancelledError as exc:
+    raise exc
+  except CatchableError as exc:
+    trace "exception in timeout", exc = exc.msg
+
+proc init*(C: type Connection,
+           peerInfo: PeerInfo,
+           dir: Direction,
+           timeout: Duration = DefaultConnectionTimeout,
+           timeoutHandler: TimeoutHandler = nil): Connection =
+  result = C(peerInfo: peerInfo,
+             dir: dir,
+             timeout: timeout,
+             timeoutHandler: timeoutHandler)
+
+  result.initStream()

libp2p/transports/tcptransport.nim

@@ -63,7 +63,7 @@ proc connHandler*(t: TcpTransport,
                   client: StreamTransport,
                   initiator: bool): Connection =
   trace "handling connection", address = $client.remoteAddress
-  let conn: Connection = Connection(newChronosStream(client))
+  let conn: Connection = Connection(ChronosStream.init(client))
   conn.observedAddr = MultiAddress.init(client.remoteAddress).tryGet()
   if not initiator:
     if not isNil(t.handler):