Fixed time conversion, reduced memory footprint, enabled s/publish, added node-redis benchmark (#28)

* Fix time conversions on the subscriber side

* minor fix

* Use only milliseconds

* Reduced the memory footprint of the js benchmark

* Enable s/publish

* Added node-redis benchmark implementation

Author: htemelski

.gitignore

@@ -83,10 +83,7 @@ pubsub-sub-bench
 ehthumbs.db
 Thumbs.db
 
-
-# Json Results #
-################
-*.json
+package-lock.json
 
 # Coverage Results #
 ####################

js/.gitignore renamed to js/ioredis/.gitignore

@@ -10,5 +10,5 @@ Supports both **standalone** and **Redis OSS Cluster** modes, with support for `
 ## 📦 Installation
 
 ```bash
-cd pubsub-sub-bench/js
+cd pubsub-sub-bench/js/ioredis
 npm install

js/.prettierrc renamed to js/ioredis/.prettierrc

@@ -1,6 +1,35 @@
 const fs = require('fs');
 const hdr = require('hdr-histogram-js');
 
+// Simple accumulator for RTT stats per tick
+class RttAccumulator {
+  constructor() {
+    this.reset();
+  }
+
+  reset() {
+    this.sum = 0;
+    this.count = 0;
+  }
+
+  add(value) {
+    this.sum += value;
+    this.count++;
+  }
+
+  getAverage() {
+    return this.count > 0 ? this.sum / this.count : null;
+  }
+}
+
+function createRttHistogram() {
+  return hdr.build({
+    lowestDiscernibleValue: 1,
+    highestTrackableValue: 10_000_000,
+    numberOfSignificantValueDigits: 3
+  });
+}
+
 function formatRow(row) {
   const widths = [6, 15, 14, 14, 22, 14];
   return row.map((val, i) => String(val).padEnd(widths[i] || 10)).join('');
@@ -18,8 +47,8 @@ function updateCLI(
   totalSubscribedRef,
   totalPublishersRef,
   messageRateTs,
-  rttValues,
-  rttArchive
+  rttAccumulator,
+  rttHistogram
 ) {
   return new Promise((resolve) => {
     let prevTime = Date.now();
@@ -66,12 +95,11 @@ function updateCLI(
       let avgRttMs = null;
 
       if (measureRTT) {
-        const tickRttValues = rttValues.splice(0);
-        if (tickRttValues.length > 0) {
-          const sum = tickRttValues.reduce((a, b) => a + b, 0n);
-          const avgRtt = Number(sum) / tickRttValues.length;
-          avgRttMs = avgRtt / 1000;
+        if (rttAccumulator.count > 0) {
+          avgRttMs = rttAccumulator.getAverage();
           metrics.push(avgRttMs.toFixed(3));
+          // Reset accumulator after using the values
+          rttAccumulator.reset();
         } else {
           metrics.push('--');
         }
@@ -119,10 +147,10 @@ function writeFinalResults(
   totalSubscribed,
   messageRateTs,
   rttValues,
-  rttArchive,
+  rttHistogram,
   perSecondStats
 ) {
-  const duration = (end - start) / 1000;
+  const duration = (end - start)/1000;
   const messageRate = totalMessages / duration;
 
   console.log('#################################################');
@@ -131,7 +159,7 @@ function writeFinalResults(
   console.log(`Message Rate: ${messageRate.toFixed(6)} msg/sec`);
 
   const result = {
-    StartTime: Math.floor(start / 1000),
+    StartTime: Math.floor(start),
     Duration: duration,
     Mode: mode,
     MessageRate: messageRate,
@@ -148,38 +176,27 @@ function writeFinalResults(
   };
 
   if (argv['measure-rtt-latency'] && !mode.includes('publish')) {
-    const histogram = hdr.build({
-      lowestDiscernibleValue: 1,
-      highestTrackableValue: 10_000_000,
-      numberOfSignificantValueDigits: 3
-    });
-
-    rttArchive.forEach((rtt) => {
-      const val = Number(rtt);
-      if (val >= 0) histogram.recordValue(val);
-    });
-
-    const avgRtt = histogram.mean / 1000;
-    const p50 = histogram.getValueAtPercentile(50) / 1000;
-    const p95 = histogram.getValueAtPercentile(95) / 1000;
-    const p99 = histogram.getValueAtPercentile(99) / 1000;
-    const p999 = histogram.getValueAtPercentile(99.9) / 1000;
+    const avgRtt = rttHistogram.mean;
+    const p50 = rttHistogram.getValueAtPercentile(50);
+    const p95 = rttHistogram.getValueAtPercentile(95);
+    const p99 = rttHistogram.getValueAtPercentile(99);
+    const p999 = rttHistogram.getValueAtPercentile(99.9);
 
     result.RTTSummary = {
       AvgMs: Number(avgRtt.toFixed(3)),
       P50Ms: Number(p50.toFixed(3)),
       P95Ms: Number(p95.toFixed(3)),
       P99Ms: Number(p99.toFixed(3)),
       P999Ms: Number(p999.toFixed(3)),
-      totalCount: histogram.totalCount
+      totalCount: rttHistogram.totalCount
     };
 
-    console.log(`Avg RTT       ${avgRtt.toFixed(3)} ms`);
-    console.log(`P50 RTT       ${p50.toFixed(3)} ms`);
-    console.log(`P95 RTT       ${p95.toFixed(3)} ms`);
-    console.log(`P99 RTT       ${p99.toFixed(3)} ms`);
-    console.log(`P999 RTT      ${p999.toFixed(3)} ms`);
-    console.log(`Total Messages tracked latency      ${histogram.totalCount} messages`);
+    console.log(`Avg  RTT       ${avgRtt.toFixed(3)} ms`);
+    console.log(`P50  RTT       ${p50.toFixed(3)} ms`);
+    console.log(`P95  RTT       ${p95.toFixed(3)} ms`);
+    console.log(`P99  RTT       ${p99.toFixed(3)} ms`);
+    console.log(`P999 RTT       ${p999.toFixed(3)} ms`);
+    console.log(`Total Messages tracked latency      ${rttHistogram.totalCount} messages`);
   }
 
   console.log('#################################################');
@@ -192,5 +209,7 @@ function writeFinalResults(
 
 module.exports = {
   updateCLI,
-  writeFinalResults
+  writeFinalResults,
+  createRttHistogram,
+  RttAccumulator
 };

js/README.md renamed to js/ioredis/README.md

@@ -0,0 +1,65 @@
+async function publisherRoutine(
+  clientName,
+  channels,
+  mode,
+  measureRTT,
+  verbose,
+  dataSize,
+  client,
+  isRunningRef,
+  totalMessagesRef,
+  rateLimiter
+) {
+  if (verbose) {
+    console.log(
+      `Publisher ${clientName} started. Mode: ${mode} | Channels: ${channels.length} | Payload: ${
+        measureRTT ? 'RTT timestamp' : `fixed size ${dataSize} bytes`
+      }`
+    );
+  }
+
+  const payload = !measureRTT ? 'A'.repeat(dataSize) : '';
+  const duplicatedClient = client.duplicate(); // Create a duplicated connection for this publisher
+
+  try {
+    while (isRunningRef.value) {
+      for (const channel of channels) {
+        try {
+          // Apply rate limiting if configured
+          if (rateLimiter) {
+            await rateLimiter.removeTokens(1);
+          }
+          
+          let msg = payload;
+          if (measureRTT) {
+            msg = Date.now().toString();
+          }
+
+          if (mode === 'spublish') {
+            await duplicatedClient.spublish(channel, msg);
+          } else {
+            await duplicatedClient.publish(channel, msg);
+          }
+          totalMessagesRef.value++;
+        } catch (err) {
+          console.error(`Error publishing to channel ${channel}:`, err);
+        }
+      }
+    }
+  } finally {
+    // Clean shutdown - disconnect the client
+    if (verbose) {
+      console.log(`Publisher ${clientName} shutting down...`);
+    }
+    try {
+      duplicatedClient.disconnect();
+      if (verbose) {
+        console.log(`Publisher ${clientName} disconnected successfully`);
+      }
+    } catch (err) {
+      console.error(`Error disconnecting publisher ${clientName}:`, err);
+    }
+  }
+}
+
+module.exports = { publisherRoutine };