trying to debug node-redis bench

Author: htemelski-oss

.gitignore

@@ -92,3 +92,4 @@ coverage.txt
 # Profiler Results #
 ####################
 *.pprof
+/js/node-redis/node_modules/

js/node-redis/.gitignore

@@ -1,14 +1,15 @@
 #!/usr/bin/env node
-// filepath: /Users/hristo.temelski/code/etc/pubsub-sub-bench/js/node-redis/bin/pubsub-sub-bench.js
 
 const { parseArgs } = require('../lib/config');
 const { runBenchmark } = require('../lib/redisManager');
 
-// Parse command line arguments
-const argv = parseArgs();
+(async () => {
+  const argv = parseArgs();
 
-// Run the benchmark
-runBenchmark(argv).catch(err => {
-  console.error('Benchmark execution error:', err);
-  process.exit(1);
-});
+  try {
+    await runBenchmark(argv);
+  } catch (err) {
+    console.error('Error in main execution:', err);
+    process.exit(1);
+  }
+})();

js/node-redis/README.md

@@ -1,4 +1,3 @@
-// filepath: /Users/hristo.temelski/code/etc/pubsub-sub-bench/js/node-redis/lib/config.js
 const yargs = require('yargs');
 
 function parseArgs() {
@@ -40,4 +39,4 @@ function parseArgs() {
     .help().argv;
 }
 
-module.exports = { parseArgs };
+module.exports = { parseArgs };

js/node-redis/bin/pubsub-sub-bench.js

@@ -212,4 +212,4 @@ module.exports = {
   writeFinalResults,
   createRttHistogram,
   RttAccumulator
-};
+};

js/node-redis/lib/config.js

@@ -1,17 +1,13 @@
-// filepath: /Users/hristo.temelski/code/etc/pubsub-sub-bench/js/node-redis/lib/publisher.js
-const { createClient } = require('redis');
-
 async function publisherRoutine(
   clientName,
   channels,
   mode,
   measureRTT,
   verbose,
   dataSize,
-  redisOptions,
+  client,
   isRunningRef,
-  totalMessagesRef,
-  rateLimiter
+  totalMessagesRef
 ) {
   if (verbose) {
     console.log(
@@ -22,27 +18,11 @@ async function publisherRoutine(
   }
 
   const payload = !measureRTT ? 'A'.repeat(dataSize) : '';
-  
-  // Create a new Redis client
-  const client = createClient(redisOptions);
-  
-  // Set up error handling
-  client.on('error', (err) => {
-    console.error(`[${clientName}] Redis error: ${err.message}`);
-  });
-  
+
   try {
-    // Connect to Redis
-    await client.connect();
-    
     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();
@@ -55,25 +35,15 @@ async function publisherRoutine(
           }
           totalMessagesRef.value++;
         } catch (err) {
-          console.error(`[${clientName}] Error publishing to channel ${channel}:`, err);
+          console.error(`Error publishing to channel ${channel}:`, err);
         }
       }
     }
-  } catch (err) {
-    console.error(`[${clientName}] Redis connection error:`, err);
   } finally {
-    // Clean shutdown - disconnect the client
+    // Clean shutdown - client is managed by redisManager
     if (verbose) {
       console.log(`Publisher ${clientName} shutting down...`);
     }
-    try {
-      await client.quit();
-      if (verbose) {
-        console.log(`Publisher ${clientName} disconnected successfully`);
-      }
-    } catch (err) {
-      console.error(`Error disconnecting publisher ${clientName}:`, err);
-    }
   }
 }
 

js/node-redis/lib/metrics.js

@@ -30,46 +30,41 @@ async function runBenchmark(argv) {
   // Create histogram for RTT recording
   const rttHistogram = argv['measure-rtt-latency'] ? createRttHistogram() : null;
 
+  const redisOptions = {
+    socket: {
+      host: argv.host,
+      port: argv.port
+    },
+    username: argv.user || undefined,
+    password: argv.a || undefined,
+    commandTimeout: argv['redis-timeout']
+  };
+
   let client;
   let nodeAddresses = [];
 
-  if (argv['oss-cluster-api-distribute-subscribers']) {
-    console.log('Using Redis Cluster mode');
+  console.log(`Using ${argv['slot-refresh-interval']} slot-refresh-interval`);
+  console.log(`Using ${argv['redis-timeout']} redis-timeout`);
+
+  if (argv['oss-cluster-api-distribute-subscribers'] === "true") {
+    // Use createCluster for Redis Cluster mode
     client = createCluster({
-      rootNodes: [{
-        socket: {
-          host: argv.host,
-          port: argv.port
-        },
-        password: argv.a || undefined,
-        username: argv.user || undefined
+      rootNodes: [{ 
+        url: `redis://${argv.host}:${argv.port}`
       }],
+      useReplicas: false,
       defaults: {
-        socket: {
-          connectTimeout: argv['redis-timeout'],
-          reconnectStrategy: false // disable auto-reconnect
-        }
+        username: argv.user || undefined,
+        password: argv.a || undefined,
+        commandTimeout: argv['redis-timeout']
       }
     });
-    
+
     await client.connect();
-    nodeAddresses = [`${argv.host}:${argv.port}`];
-    console.log('Cluster mode - connecting through cluster client');
   } else {
-    // Single node mode
-    client = createClient({
-      socket: {
-        host: argv.host,
-        port: argv.port,
-        connectTimeout: argv['redis-timeout'],
-        reconnectStrategy: false // disable auto-reconnect
-      },
-      password: argv.a || undefined,
-      username: argv.user || undefined
-    });
-    
+    // Standalone mode
+    client = createClient(redisOptions);
     await client.connect();
-    nodeAddresses = [`${argv.host}:${argv.port}`];
     console.log('Standalone mode - using single Redis instance');
   }
 
@@ -80,31 +75,9 @@ async function runBenchmark(argv) {
   console.log(`Will use a subscriber prefix of: ${argv['subscriber-prefix']}<channel id>`);
   console.log(`Total channels: ${totalChannels}`);
   console.log('Final setup used for benchmark:');
-  nodeAddresses.forEach((addr, i) => {
-    console.log(`Node #${i}: Address: ${addr}`);
-  });
 
   const promises = [];
 
-  function randomInt(min, max) {
-    if (min === max) return min;
-    return Math.floor(Math.random() * (max - min + 1)) + min;
-  }
-
-  function pickChannelCount(argv) {
-    return randomInt(
-      argv['min-number-channels-per-subscriber'],
-      argv['max-number-channels-per-subscriber']
-    );
-  }
-
-  function randomChannel(argv) {
-    return (
-      Math.floor(Math.random() * (argv['channel-maximum'] - argv['channel-minimum'] + 1)) +
-      argv['channel-minimum']
-    );
-  }
-
   if (argv.mode.includes('publish')) {
     // Run publishers
     totalPublishersRef.value = argv.clients;
@@ -159,7 +132,6 @@ async function runBenchmark(argv) {
         }
 
         const subscriberName = `subscriber#${clientId}`;
-
         const reconnectInterval = randomInt(
           argv['min-reconnect-interval'],
           argv['max-reconnect-interval']
@@ -213,15 +185,14 @@ async function runBenchmark(argv) {
       totalPublishersRef,
       messageRateTs,
       rttAccumulator,
-      rttHistogram,
-      () => {} // no-op, outputResults is handled after await
+      rttHistogram
     );
 
     // Wait for all routines to finish
     console.log('Waiting for all clients to shut down cleanly...');
     await Promise.all(promises);
 
-    // Output final results
+    // THEN output final results
     writeFinalResults(
       startTime,
       now,
@@ -234,19 +205,38 @@ async function runBenchmark(argv) {
       rttHistogram,
       perSecondStats
     );
-  } catch (err) {
-    console.error('Benchmark error:', err);
-  }
-
-  // Clean up and disconnect the main client
-  try {
-    await client.quit();
-  } catch (err) {
-    console.error('Error disconnecting main client:', err);
+  } finally {
+    // Clean shutdown of Redis connection
+    console.log('Shutting down Redis connection...');
+    try {
+      await client.quit();
+      console.log('Redis connection closed successfully');
+    } catch (err) {
+      console.error('Error disconnecting Redis client:', err);
+    }
   }
 
-  // Clean exit
+  // cleanly exit the process once done
   process.exit(0);
 }
 
+function randomInt(min, max) {
+  if (min === max) return min;
+  return Math.floor(Math.random() * (max - min + 1)) + min;
+}
+
+function pickChannelCount(argv) {
+  return randomInt(
+    argv['min-number-channels-per-subscriber'],
+    argv['max-number-channels-per-subscriber']
+  );
+}
+
+function randomChannel(argv) {
+  return (
+    Math.floor(Math.random() * (argv['channel-maximum'] - argv['channel-minimum'] + 1)) +
+    argv['channel-minimum']
+  );
+}
+
 module.exports = { runBenchmark };