Reduced the memory footprint of the js benchmark

Author: htemelski-oss

js/lib/metrics.js

@@ -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, 0);
-          const avgRtt = Number(sum) / tickRttValues.length;
-          avgRttMs = avgRtt;
+        if (rttAccumulator.count > 0) {
+          avgRttMs = rttAccumulator.getAverage();
           metrics.push(avgRttMs.toFixed(3));
+          // Reset accumulator after using the values
+          rttAccumulator.reset();
         } else {
           metrics.push('--');
         }
@@ -119,7 +147,7 @@ function writeFinalResults(
   totalSubscribed,
   messageRateTs,
   rttValues,
-  rttArchive,
+  rttHistogram,
   perSecondStats
 ) {
   const duration = (end - start);
@@ -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 ;
-    const p50 = histogram.getValueAtPercentile(50);
-    const p95 = histogram.getValueAtPercentile(95);
-    const p99 = histogram.getValueAtPercentile(99);
-    const p999 = histogram.getValueAtPercentile(99.9);
+    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/lib/redisManager.js

@@ -2,7 +2,7 @@ const Redis = require('ioredis');
 const clusterKeySlot = require('cluster-key-slot');
 const { publisherRoutine } = require('./publisher');
 const { subscriberRoutine } = require('./subscriber');
-const { updateCLI, writeFinalResults } = require('./metrics');
+const { updateCLI, writeFinalResults, createRttHistogram, RttAccumulator } = require('./metrics');
 const seedrandom = require('seedrandom');
 
 async function runBenchmark(argv) {
@@ -25,8 +25,11 @@ async function runBenchmark(argv) {
   const totalConnectsRef = { value: 0 };
   const isRunningRef = { value: true };
   const messageRateTs = [];
-  const rttValues = [];
-  const rttArchive = [];
+  
+  // Create efficient RTT tracking
+  const rttAccumulator = argv['measure-rtt-latency'] ? new RttAccumulator() : null;
+  // Create histogram for RTT recording
+  const rttHistogram = argv['measure-rtt-latency'] ? createRttHistogram() : null;
 
   const redisOptions = {
     host: argv.host,
@@ -166,8 +169,8 @@ async function runBenchmark(argv) {
             argv['measure-rtt-latency'],
             client,
             isRunningRef,
-            rttValues,
-            rttArchive,
+            rttAccumulator,
+            rttHistogram,
             totalMessagesRef,
             totalSubscribedRef,
             totalConnectsRef,
@@ -194,8 +197,8 @@ async function runBenchmark(argv) {
     totalSubscribedRef,
     totalPublishersRef,
     messageRateTs,
-    rttValues,
-    rttArchive,
+    rttAccumulator,
+    rttHistogram,
     () => {} // no-op, outputResults is handled after await
   );
 
@@ -211,8 +214,8 @@ async function runBenchmark(argv) {
     totalMessagesRef.value,
     totalSubscribedRef.value,
     messageRateTs,
-    rttValues,
-    rttArchive,
+    rttAccumulator,
+    rttHistogram,
     perSecondStats
   );
 

js/lib/subscriber.js

@@ -11,8 +11,8 @@ async function subscriberRoutine(
   measureRTT,
   client,
   isRunningRef,
-  rttValues,
-  rttArchive,
+  rttAccumulator,
+  rttHistogram,
   totalMessagesRef,
   totalSubscribedRef,
   totalConnectsRef,
@@ -71,13 +71,19 @@ async function subscriberRoutine(
     if (measureRTT) {
       try {
         const now = Date.now();
-        const timestamp = Number(message); // µs
+        const timestamp = Number(message); // Timestamp from publisher
         const rtt = now - timestamp;
-        if (rtt >= 0n) {
-          rttValues.push(rtt);
-          rttArchive.push(rtt);
+        if (rtt >= 0) {
+          // Add to accumulator for per-tick average calculation
+          if (rttAccumulator) {
+            rttAccumulator.add(rtt);
+          }
+          // Record directly to histogram for final stats
+          if (rttHistogram) {
+            rttHistogram.recordValue(rtt);
+          }
           if (verbose) {
-            console.log(`[${clientName}] RTT: ${rtt} µs`);
+            console.log(`[${clientName}] RTT: ${rtt} ms`);
           }
         } else {
           console.warn(`[${clientName}] Skipping negative RTT: now=${now}, ts=${timestamp}`);