diff --git a/logos_delivery/waku/waku_core/message.nim b/logos_delivery/waku/waku_core/message.nim index 6b698fb01..3f2d93d8a 100644 --- a/logos_delivery/waku/waku_core/message.nim +++ b/logos_delivery/waku/waku_core/message.nim @@ -1,3 +1,8 @@ -import ./message/message, ./message/default_values, ./message/codec, ./message/digest +import + ./message/message, + ./message/default_values, + ./message/codec, + ./message/digest, + ./message/path_counters -export message, default_values, codec, digest +export message, default_values, codec, digest, path_counters diff --git a/logos_delivery/waku/waku_core/message/codec.nim b/logos_delivery/waku/waku_core/message/codec.nim index 9b01cf667..f5cc65c1a 100644 --- a/logos_delivery/waku/waku_core/message/codec.nim +++ b/logos_delivery/waku/waku_core/message/codec.nim @@ -4,7 +4,7 @@ # - Proto definition: https://github.com/vacp2p/waku/blob/main/waku/message/v1/message.proto {.push raises: [].} -import ../../common/protobuf, ../topics, ../time, ./message +import ../../common/protobuf, ../topics, ../time, ./message, ./path_counters proc encode*(message: WakuMessage): ProtoBuffer = var buf = initProtoBuffer() @@ -21,6 +21,7 @@ proc encode*(message: WakuMessage): ProtoBuffer = buf proc decode*(T: type WakuMessage, buffer: seq[byte]): ProtobufResult[T] = + countDecode(buffer.len) var msg = WakuMessage() let pb = initProtoBuffer(buffer) diff --git a/logos_delivery/waku/waku_core/message/digest.nim b/logos_delivery/waku/waku_core/message/digest.nim index 3f82ce8f6..215445f68 100644 --- a/logos_delivery/waku/waku_core/message/digest.nim +++ b/logos_delivery/waku/waku_core/message/digest.nim @@ -1,7 +1,7 @@ {.push raises: [].} import std/sequtils, stew/[byteutils, endians2, arrayops], nimcrypto/sha2, results -import ../topics, ./message +import ../topics, ./message, ./path_counters ## 14/WAKU2-MESSAGE: Deterministic message hashing ## https://rfc.vac.dev/spec/14/#deterministic-message-hashing @@ -51,6 +51,7 @@ proc hexToHash*(hexString: string): Result[WakuMessageHash, string] = return ok(hash) proc computeMessageHash*(pubsubTopic: PubsubTopic, msg: WakuMessage): WakuMessageHash = + countHash(msg.payload.len + msg.meta.len) var ctx: sha256 ctx.init() defer: diff --git a/logos_delivery/waku/waku_core/message/path_counters.nim b/logos_delivery/waku/waku_core/message/path_counters.nim new file mode 100644 index 000000000..3fed1db6f --- /dev/null +++ b/logos_delivery/waku/waku_core/message/path_counters.nim @@ -0,0 +1,40 @@ +## Message-path instrumentation counters (compile-time gated). +## +## Enabled only with `-d:msgPathCounters`. When the define is absent every hook +## below expands to `discard` — no globals, no runtime cost, no behaviour change. +## These counters exist purely so the message-path benchmark +## (`apps/benchmarks/message_path_bench.nim`) can validate the decode/hash-per- +## message claims in `docs/analysis/async_copy_analysis.md`. +## +## Single-threaded chronos assumption: plain `int64`, no atomics. + +{.push raises: [].} + +when defined(msgPathCounters): + type MsgPathCounters* = object + decodeCalls*: int64 ## `WakuMessage.decode` invocations + hashCalls*: int64 ## `computeMessageHash` invocations + decodedBytes*: int64 ## bytes fed into decode (proto buffer length) + hashedBytes*: int64 ## bytes fed into the hash (payload + meta) + + var msgPathCounters* {.global.}: MsgPathCounters + + template countDecode*(bytes: int) = + msgPathCounters.decodeCalls += 1 + msgPathCounters.decodedBytes += int64(bytes) + + template countHash*(bytes: int) = + msgPathCounters.hashCalls += 1 + msgPathCounters.hashedBytes += int64(bytes) + + proc resetMsgPathCounters*() = + msgPathCounters = MsgPathCounters() + +else: + template countDecode*(bytes: int) = + discard + + template countHash*(bytes: int) = + discard + +{.pop.} diff --git a/logos_delivery/waku/waku_relay/protocol.nim b/logos_delivery/waku/waku_relay/protocol.nim index 409b02929..0d4bd65f5 100644 --- a/logos_delivery/waku/waku_relay/protocol.nim +++ b/logos_delivery/waku/waku_relay/protocol.nim @@ -633,6 +633,18 @@ proc subscribe*(w: WakuRelay, pubsubTopic: PubsubTopic, handler: WakuRelayHandle w.topicHealthDirty.incl(pubsubTopic) w.topicHealthUpdateEvent.fire() +when defined(msgPathCounters): + ## Benchmark-only accessors (see `apps/benchmarks/message_path_bench.nim`). + ## They expose the per-topic gossipsub artifacts so the micro benchmark can + ## replay the inbound validator + handler decode path without a live network. + ## Guarded by `-d:msgPathCounters` so production builds never see them and the + ## `topicValidator`/`topicHandlers` fields stay module-private. + proc benchTopicValidator*(w: WakuRelay, pubsubTopic: PubsubTopic): ValidatorHandler = + w.topicValidator.getOrDefault(pubsubTopic) + + proc benchTopicHandler*(w: WakuRelay, pubsubTopic: PubsubTopic): TopicHandler = + w.topicHandlers.getOrDefault(pubsubTopic) + proc unsubscribeAll*(w: WakuRelay, pubsubTopic: PubsubTopic) = ## Unsubscribe all handlers on this pubsub topic