fix: prevent consumer starvation in lavinmqperf throughput (#1712)

## Summary
- Add `Fiber.yield` every 128k publishes when no rate limit is set,
preventing the publisher from monopolizing a shared thread and starving
consumer fibers
- Fix ack/tx_commit to use per-consumer local counters instead of the
global atomic counter, preventing deadlocks with multiple consumers and
prefetch
- Fix `RateLimiter` to use its local counter for yield frequency (was
using global counter or stuck at 0)

## Context
When running `lavinmqperf throughput` on machines with few CPU cores,
`Fiber::ExecutionContext::Parallel` has limited threads. The publisher
and consumer fibers (plus AMQP client internal fibers) can share a
thread. The publisher's tight loop with zero yielding monopolized the
thread, causing the consumer to stop reading from its socket. The
server's TCP send buffer filled up and delivery stalled permanently.

Closes #1711

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Carl Hörberg <[email protected]>

Author: viktorerlingsson

src/lavinmqperf/amqp/throughput.cr

@@ -354,6 +354,7 @@ module LavinMQPerf
           pubs_this_second = 0
           queues = queue_names
           queue_idx = 0
+          local_pubs = 0_u64
           until @stopped
             if @measure_latency
               # Write timestamp at the beginning of the message
@@ -383,8 +384,10 @@ module LavinMQPerf
             end
             queue_idx += 1 if @queue_pattern
             pubs = @pubs.add(1, :relaxed) + 1
-            ch.tx_commit if @pub_in_transaction > 0 && (pubs % @pub_in_transaction) == 0
+            local_pubs &+= 1
+            ch.tx_commit if @pub_in_transaction > 0 && (local_pubs % @pub_in_transaction) == 0
             break if pubs >= @pmessages > 0
+            Fiber.yield if @rate.zero? && local_pubs % (128*1024) == 0
             unless @rate.zero?
               pubs_this_second += 1
               if pubs_this_second >= @rate
@@ -418,23 +421,25 @@ module LavinMQPerf
           q.bind(@exchange, @routing_key) unless @exchange.empty?
           wait_until_all_are_connected(connected)
           rate_limiter = RateLimiter.new(@consume_rate)
+          local_consumes = 0_u64
           q.subscribe(tag: "c", no_ack: @ack.zero?, block: true, args: @consumer_args) do |m|
-            handle_consumed_message(ch, m, data, rate_limiter)
+            local_consumes &+= 1
+            handle_consumed_message(ch, m, data, rate_limiter, local_consumes)
           end
         end
       end
 
-      private def handle_consumed_message(ch, m, data, rate_limiter)
+      private def handle_consumed_message(ch, m, data, rate_limiter, local_consumes : UInt64)
         consumes = @consumes.add(1, :relaxed) + 1
         body = m.body_io.to_slice
         record_latency(body) if @measure_latency
         verify_message_body(body, data) if @verify
-        m.ack(multiple: true) if @ack > 0 && consumes % @ack == 0
-        ch.tx_commit if @ack_in_transaction > 0 && (consumes % @ack_in_transaction) == 0
+        m.ack(multiple: true) if @ack > 0 && local_consumes % @ack == 0
+        ch.tx_commit if @ack_in_transaction > 0 && (local_consumes % @ack_in_transaction) == 0
         if @stopped || consumes >= @cmessages > 0
           ch.close
         end
-        rate_limiter.limit(consumes)
+        rate_limiter.limit
       end
 
       private def poll_consume(connected, queue_name : String)
@@ -453,11 +458,13 @@ module LavinMQPerf
           q.bind(@exchange, @routing_key) unless @exchange.empty?
           wait_until_all_are_connected(connected)
           rate_limiter = RateLimiter.new(@consume_rate)
+          local_consumes = 0_u64
           loop do
             if msg = q.get(no_ack: @ack.zero?)
               consumes = @consumes.add(1, :relaxed) + 1
+              local_consumes &+= 1
               record_latency(msg.body_io.to_slice) if @measure_latency
-              msg.ack(multiple: true) if @ack > 0 && consumes % @ack == 0
+              msg.ack(multiple: true) if @ack > 0 && local_consumes % @ack == 0
               break if @stopped || consumes >= @cmessages > 0
               rate_limiter.limit
             end
@@ -492,16 +499,15 @@ module LavinMQPerf
           @total_consumes = 0_u64
         end
 
-        def limit(consume_count : UInt64? = nil)
+        def limit
+          @total_consumes &+= 1
           if @rate.zero?
             # When no rate limiting, yield periodically to avoid blocking other fibers
-            count = consume_count || @total_consumes
-            Fiber.yield if count % (128*1024) == 0
+            Fiber.yield if @total_consumes % (128*1024) == 0
             return
           end
 
           @consumes_this_second += 1
-          @total_consumes += 1
           if @consumes_this_second >= @rate
             until_next_second = (@start + 1.seconds) - Time.instant
             sleep until_next_second if until_next_second > Time::Span.zero