Per callback duration stats

chronos/asyncfutures2.nim

@@ -303,6 +303,16 @@ proc internalContinue(fut: pointer) {.raises: [], gcsafe.} =
   GC_unref(asFut)
   futureContinue(asFut)
 
+template timeClosureDuration(fut: FutureBase, cond, blk: untyped) =
+  when cond:
+    let startTick = Moment.now()
+    `blk`
+    let stopTick = Moment.now()
+    fut.internalDuration += (stopTick - startTick)
+  else:
+    `blk`
+
+
 proc futureContinue*(fut: FutureBase) {.raises: [], gcsafe.} =
   # This function is responsible for calling the closure iterator generated by
   # the `{.async.}` transformation either until it has completed its iteration
@@ -315,8 +325,13 @@ proc futureContinue*(fut: FutureBase) {.raises: [], gcsafe.} =
     while true:
       # Call closure to make progress on `fut` until it reaches `yield` (inside
       # `await` typically) or completes / fails / is cancelled
-      next = fut.internalClosure(fut)
+
+      timeClosureDuration(fut, chronosClosureDurationMetric):
+        next = fut.internalClosure(fut)
+
       if fut.internalClosure.finished(): # Reached the end of the transformed proc
+        when chronosClosureDurationMetric:
+          fut.setFutureDuration()
         break
 
       if next == nil:

chronos/config.nim

@@ -70,6 +70,10 @@ when (NimMajor, NimMinor) >= (1, 4):
         ""
       ## OS polling engine type which is going to be used by chronos.
 
+    chronosClosureDurationMetric* {.booldefine.} = defined(chronosClosureDurationMetric)
+      ## At the cost of some performance, produce a 'chronosCallbackDuration' metric.
+      ## Useful for detecting application stalling/blocking.
+
 else:
   # 1.2 doesn't support `booldefine` in `when` properly
   const
@@ -105,6 +109,7 @@ else:
         "poll"
       else:
         ""
+    chronosClosureDurationMetric* {.booldefine.} = defined(chronosClosureDurationMetric)
 
 when defined(debug) or defined(chronosConfig):
   import std/macros

chronos/futures.nim

@@ -14,6 +14,24 @@ import "."/[config, srcloc]
 
 export srcloc
 
+when chronosClosureDurationMetric:
+  import std/tables
+  import timer
+
+  type
+    CallbackDurationMetric* = ref object
+      ## Holds average timing information for a given closure
+      closureLoc*: ptr SrcLoc
+      totalDuration*: Duration
+      minSingleTime*: Duration
+      maxSingleTime*: Duration
+      count*: int64
+
+  var callbackDurations {.threadvar.}: Table[ptr SrcLoc, CallbackDurationMetric]
+
+  proc getCallbackDurations*(): Table[ptr SrcLoc, CallbackDurationMetric] =
+    callbackDurations
+
 when chronosStackTrace:
   type StackTrace = string
 
@@ -54,6 +72,9 @@ type
     when chronosFutureId:
       internalId*: uint
 
+    when chronosClosureDurationMetric:
+      internalDuration*: Duration
+
     when chronosStackTrace:
       internalErrorStackTrace*: StackTrace
       internalStackTrace*: StackTrace ## For debugging purposes only.
@@ -121,6 +142,23 @@ proc internalInitFutureBase*(
         futureList.head = fut
       futureList.count.inc()
 
+  when chronosClosureDurationMetric:
+    discard callbackDurations.hasKeyOrPut(loc, CallbackDurationMetric(minSingleTime: InfiniteDuration))
+
+when chronosClosureDurationMetric:
+  proc setFutureDuration*(fut: FutureBase) {.inline.} =
+    ## used for setting the duration
+    let loc = fut.internalLocation[Create]
+    callbackDurations.withValue(loc, metric):
+      metric.totalDuration += fut.internalDuration
+      metric.count.inc
+      metric.minSingleTime = min(metric.minSingleTime, fut.internalDuration)
+      metric.maxSingleTime = max(metric.maxSingleTime, fut.internalDuration)
+      # handle overflow
+      if metric.count == metric.count.typeof.high:
+        metric.totalDuration = ZeroDuration
+        metric.count = 0
+
 # Public API
 template init*[T](F: type Future[T], fromProc: static[string] = ""): Future[T] =
   ## Creates a new pending future.

config.nims

@@ -8,3 +8,7 @@ when system.fileExists("nimble.paths"):
 # Should be removed when https://github.com/status-im/nim-chronos/issues/284
 # will be implemented.
 switch("define", "nimRawSetjmp")
+
+## REMOVE BEFORE MERGE!
+--d:chronosFutureTracking
+--d:chronosClosureDurationMetric

tests/testutils.nim

@@ -7,6 +7,7 @@
 #              MIT license (LICENSE-MIT)
 import unittest2
 import ../chronos, ../chronos/config
+import std/[tables, os] 
 
 {.used.}
 
@@ -85,3 +86,56 @@ suite "Asynchronous utilities test suite":
         pendingFuturesCount() == 0'u
     else:
       skip()
+
+
+  test "Test Closure During Metrics":
+    when chronosClosureDurationMetric:
+      proc simpleAsync1() {.async.} =
+        os.sleep(50)
+
+      waitFor(simpleAsync1())
+
+      let metrics = getCallbackDurations()
+      for (k,v) in metrics.pairs():
+        let count = v.count
+        let totalDuration = v.totalDuration
+        if count > 0:
+          echo ""
+          echo "metric: ", $k
+          echo "count: ", count
+          echo "total: ", totalDuration
+          echo "avg: ", totalDuration div count
+        if k.procedure == "simpleAsync1":
+          check v.totalDuration <= 60.milliseconds()
+          check v.totalDuration >= 50.milliseconds()
+
+    else:
+      skip()
+
+  test "Test Closure During Metrics await":
+    when chronosClosureDurationMetric:
+      proc simpleAsync2() {.async.} =
+        os.sleep(50)
+        await sleepAsync(50.milliseconds)
+        os.sleep(50)
+
+      waitFor(simpleAsync2())
+
+      let metrics = getCallbackDurations()
+      for (k,v) in metrics.pairs():
+        let count = v.count
+        let totalDuration = v.totalDuration
+        if count > 0:
+          echo ""
+          echo "metric: ", $k
+          echo "count: ", count
+          echo "total: ", totalDuration
+          echo "min: ", v.minSingleTime
+          echo "max: ", v.maxSingleTime
+          echo "avg: ", totalDuration div count
+        if k.procedure == "simpleAsync2":
+          check v.totalDuration <= 120.milliseconds()
+          check v.totalDuration >= 100.milliseconds()
+
+    else:
+      skip()