Further optimize Bunny::Channel#basic_publish_batch

From 14% to 23% depending on the message size.
1 kib and 4 kiB workloads benefit the most.

Author: michaelklishin

ChangeLog.md

@@ -71,7 +71,7 @@ GitHub issue: [#700](https://github.com/ruby-amqp/bunny/issues/700).
 
 ### Significant Publisher Performance Improvements
 
-**Performance** (100K messages, with [amq-protocol `2.4.0`](https://github.com/ruby-amqp/amq-protocol/releases/tag/v2.4.0) or later)
+Publisher performance improvements (100K messages, with [amq-protocol `2.4.0`](https://github.com/ruby-amqp/amq-protocol/releases/tag/v2.4.0) or later)
 with automatic publisher confirm tracking enabled (documented below):
 
 | Approach | Throughput | vs 2.x confirms |
@@ -90,6 +90,9 @@ connection flow control on the RabbitMQ end).
 To migrate from `2.x`, simply replace `Channel#confirm_select` calls with with `Channel#confirm_select(tracking: true)`.
 That's it.
 
+In addition, `Bunny::Channel#basic_publish_batch` benefits further from the write hot path optimizations
+that do not benefit `Bunny::Channel#basic_publish` much.
+
 ### Publisher Confirm Tracking
 
 Bunny now supports [publisher confirm](https://www.rabbitmq.com/docs/publishers#data-safety)

benchmarks/throughput.rb

@@ -0,0 +1,100 @@
+#!/usr/bin/env ruby
+
+require "bunny"
+
+MESSAGE_SIZES = [
+  ["12 B",    12,       100_000],
+  ["1 KiB",   1_024,    100_000],
+  ["4 KiB",   4_096,    100_000],
+  ["16 KiB",  16_384,   100_000],
+]
+
+PREFETCH   = 500
+MULTI_ACK  = 100
+BATCH_SIZE = 500
+
+puts
+puts "-" * 72
+puts "Bunny #{Bunny::VERSION} on #{RUBY_DESCRIPTION}"
+puts "-" * 72
+
+def run_benchmark(label, size, count, use_batch: false)
+  payload = "x" * size
+
+  pub_conn = Bunny.new
+  pub_conn.start
+  con_conn = Bunny.new
+  con_conn.start
+
+  pub_ch = pub_conn.create_channel
+  con_ch = con_conn.create_channel
+  con_ch.prefetch(PREFETCH)
+
+  q = con_ch.queue("bunny.bench", auto_delete: true)
+  q.purge
+
+  received = 0
+  done = Queue.new
+
+  q.subscribe(manual_ack: true) do |delivery_info, _properties, _body|
+    received += 1
+    if (received % MULTI_ACK).zero?
+      con_ch.basic_ack(delivery_info.delivery_tag, true)
+    end
+    done.push(true) if received == count
+  end
+
+  t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+
+  if use_batch
+    payloads = Array.new(BATCH_SIZE, payload)
+    (count / BATCH_SIZE).times { pub_ch.basic_publish_batch(payloads, "", q.name, persistent: false) }
+    remainder = count % BATCH_SIZE
+    if remainder > 0
+      pub_ch.basic_publish_batch(Array.new(remainder, payload), "", q.name, persistent: false)
+    end
+  else
+    count.times { pub_ch.basic_publish(payload, "", q.name, persistent: false) }
+  end
+  done.pop
+
+  elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
+  rate = (count / elapsed).round(0)
+  mb_sec = (count * size / elapsed / 1_048_576.0).round(1)
+
+  pub_conn.close
+  con_conn.close
+
+  [label, count, rate, mb_sec]
+end
+
+puts
+puts "## basic_publish (per-message)"
+puts
+
+results = []
+MESSAGE_SIZES.each do |label, size, count|
+  r = run_benchmark(label, size, count)
+  results << r
+  printf "%-12s  %6dK msgs  %8d msg/sec  %8.1f MB/sec\n", r[0], r[1] / 1000, r[2], r[3]
+end
+
+puts
+puts "## basic_publish_batch (#{BATCH_SIZE} msgs/batch)"
+puts
+
+batch_results = []
+MESSAGE_SIZES.each do |label, size, count|
+  r = run_benchmark(label, size, count, use_batch: true)
+  batch_results << r
+  printf "%-12s  %6dK msgs  %8d msg/sec  %8.1f MB/sec\n", r[0], r[1] / 1000, r[2], r[3]
+end
+
+puts
+puts "-" * 72
+puts "| Workload | basic_publish | batch (#{BATCH_SIZE}) |"
+puts "|----------|--------:|-------:|"
+results.zip(batch_results).each do |r, br|
+  puts "| #{r[1] / 1000}K x #{r[0]} | #{r[2]} msg/sec | #{br[2]} msg/sec |"
+end
+puts "-" * 72

lib/bunny/channel.rb

@@ -261,6 +261,7 @@ def open
       @connection.open_channel(self)
       # clear last channel error
       @last_channel_error = nil
+      @frame_max = @connection.frame_max
 
       @status = :open
 
@@ -712,7 +713,7 @@ def basic_publish(payload, exchange, routing_key, opts = {})
         routing_key,
         opts[:mandatory],
         false,
-        @connection.frame_max)
+        @frame_max)
       @connection.send_frameset(frames, self)
 
       wait_for_publish_confirm(seq_no, continuation) if continuation
@@ -781,7 +782,8 @@ def basic_publish_batch(payloads, exchange, routing_key, opts = {})
         end
       end
 
-      # Send all frames (outside the mutex)
+      # Encode all messages into a single buffer and write once
+      data = +""
       payloads.each do |payload|
         frames = AMQ::Protocol::Basic::Publish.encode(@id,
           payload,
@@ -790,9 +792,10 @@ def basic_publish_batch(payloads, exchange, routing_key, opts = {})
           routing_key,
           opts[:mandatory],
           false,
-          @connection.frame_max)
-        @connection.send_frameset(frames, self)
+          @frame_max)
+        frames.each { |frame| data << frame.encode }
       end
+      @connection.send_raw_without_timeout(data, self)
 
       self
     end

lib/bunny/session.rb

@@ -1474,7 +1474,8 @@ def send_frameset_without_timeout(frames, channel)
       # locking. See a note about "single frame" methods in a comment in `send_frameset`. MK.
       channel.synchronize do
         if open?
-          frames.each { |frame| self.send_frame_without_timeout(frame, false) }
+          data = frames.reduce(+"") { |acc, frame| acc << frame.encode }
+          @transport.write_without_timeout(data)
           signal_activity!
         else
           raise ConnectionClosedError.new(frames)
@@ -1490,10 +1491,14 @@ def send_raw_without_timeout(data, channel)
       # If we synchronize on the channel, however, this is both thread safe and pretty fine-grained
       # locking. Note that "single frame" methods do not need this kind of synchronization. MK.
       channel.synchronize do
-        @transport.write(data)
-        signal_activity!
+        if open?
+          @transport.write(data)
+          signal_activity!
+        else
+          raise ConnectionClosedError.new("pre-encoded data (#{data.bytesize} bytes)")
+        end
       end
-    end # send_frameset_without_timeout(frames)
+    end # send_raw_without_timeout(data)
 
     # @return [String]
     # @api public

lib/bunny/transport.rb

@@ -180,8 +180,10 @@ def write(data)
     # Writes data to the socket without timeout checks
     def write_without_timeout(data, raise_exceptions = false)
       begin
-        @writes_mutex.synchronize { @socket.write(data) }
-        @socket.flush
+        @writes_mutex.synchronize do
+          @socket.write(data)
+          @socket.flush
+        end
       rescue SystemCallError, Bunny::ConnectionError, IOError => e
         close
         raise e if raise_exceptions