Merge pull request #11181 from nim-lang/pending-future-tracking

Implements pending future tracking in async.

lib/pure/asyncdispatch.nim

@@ -140,6 +140,27 @@
 ## asynchronous IO. A good place to start is the
 ## `asyncnet module <asyncnet.html>`_.
 ##
+## Investigating pending futures
+## =============================
+##
+## It's possible to get into a situation where an async proc, or more accurately
+## a ``Future[T]`` gets stuck and
+## never completes. This can happen for various reasons and can cause serious
+## memory leaks. When this occurs it's hard to identify the procedure that is
+## stuck.
+##
+## Thankfully there is a mechanism which tracks the count of each pending future.
+## All you need to do to enable it is compile with ``-d:futureLogging`` and
+## use the ``getFuturesInProgress`` procedure to get the list of pending futures
+## together with the stack traces to the moment of their creation.
+##
+## You may also find it useful to use this
+## `prometheus package <https://github.com/dom96/prometheus>`_ which will log
+## the pending futures into prometheus, allowing you to analyse them via a nice
+## graph.
+##
+##
+##
 ## Limitations/Bugs
 ## ================
 ##

lib/pure/asyncfutures.nim

@@ -15,7 +15,7 @@ type
     error*: ref Exception ## Stored exception
     errorStackTrace*: string
     when not defined(release):
-      stackTrace: string ## For debugging purposes only.
+      stackTrace: seq[StackTraceEntry] ## For debugging purposes only.
       id: int
       fromProc: string
 
@@ -30,6 +30,44 @@ type
 when not defined(release):
   var currentID = 0
 
+const isFutureLoggingEnabled* = defined(futureLogging)
+when isFutureLoggingEnabled:
+  import hashes
+  type
+    FutureInfo* = object
+      stackTrace*: seq[StackTraceEntry]
+      fromProc*: string
+
+  var futuresInProgress {.threadvar.}: Table[FutureInfo, int]
+
+  proc getFuturesInProgress*(): var Table[FutureInfo, int] =
+    return futuresInProgress
+
+  proc hash(s: StackTraceEntry): Hash =
+    result = hash(s.procname) !& hash(s.line) !&
+      hash(s.filename)
+    result = !$result
+
+  proc hash(fi: FutureInfo): Hash =
+    result = hash(fi.stackTrace) !& hash(fi.fromProc)
+    result = !$result
+
+  proc getFutureInfo(fut: FutureBase): FutureInfo =
+    let info = FutureInfo(
+      stackTrace: fut.stackTrace,
+      fromProc: fut.fromProc
+    )
+    return info
+
+  proc logFutureStart(fut: FutureBase) =
+    let info = getFutureInfo(fut)
+    if info notin getFuturesInProgress():
+      getFuturesInProgress()[info] = 0
+    getFuturesInProgress()[info].inc()
+
+  proc logFutureFinish(fut: FutureBase) =
+    getFuturesInProgress()[getFutureInfo(fut)].dec()
+
 var callSoonProc {.threadvar.}: proc (cbproc: proc ()) {.gcsafe.}
 
 proc getCallSoonProc*(): (proc(cbproc: proc ()) {.gcsafe.}) =
@@ -56,7 +94,7 @@ template setupFutureBase(fromProc: string) =
   new(result)
   result.finished = false
   when not defined(release):
-    result.stackTrace = getStackTrace()
+    result.stackTrace = getStackTraceEntries()
     result.id = currentID
     result.fromProc = fromProc
     currentID.inc()
@@ -67,6 +105,7 @@ proc newFuture*[T](fromProc: string = "unspecified"): Future[T] =
   ## Specifying ``fromProc``, which is a string specifying the name of the proc
   ## that this future belongs to, is a good habit as it helps with debugging.
   setupFutureBase(fromProc)
+  when isFutureLoggingEnabled: logFutureStart(result)
 
 proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] =
   ## Create a new ``FutureVar``. This Future type is ideally suited for
@@ -75,6 +114,7 @@ proc newFutureVar*[T](fromProc = "unspecified"): FutureVar[T] =
   ## Specifying ``fromProc``, which is a string specifying the name of the proc
   ## that this future belongs to, is a good habit as it helps with debugging.
   result = FutureVar[T](newFuture[T](fromProc))
+  when isFutureLoggingEnabled: logFutureStart(Future[T](result))
 
 proc clean*[T](future: FutureVar[T]) =
   ## Resets the ``finished`` status of ``future``.
@@ -92,7 +132,7 @@ proc checkFinished[T](future: Future[T]) =
       msg.add("\n  Future ID: " & $future.id)
       msg.add("\n  Created in proc: " & future.fromProc)
       msg.add("\n  Stack trace to moment of creation:")
-      msg.add("\n" & indent(future.stackTrace.strip(), 4))
+      msg.add("\n" & indent(($future.stackTrace).strip(), 4))
       when T is string:
         msg.add("\n  Contents (string): ")
         msg.add("\n" & indent(future.value.repr, 4))
@@ -143,6 +183,7 @@ proc complete*[T](future: Future[T], val: T) =
   future.value = val
   future.finished = true
   future.callbacks.call()
+  when isFutureLoggingEnabled: logFutureFinish(future)
 
 proc complete*(future: Future[void]) =
   ## Completes a void ``future``.
@@ -151,6 +192,7 @@ proc complete*(future: Future[void]) =
   assert(future.error == nil)
   future.finished = true
   future.callbacks.call()
+  when isFutureLoggingEnabled: logFutureFinish(future)
 
 proc complete*[T](future: FutureVar[T]) =
   ## Completes a ``FutureVar``.
@@ -159,6 +201,7 @@ proc complete*[T](future: FutureVar[T]) =
   assert(fut.error == nil)
   fut.finished = true
   fut.callbacks.call()
+  when isFutureLoggingEnabled: logFutureFinish(Future[T](future))
 
 proc complete*[T](future: FutureVar[T], val: T) =
   ## Completes a ``FutureVar`` with value ``val``.
@@ -170,6 +213,7 @@ proc complete*[T](future: FutureVar[T], val: T) =
   fut.finished = true
   fut.value = val
   fut.callbacks.call()
+  when isFutureLoggingEnabled: logFutureFinish(future)
 
 proc fail*[T](future: Future[T], error: ref Exception) =
   ## Completes ``future`` with ``error``.
@@ -180,6 +224,7 @@ proc fail*[T](future: Future[T], error: ref Exception) =
   future.errorStackTrace =
     if getStackTrace(error) == "": getStackTrace() else: getStackTrace(error)
   future.callbacks.call()
+  when isFutureLoggingEnabled: logFutureFinish(future)
 
 proc clearCallbacks*(future: FutureBase) =
   future.callbacks.function = nil

lib/system/excpt.nim

@@ -445,6 +445,12 @@ proc getStackTraceEntries*(e: ref Exception): seq[StackTraceEntry] =
   else:
     result = move(e.trace)
 
+proc getStackTraceEntries*(): seq[StackTraceEntry] =
+  ## Returns the stack trace entries for the current stack trace.
+  ## This is not yet available for the JS backend.
+  when hasSomeStackTrace:
+    rawWriteStackTrace(result)
+
 const nimCallDepthLimit {.intdefine.} = 2000
 
 proc callDepthLimitReached() {.noinline.} =