add instrumentation via optick

libp2p/errors.nim

@@ -65,3 +65,75 @@ template tryAndWarn*(message: static[string]; body: untyped): untyped =
   except CatchableError as exc:
     warn "An exception has ocurred, enable trace logging for details", name = exc.name, msg = message
     trace "Exception details", exc = exc.msg
+
+when defined(profiler_optick):
+  import dynlib
+
+  type
+    OptickEvent* = distinct uint64
+    OptickEventCtx* = distinct uint64
+
+    OptickCreateEvent* = proc(inFunctionName: cstring, inFunctionLength: uint16, inFileName: cstring, inFileNameLenght: uint16, inFileLine: uint32): OptickEvent {.cdecl.}
+    OptickPushEvent* = proc(event: OptickEvent): OptickEventCtx {.cdecl.}
+    OptickPopEvent* = proc(ctx: OptickEventCtx) {.cdecl.}
+    OptickStartCapture* = proc() {.cdecl.}
+    OptickStopCapture* = proc(filename: cstring, nameLen: uint16) {.cdecl.}
+    OptickNextFrame* = proc() {.cdecl.}
+
+  var
+    createEvent*: OptickCreateEvent
+    pushEvent*: OptickPushEvent
+    popEvent*: OptickPopEvent
+    startCapture*: OptickStartCapture
+    stopCapture*: OptickStopCapture
+    nextFrame*: OptickNextFrame
+
+  template profile*(name: string): untyped =
+    {.gcsafe.}:
+      const pos = instantiationInfo()
+      let event_desc {.global.} = createEvent(name.cstring, name.len.uint16, pos.filename.cstring, pos.filename.len.uint16, pos.line.uint32)
+      let ev = pushEvent(event_desc)
+      defer:
+        popEvent(ev)
+
+  proc load() =
+    var candidates: seq[string]
+    libCandidates("OptickCore", candidates)
+    for c in candidates:
+      let lib = loadLib("OptickCore")
+      if lib != nil:
+        createEvent = cast[OptickCreateEvent](lib.symAddr("OptickAPI_CreateEventDescription"))
+        pushEvent = cast[OptickPushEvent](lib.symAddr("OptickAPI_PushEvent"))
+        popEvent = cast[OptickPopEvent](lib.symAddr("OptickAPI_PopEvent"))
+        startCapture = cast[OptickStartCapture](lib.symAddr("OptickAPI_StartCapture"))
+        stopCapture = cast[OptickStopCapture](lib.symAddr("OptickAPI_StopCapture"))
+        nextFrame = cast[OptickNextFrame](lib.symAddr("OptickAPI_NextFrame"))
+        return
+    doAssert(false, "OptickCore failed to load")
+  
+  load()
+
+  startCapture()
+  addQuitProc(proc () {.noconv.} = 
+    stopCapture("profiled.opt", "profiled.opt".len))
+
+  proc frameTicker() {.async.} =
+    while true:
+      {.gcsafe.}:
+        nextFrame()
+      await sleepAsync(100.millis)
+  
+  # let poll_event = createEvent("poll", "poll".len, "", 0, 0)
+  # proc pollHook() {.async.} =
+  #   while true:
+  #     {.gcsafe.}:
+  #       let ev = pushEvent(poll_event)
+  #       await sleepAsync(0)
+  #       defer:
+  #         popEvent(ev)
+      
+  
+  asyncCheck frameTicker()
+  # asyncCheck pollHook()
+else:
+   template profile*(name: string): untyped = discard

libp2p/protocols/pubsub/gossipsub.nim

@@ -101,6 +101,8 @@ proc replenishFanout(g: GossipSub, topic: string) =
   trace "fanout replenished with peers", peers = g.fanout.peers(topic)
 
 proc rebalanceMesh(g: GossipSub, topic: string) {.async.} =
+  profile "rebalanceMesh"
+  
   logScope:
     topic
 
@@ -217,6 +219,8 @@ proc getGossipPeers(g: GossipSub): Table[PubSubPeer, ControlMessage] {.gcsafe.}
 proc heartbeat(g: GossipSub) {.async.} =
   while g.heartbeatRunning:
     try:
+      profile "heartbeat"
+      
       trace "running heartbeat"
 
       for t in toSeq(g.topics.keys):
@@ -386,6 +390,8 @@ proc handleIWant(g: GossipSub,
 method rpcHandler*(g: GossipSub,
                   peer: PubSubPeer,
                   rpcMsgs: seq[RPCMsg]) {.async.} =
+  profile "rpcHandler"
+  
   await procCall PubSub(g).rpcHandler(peer, rpcMsgs)
 
   for m in rpcMsgs:                                  # for all RPC messages
@@ -493,6 +499,8 @@ method publish*(g: GossipSub,
                 topic: string,
                 data: seq[byte],
                 timeout: Duration = InfiniteDuration): Future[int] {.async.} =
+  profile "publish"
+
   # base returns always 0
   discard await procCall PubSub(g).publish(topic, data, timeout)
   trace "publishing message on topic", topic, data = data.shortLog

libp2p/protocols/pubsub/pubsub.nim

@@ -76,6 +76,8 @@ proc broadcast*(p: PubSub,
                 sendPeers: HashSet[PubSubPeer] | seq[PubSubPeer] | array[1, PubSubPeer],
                 msg: RPCMsg,
                 timeout: Duration): Future[int] {.async.} =
+  profile "broadcast"
+
   # send messages and cleanup failed peers
   var sent: seq[tuple[id: PeerID, fut: Future[void]]]
   for sendPeer in sendPeers:

libp2p/switch.nim

@@ -386,6 +386,7 @@ proc dial*(s: Switch,
            addrs: seq[MultiAddress],
            proto: string):
            Future[Connection] {.async.} =
+  profile "dial"
   let conn = await s.internalConnect(peerId, addrs)
   let stream = await s.connManager.getMuxedStream(conn)