v0.17.0

- drop Readable prefetch buffer; #receive dequeues single message
- tiny routing.drain_send_queue cleanup

Author: paddor

CHANGELOG.md

@@ -1,6 +1,15 @@
 # Changelog
 
-## Unreleased
+## 0.17.0 — 2026-04-10
+
+### Changed
+
+- **`Readable#receive` no longer prefetches a batch.** Each `#receive`
+  call dequeues exactly one message from the engine recv queue. The
+  per-socket prefetch buffer (`@recv_buffer` + `@recv_mutex`) and
+  `dequeue_recv_batch` are gone, along with `Readable::RECV_BATCH_SIZE`.
+  Simpler code; ~5–10% inproc microbench regression accepted (tcp/ipc
+  unchanged — wire I/O dominates dispatch overhead).
 
 ### Added
 

lib/omq/engine.rb

@@ -235,30 +235,6 @@ def dequeue_recv
     end
 
 
-    # Dequeues up to +max+ messages or +max_bytes+ total. Blocks
-    # on the first, then drains non-blocking.
-    #
-    # @param max [Integer] message count limit
-    # @param max_bytes [Integer] byte size limit
-    # @return [Array<Array<String>>]
-    #
-    def dequeue_recv_batch(max, max_bytes: 1 << 20)
-      raise @fatal_error if @fatal_error
-      queue = routing.recv_queue
-      msg   = queue.dequeue
-      raise @fatal_error if msg.nil? && @fatal_error
-      batch = [msg]
-      bytes = msg.sum(&:bytesize)
-      while batch.size < max && bytes < max_bytes
-        msg = queue.dequeue(timeout: 0)
-        break unless msg
-        batch << msg
-        bytes += msg.sum(&:bytesize)
-      end
-      batch
-    end
-
-
     # Pushes a nil sentinel into the recv queue, unblocking a
     # pending {#dequeue_recv} with a nil return value.
     #

lib/omq/queue_interface.rb

@@ -16,13 +16,7 @@ module QueueReadable
     # @raise [IO::TimeoutError] if timeout exceeded
     #
     def dequeue(timeout: @options.read_timeout)
-      msg = @recv_mutex.synchronize { @recv_buffer.shift }
-      return msg if msg
-
-      batch = Reactor.run { with_timeout(timeout) { @engine.dequeue_recv_batch(Readable::RECV_BATCH_SIZE) } }
-      msg = batch.shift
-      @recv_mutex.synchronize { @recv_buffer.concat(batch) } unless batch.empty?
-      msg
+      Reactor.run { with_timeout(timeout) { @engine.dequeue_recv } }
     end
 
     alias_method :pop, :dequeue

lib/omq/readable.rb

@@ -8,20 +8,13 @@ module OMQ
   module Readable
     include QueueReadable
 
-    # Maximum messages to prefetch from the recv queue per drain.
-    RECV_BATCH_SIZE = 64
-
-
-    # Receives the next message. Returns from a local prefetch
-    # buffer when available, otherwise drains up to
-    # {RECV_BATCH_SIZE} messages from the recv queue in one
-    # synchronized dequeue.
+    # Receives the next message directly from the engine recv queue.
     #
     # @return [Array<String>] message parts
     # @raise [IO::TimeoutError] if read_timeout exceeded
     #
     def receive
-      @recv_mutex.synchronize { @recv_buffer.shift } || fill_recv_buffer
+      Reactor.run { with_timeout(@options.read_timeout) { @engine.dequeue_recv } }
     end
 
 
@@ -33,14 +26,5 @@ def receive
     def wait_readable(timeout = @options.read_timeout)
       true
     end
-
-    private
-
-    def fill_recv_buffer
-      batch = Reactor.run { with_timeout(@options.read_timeout) { @engine.dequeue_recv_batch(RECV_BATCH_SIZE) } }
-      msg = batch.shift
-      @recv_mutex.synchronize { @recv_buffer.concat(batch) } unless batch.empty?
-      msg
-    end
   end
 end

lib/omq/routing.rb

@@ -70,8 +70,7 @@ def self.build_queue(hwm, on_mute)
     #
     def self.drain_send_queue(queue, batch)
       loop do
-        msg = queue.dequeue(timeout: 0)
-        break unless msg
+        msg = queue.dequeue(timeout: 0) or break
         batch << msg
       end
     end

lib/omq/socket.rb

@@ -316,9 +316,7 @@ def _init_engine(socket_type, linger:, send_hwm: nil, recv_hwm: nil,
       @options.recv_timeout   = recv_timeout if recv_timeout
       @options.conflate       = conflate
       @options.on_mute        = on_mute      if on_mute
-      @recv_buffer = []
-      @recv_mutex  = Mutex.new
-      @engine      = case backend
+      @engine                 = case backend
                      when nil, :ruby
                        Engine.new(socket_type, @options)
                      when :ffi

lib/omq/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module OMQ
-  VERSION = "0.16.2"
+  VERSION = "0.17.0"
 end

test/omq/push_pull_test.rb

@@ -353,37 +353,6 @@
     end
   end
 
-  it "prefetch batch is capped by byte size" do
-    Async do
-      pull = OMQ::PULL.bind("inproc://pushpull-prefetch-bytes")
-      push = OMQ::PUSH.connect("inproc://pushpull-prefetch-bytes")
-
-      # Each message is ~512 KB, so 1 MB limit should cap at 2 messages
-      big = "x" * (512 * 1024)
-      5.times { push.send(big) }
-
-      # Let messages arrive
-      sleep 0.01
-
-      # Peek at internal prefetch: first receive triggers a batch drain
-      msg1 = pull.receive
-      assert_equal big, msg1.first
-
-      # Check recv_buffer size — should have at most 1 more (batch was ~2)
-      buffer_size = pull.instance_variable_get(:@recv_mutex).synchronize do
-        pull.instance_variable_get(:@recv_buffer).size
-      end
-      assert_operator buffer_size, :<=, 2, "prefetch should be capped by 1 MB byte limit"
-
-      # Drain remaining
-      4.times { pull.receive }
-    ensure
-      push&.close
-      pull&.close
-    end
-  end
-
-
   it "unbounded via HWM=nil" do
     Async do
       push = OMQ::PUSH.new(nil, linger: 0)