v0.22.1 — reduce per-message allocations in send pumps and routing

Reuse batch arrays across send pump cycles. Consolidate
blocking-dequeue + non-blocking-sweep into Routing.dequeue_batch.
REP envelope as [conn, envelope] + concat instead of Hash + splat.
REQ transform_send uses dup.unshift instead of splat.
Bench harness accepts OMQ_BENCH_SIZES, OMQ_BENCH_TRANSPORTS,
OMQ_BENCH_PEERS env vars.

Author: paddor

CHANGELOG.md

@@ -1,5 +1,29 @@
 # Changelog
 
+## 0.22.1 — 2026-04-16
+
+### Changed
+
+- **Reuse batch arrays in send pumps.** All send pumps (RoundRobin,
+  Pair, ConnSendPump, FanOut, FanOut-conflate) now pre-allocate a
+  single batch array and clear it between cycles instead of
+  allocating a fresh `[msg]` per dequeue.
+
+- **`Routing.dequeue_batch`** consolidates the blocking-dequeue +
+  non-blocking-sweep pattern that was duplicated across four call
+  sites into one method. `dequeue_batch_capped` does the same for
+  the byte/message-capped RoundRobin variant.
+
+- **REP envelope stored as `[conn, envelope]`** instead of a Hash,
+  and reply assembly uses `<<` + `concat` instead of double splat.
+
+- **Heartbeat drops redundant `context: "".b`** — the default is
+  now `EMPTY_BINARY` in protocol-zmtp.
+
+- **Bench harness accepts `OMQ_BENCH_SIZES`, `OMQ_BENCH_TRANSPORTS`,
+  and `OMQ_BENCH_PEERS`** env vars to scope runs without editing
+  code.
+
 ## 0.22.0 — 2026-04-15
 
 ### Fixed

bench/bench_helper.rb

@@ -22,7 +22,7 @@
 
 module BenchHelper
   # ×4 geometric sweep from 128 B to 32 KiB.
-  SIZES = [128, 512, 2048, 8192, 32_768].freeze
+  SIZES = (ENV["OMQ_BENCH_SIZES"] || "128,512,2048,8192,32768").split(",").map(&:to_i).freeze
 
   # Each cell runs ROUNDS timed rounds and reports the fastest one.
   # Transient jitter (GC, scheduler preemption, YJIT tier-up, kernel
@@ -58,7 +58,7 @@ module BenchHelper
   # curve and blake3 are intentionally excluded from this suite.
   # CURVE regressions are caught by protocol-zmtp tests; BLAKE3-ZMTP
   # perf is tracked in the omq-rfc-blake3zmq repo.
-  TRANSPORTS = %w[inproc ipc tcp].freeze
+  TRANSPORTS = (ENV["OMQ_BENCH_TRANSPORTS"] || "inproc,ipc,tcp").split(",").freeze
 
   def run_id
     @run_id ||= ENV["OMQ_BENCH_RUN_ID"] || Time.now.strftime("%Y-%m-%dT%H:%M:%S")
@@ -70,6 +70,7 @@ def run_id
   RUN_TIMEOUT = Integer(ENV.fetch("OMQ_BENCH_TIMEOUT", 30))
 
   def run(label, dir:, peer_counts: [1, 3], &block)
+    peer_counts = ENV["OMQ_BENCH_PEERS"].split(",").map(&:to_i) if ENV["OMQ_BENCH_PEERS"]
     pattern = File.basename(dir)
     jit     = defined?(RubyVM::YJIT) && RubyVM::YJIT.enabled? ? "+YJIT" : "no JIT"
     puts "#{label} | OMQ #{OMQ::VERSION} | Ruby #{RUBY_VERSION} (#{jit}) | #{KERNEL}"

lib/omq/engine/heartbeat.rb

@@ -24,7 +24,7 @@ def self.start(parent, conn, options, tasks)
         tasks << parent.async(transient: true, annotation: "heartbeat") do
           loop do
             sleep interval
-            conn.send_command(Protocol::ZMTP::Codec::Command.ping(ttl: ttl, context: "".b))
+            conn.send_command(Protocol::ZMTP::Codec::Command.ping(ttl: ttl))
             if conn.heartbeat_expired?(timeout)
               conn.close
               break

lib/omq/routing.rb

@@ -57,7 +57,8 @@ def self.build_queue(hwm, on_mute)
 
 
     # Drains all available messages from +queue+ into +batch+ without
-    # blocking. Call after the initial blocking dequeue.
+    # Blocks for the first message, then sweeps all immediately
+    # available messages into +batch+ without blocking.
     #
     # No cap is needed: IO::Stream auto-flushes at 64 KB, so the
     # write buffer hits the wire naturally under sustained load.
@@ -67,7 +68,9 @@ def self.build_queue(hwm, on_mute)
     # @param batch [Array]
     # @return [void]
     #
-    def self.drain_send_queue(queue, batch)
+    def self.dequeue_batch(queue, batch = [])
+      batch << queue.dequeue
+
       loop do
         msg = queue.dequeue(timeout: 0) or break
         batch << msg

lib/omq/routing/conn_send_pump.rb

@@ -18,9 +18,10 @@ module ConnSendPump
       #
       def self.start(engine, conn, q, tasks)
         task = engine.spawn_conn_pump_task(conn, annotation: "send pump") do
+          batch = []
+
           loop do
-            batch = [q.dequeue]
-            Routing.drain_send_queue(q, batch)
+            Routing.dequeue_batch(q, batch)
 
             if batch.size == 1
               conn.write_message batch.first
@@ -30,7 +31,11 @@ def self.start(engine, conn, q, tasks)
 
             conn.flush
 
-            batch.each { |parts| engine.emit_verbose_msg_sent(conn, parts) }
+            batch.each do |parts|
+              engine.emit_verbose_msg_sent(conn, parts)
+            end
+
+            batch.clear
           end
         end
 

lib/omq/routing/fan_out.rb

@@ -181,14 +181,16 @@ def start_conn_send_pump(conn, q)
       #
       def start_conn_send_pump_normal(conn, q, use_wire)
         @engine.spawn_conn_pump_task(conn, annotation: "send pump") do
+          batch = []
+
           loop do
-            batch = [q.dequeue]
-            Routing.drain_send_queue(q, batch)
+            Routing.dequeue_batch(q, batch)
 
             if write_matching_batch(conn, batch, use_wire)
               conn.flush
               batch.each { |parts| @engine.emit_verbose_msg_sent(conn, parts) }
             end
+            batch.clear
           end
         end
       end
@@ -227,14 +229,17 @@ def write_matching_batch(conn, batch, use_wire)
       #
       def start_conn_send_pump_conflate(conn, q)
         @engine.spawn_conn_pump_task(conn, annotation: "send pump") do
+          batch = []
+
           loop do
-            batch = [q.dequeue]
-            Routing.drain_send_queue(q, batch)
+            Routing.dequeue_batch(q, batch)
 
             # Keep only the latest message that matches the subscription.
             latest = batch.reverse.find do |parts|
               subscribed?(conn, parts.first || EMPTY_BINARY)
             end
+
+            batch.clear
             next unless latest
 
             conn.write_message(latest)

lib/omq/routing/pair.rb

@@ -106,9 +106,10 @@ def send_queues_drained?
 
       def start_send_pump(conn)
         @send_pump = @engine.spawn_conn_pump_task(conn, annotation: "send pump") do
+          batch = []
+
           loop do
-            batch = [@send_queue.dequeue]
-            Routing.drain_send_queue(@send_queue, batch)
+            Routing.dequeue_batch(@send_queue, batch)
 
             if batch.size == 1
               conn.write_message(batch.first)
@@ -120,6 +121,7 @@ def start_send_pump(conn)
             batch.each do |parts|
               @engine.emit_verbose_msg_sent(conn, parts)
             end
+            batch.clear
           end
         end
 

lib/omq/routing/rep.rb

@@ -55,7 +55,7 @@ def connection_added(connection)
           envelope  = msg[0, delimiter]
           body      = msg[(delimiter + 1)..] || []
 
-          @pending_replies << { conn: connection, envelope: envelope }
+          @pending_replies << [connection, envelope]
           body
         end
         @tasks << task if task
@@ -69,7 +69,7 @@ def connection_added(connection)
       # @param connection [Connection]
       #
       def connection_removed(connection)
-        @pending_replies.reject! { |r| r[:conn] == connection }
+        @pending_replies.reject! { |r| r[0] == connection }
         @conn_queues.delete(connection)
         @conn_send_tasks.delete(connection)&.stop
       end
@@ -83,8 +83,12 @@ def connection_removed(connection)
       def enqueue(parts)
         reply_info = @pending_replies.shift
         return unless reply_info
-        conn = reply_info[:conn]
-        @conn_queues[conn]&.enqueue([*reply_info[:envelope], EMPTY_FRAME, *parts])
+
+        conn, envelope = reply_info
+        msg = envelope
+        msg << EMPTY_FRAME
+        msg.concat(parts)
+        @conn_queues[conn]&.enqueue(msg)
       end
 
 

lib/omq/routing/req.rb

@@ -93,7 +93,7 @@ def stop
       # REQ prepends empty delimiter frame on the wire.
       #
       def transform_send(parts)
-        [EMPTY_BINARY, *parts]
+        parts.dup.unshift(EMPTY_BINARY)
       end
 
     end

lib/omq/routing/round_robin.rb

@@ -133,9 +133,10 @@ def transform_send(parts)
       #
       def start_conn_send_pump(conn)
         task = @engine.spawn_conn_pump_task(conn, annotation: "send pump") do
+          batch = []
+
           loop do
-            batch = [@send_queue.dequeue]
-            drain_send_queue_capped(batch)
+            dequeue_batch_capped(batch)
             @in_flight += batch.size
 
             begin
@@ -147,6 +148,7 @@ def start_conn_send_pump(conn)
             batch.each do |parts|
               @engine.emit_verbose_msg_sent(conn, parts)
             end
+            batch.clear
 
             Async::Task.current.yield
           end
@@ -157,7 +159,8 @@ def start_conn_send_pump(conn)
       end
 
 
-      def drain_send_queue_capped(batch)
+      def dequeue_batch_capped(batch = [])
+        batch << @send_queue.dequeue
         bytes = batch_bytes(batch.first)
 
         while batch.size < BATCH_MSG_CAP && bytes < BATCH_BYTE_CAP