Track RTT in nanos.

.gitignore

@@ -87,6 +87,8 @@ Thumbs.db
 # Coverage Results #
 ####################
 coverage.txt
+.coverdata/
+pubsub-sub-bench-test
 
 # Profiler Results #
 ####################

Makefile

@@ -33,6 +33,11 @@ build-race:
 	$(GOBUILDRACE) \
         -ldflags=$(LDFLAGS) .
 
+build-cover:
+	@echo "Building binary with coverage instrumentation..."
+	$(GOBUILD) -cover \
+        -ldflags=$(LDFLAGS) .
+
 checkfmt:
 	@echo 'Checking gofmt';\
  	bash -c "diff -u <(echo -n) <(go fmt .)";\
@@ -52,9 +57,21 @@ fmt:
 get:
 	$(GOGET) -t -v ./...
 
-test: get
+test: get build-cover
 	$(GOFMT) ./...
-	$(GOTEST) -race -covermode=atomic ./...
+	@rm -rf .coverdata
+	@mkdir -p .coverdata
+	$(GOTEST) -v -race -covermode=atomic ./...
 
-coverage: get test
-	$(GOTEST) -race -coverprofile=coverage.txt -covermode=atomic .
+coverage: get build-cover
+	$(GOFMT) ./...
+	@rm -rf .coverdata
+	@mkdir -p .coverdata
+	$(GOTEST) -v -race -covermode=atomic .
+	@if [ -d .coverdata ] && [ -n "$$(ls -A .coverdata 2>/dev/null)" ]; then \
+		echo "Converting coverage data..."; \
+		go tool covdata textfmt -i=.coverdata -o coverage.txt; \
+	else \
+		echo "No coverage data found, creating empty coverage file"; \
+		touch coverage.txt; \
+	fi

README.md

@@ -97,7 +97,7 @@ Usage of ./pubsub-sub-bench:
   -cpuprofile string
     	write cpu profile to file
   -data-size int
-    	Payload size in bytes. In RTT mode, timestamp (13 bytes) + space + padding to reach this size. (default 128)
+    	Payload size in bytes. In RTT mode, timestamp (19 bytes) + space + padding to reach this size. (default 128)
   -host string
     	redis host. (default "127.0.0.1")
   -json-out-file string

go.mod

@@ -1,16 +1,16 @@
 module github.com/RedisLabs/pubsub-sub-bench
 
-go 1.23.0
+go 1.24.0
 
 toolchain go1.24.1
 
 require (
 	github.com/HdrHistogram/hdrhistogram-go v1.1.2
-	github.com/redis/go-redis/v9 v9.0.5
-	golang.org/x/time v0.11.0
+	github.com/redis/go-redis/v9 v9.16.0
+	golang.org/x/time v0.14.0
 )
 
 require (
-	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 )

go.sum

@@ -3,12 +3,12 @@ github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym
 github.com/HdrHistogram/hdrhistogram-go v1.1.2 h1:5IcZpTvzydCQeHzK4Ef/D5rrSqwxob0t8PQPMybUNFM=
 github.com/HdrHistogram/hdrhistogram-go v1.1.2/go.mod h1:yDgFjdqOqDEKOvasDdhWNXYg9BVp4O+o5f6V/ehm6Oo=
 github.com/ajstarks/svgo v0.0.0-20180226025133-644b8db467af/go.mod h1:K08gAheRH3/J6wwsYMMT4xOr94bZjxIelGM0+d/wbFw=
-github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao=
-github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w=
-github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y=
-github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
-github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
-github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -27,8 +27,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o=
-github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk=
+github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERSEP4=
+github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@@ -52,8 +52,8 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0=
-golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
+golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
+golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190206041539-40960b6deb8e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=

js/ioredis/lib/config.js

@@ -29,13 +29,14 @@ function parseArgs() {
     .option('test-time', { default: 0 })
     .option('rand-seed', { default: 12345 })
     .option('subscriber-prefix', { default: 'channel-' })
-    .option('oss-cluster-api-distribute-subscribers', { default: false })
+    .option('oss-cluster-api-distribute-subscribers', { type: 'boolean', default: false })
     .option('slot-refresh-interval', { default: -1 })
-    .option('print-messages', { default: false })
-    .option('verbose', { default: false })
-    .option('measure-rtt-latency', { default: false })
+    .option('print-messages', { type: 'boolean', default: false })
+    .option('verbose', { type: 'boolean', default: false })
+    .option('measure-rtt-latency', { type: 'boolean', default: false })
     .option('redis-timeout', { default: 120000 })
     .option('pool-size', { default: 0 })
+    .option('rps', { description: 'Target requests per second (RPS) for publishers', default: 0 })
     .help().argv;
 }
 

js/ioredis/lib/publisher.js

@@ -8,7 +8,8 @@ async function publisherRoutine(
   client,
   isRunningRef,
   totalMessagesRef,
-  rateLimiter
+  rateLimiter,
+  skipDuplicate = false
 ) {
   if (verbose) {
     console.log(
@@ -18,31 +19,88 @@ async function publisherRoutine(
     );
   }
 
-  const payload = !measureRTT ? 'A'.repeat(dataSize) : '';
-  const duplicatedClient = client.duplicate(); // Create a duplicated connection for this publisher
+  // Pre-allocate payload once per publisher to avoid repeated allocations
+  // Timestamp format: 13 bytes for milliseconds (e.g., "1730745600000")
+  // Format: "<timestamp> <padding>" to reach dataSize
+  const timestampSize = 13; // Date.now() returns milliseconds (13 digits)
+  let paddingPayload = '';
+
+  if (measureRTT && dataSize > timestampSize + 1) {
+    // +1 for space separator
+    const paddingSize = dataSize - timestampSize - 1;
+    paddingPayload = 'A'.repeat(paddingSize);
+  } else if (!measureRTT) {
+    paddingPayload = 'A'.repeat(dataSize);
+  }
+
+  // For cluster mode, use the client directly without duplication
+  // The cluster client handles routing and can be shared across publishers
+  const duplicatedClient = skipDuplicate ? client : client.duplicate();
+
+  // Wait for client to be ready
+  if (!skipDuplicate) {
+    await new Promise((resolve, reject) => {
+      if (duplicatedClient.status === 'ready') {
+        resolve();
+      } else {
+        duplicatedClient.once('ready', resolve);
+        duplicatedClient.once('error', reject);
+      }
+    });
+  }
 
   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 (measureRTT) {
+      // RTT mode: generate timestamp for each message with padding to reach dataSize
+      while (isRunningRef.value) {
+        for (const channel of channels) {
+          try {
+            // Apply rate limiting if configured
+            if (rateLimiter) {
+              await rateLimiter.removeTokens(1);
+            }
+
+            let msg;
+            if (dataSize > timestampSize + 1) {
+              // Format: "<timestamp> <padding>"
+              msg = Date.now().toString() + ' ' + paddingPayload;
+            } else {
+              // Just timestamp if dataSize is too small
+              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);
+            process.exit(1);
           }
+        }
+      }
+    } else {
+      // Fixed payload mode: reuse pre-allocated payload
+      while (isRunningRef.value) {
+        for (const channel of channels) {
+          try {
+            // Apply rate limiting if configured
+            if (rateLimiter) {
+              await rateLimiter.removeTokens(1);
+            }
 
-          if (mode === 'spublish') {
-            await duplicatedClient.spublish(channel, msg);
-          } else {
-            await duplicatedClient.publish(channel, msg);
+            if (mode === 'spublish') {
+              await duplicatedClient.spublish(channel, paddingPayload);
+            } else {
+              await duplicatedClient.publish(channel, paddingPayload);
+            }
+            totalMessagesRef.value++;
+          } catch (err) {
+            console.error(`Error publishing to channel ${channel}:`, err);
+            process.exit(1);
           }
-          totalMessagesRef.value++;
-        } catch (err) {
-          console.error(`Error publishing to channel ${channel}:`, err);
         }
       }
     }