Feature #21869: Add receive_all Method to Ractor API for Message Batching

• Status: Open
• Assignee: ractor
• Target version: -

[synacker (Mikhail Milovidov)]

Summary
The Ractor API provides an excellent mechanism for inter‑thread communication, but it currently lacks a built‑in message batching technique. I propose adding a receive_all method to enable batch processing of messages, which can significantly improve performance in high‑load scenarios.

Motivation
In distributed queued systems, processing messages one‑by‑one (as with the current receive method) can introduce unnecessary overhead. Batch processing allows:

Reduced context‑switching overhead.

More efficient I/O operations (e.g., fewer file writes).

Better throughput in high‑concurrency environments.

Proposed Solution
Add a receive_all method to the Ractor API that:

Returns all available messages in the Ractor’s mailbox at once (as an array).

Demonstration Code
Below is a benchmark comparing individual receive vs. batch receive_all:

require 'benchmark'
class RactorsTest

  def initialize(count)
    @count = count
    @ractor1 = Ractor.new(count, 'output1.txt') do |count, filename|
      File.open(filename, 'w') do |file|
        while count.positive?
          message = receive
          file.write("Ractor 1 received message: #{message}\n")
          file.flush
          count -= 1
        end
      end
    end
    
    @ractor2 = Ractor.new(count, 'output2.txt') do |count, filename|
      File.open(filename, 'w') do |file|
        while count.positive?
          messages = receive_all
          messages.each do |message|
            file.write("Ractor 2 received message: #{message}\n")
          end
          count -= messages.length
          file.flush
        end
      end
    end
  end

  def run1
    @count.times do |i|
      @ractor1.send("Message #{i + 1}")
    end
    @ractor1.join
  end

  def run2
    @count.times do |i|
      @ractor2.send("Message #{i + 1}")
    end
    @ractor2.join
  end
end

records = 1_000_000

test = RactorsTest.new(records)

p [:once, Benchmark.realtime { test.run1 }.round(2)]
p [:all, Benchmark.realtime { test.run2 }.round(2)]

Benchmark Results
On my system, receive_all shows ~4x improvement over individual receive:

Key Observations:
Ractor1 (using receive): Processes each message individually, resulting in frequent I/O calls.

Ractor2 (using receive_all): Processes all queued messages at once, minimizing I/O overhead

[synacker (Mikhail Milovidov)]

PR in github: https://github.com/ruby/ruby/pull/16105

[ko1 (Koichi Sasada)]

Does it block when the queue is empty or returns []? (I think the example expects blocking)

[Eregon (Benoit Daloze)]

synacker (Mikhail Milovidov) wrote:

More efficient I/O operations (e.g., fewer file writes).

Is it? In your example you call file.write for each message in both cases.

But you also call file.flush after each file.write in ractor1 and only only once per batch in ractor2.
Could you benchmark without the file.flushs? I suspect the difference is much smaller then.

I understand the idea that batching helps in this case where you want to explicitly flush, but that's a pretty specific example, e.g. it's uncommon to even call IO#flush at all in Ruby.
One could also flush after N messages/bytes in ractor1.

[synacker (Mikhail Milovidov)]

ko1 (Koichi Sasada) wrote in #note-2:

Does it block when the queue is empty or returns []? (I think the example expects blocking)

Yes, it blocks if the queue empty. The method receive_all accepts a limit parameter:

1. limit > 0: collects up to limit messages (may return fewer if fewer are queued). Blocks if the queue is empty.
2. limit == 0: returns an empty array immediately (no blocking)
3. limit < 0 or nil (default): returns all messages from the queue or blocks if the queue is empty

Eregon (Benoit Daloze) wrote in #note-3:

I understand the idea that batching helps in this case where you want to explicitly flush, but that's a pretty specific example, e.g. it's uncommon to even call IO#flush at all in Ruby.

The example demonstrates how you can collect messages during long I/O operations and batch them together to reduce the number of subsequent I/O calls. The file.flush call simulates a long I/O operation - it could equally represent a database call or something like that

[synacker (Mikhail Milovidov)]

Eregon (Benoit Daloze) wrote in #note-3:

Is it? In your example you call file.write for each message in both cases.

But you also call file.flush after each file.write in ractor1 and only only once per batch in ractor2.

This is also a realistic scenario. To guarantee messages are saved to file, you'd normally need to call flush after each message - but that's inefficient when processing single messages.

[luke-gru (Luke Gruber)]

• Assignee set to ractor

[byroot (Jean Boussier)]

I understand the idea that batching helps in this case where you want to explicitly flush, but that's a pretty specific example, e.g. it's uncommon to even call IO#flush at all in Ruby.

Not specific to Ractor, but I relatively often had a similar use case with Thread::Queue. When implementing instrumentation libraries (e.g. statsd or opentelemetry) what you generally want to do is to push as much work as possible to a background thread (possibly ractor in the future), but you also want this background thread to serialize and send packets in batch.

In such context, being able to pop 1..N elements from the queue would be convenient. Currently with queue it's semi-doable by first doing a blocking pop, followed by a bounded number of non-blocking one.

[ko1 (Koichi Sasada)]

synacker (Mikhail Milovidov) wrote in #note-4:

Yes, it blocks if the queue empty. The method receive_all accepts a limit parameter:

1. limit > 0: collects up to limit messages (may return fewer if fewer are queued). Blocks if the queue is empty.
2. limit == 0: returns an empty array immediately (no blocking)
3. limit < 0 or nil (default): returns all messages from the queue or blocks if the queue is empty

The parameter limit is not in the description. Let me clarify limit == 0. In above quote, it returns empty array.
If it returns [] always, I think we don't need to introduce it.
If it returns existing messages, we can consider about it.

For this purpose, it is possible to introduce another parameter such as timeout: 0.

• receive_all(3, timeout: 0) returns up to 3 messages already arrived. If there is no messages arrived, returns nil (timeout).
• receive_all(timeout: 1) returns all messages in a queue, or wait for 1 second. If a message arrived, return the [received_one_message]. If no message arrived in 1 second, return nil.
• receive_all(0 or negative parameter) raises an exception.

We can extend timeout: on receive method with same manner.

[synacker (Mikhail Milovidov)]

ko1 (Koichi Sasada) wrote in #note-8:

For this purpose, it is possible to introduce another parameter such as timeout: 0.

• receive_all(3, timeout: 0) returns up to 3 messages already arrived. If there is no messages arrived, returns nil (timeout).
• receive_all(timeout: 1) returns all messages in a queue, or wait for 1 second. If a message arrived, return the [received_one_message]. If no message arrived in 1 second, return nil.
• receive_all(0 or negative parameter) raises an exception.

LGTM, I’ll update the PR with this suggestion.

[synacker (Mikhail Milovidov)]

ko1 (Koichi Sasada) wrote in #note-8:

If it returns [] always, I think we don't need to introduce it.

I've updated pr. The limit parameter behaves as fallow:

1. limit == nil: returns all messages in the queue, or blocks if the queue is empty.
2. limit > 0: returns an array of up to limit messages, or blocks if the queue is empty.
3. limit <= 0: raises an ArgumentError

ko1 (Koichi Sasada) wrote in #note-8:

For this purpose, it is possible to introduce another parameter such as timeout: 0.

Regarding the timeout suggestion: implementing it would require significant changes to the Ractor logic. I propose addressing this in a separate task.

[synacker (Mikhail Milovidov)]

This is my first pr and issue. I’m unsure about the current status — is the request rejected, or are there any actions required from my side?

[ko1 (Koichi Sasada)]

Sorry for late response.

Could you rebase it to the current master, and could you show some benchmark example we can understand it should be merged?
Now we are not sure it is valuable to introduce or not.

Or I can take over this proposal (and ask Claude code).

BTW

Reduced context‑switching overhead.

doesn't make sense nowadays because a << o while o = Ractor.receive(timeout:0)) can emulate it and no context switch now.

[synacker (Mikhail Milovidov)]

Thank you for the feedback, Koichi!

I’ve rebased the PR to the current master and added benchmark examples to illustrate the performance characteristics of receive_all. You can find the benchmarks here:

https://github.com/ruby/ruby/pull/16105/changes#diff-49ab12719789e1f00ae1d1a885ae83bfeda3a0f2be2348c1521fc8462c8f1f25

Here are the local results:

backlog fixnums             receive    0.04s   25637547/s   receive_all    0.02s   49357428/s   x1.93
backlog frozen strings      receive    0.06s   17879134/s   receive_all    0.05s   21309728/s   x1.19
backlog strings             receive    0.21s    4681628/s   receive_all    0.26s    3904338/s   x0.83
backlog 4 producers         receive    0.22s    4585560/s   receive_all    0.27s    3768649/s   x0.82
backlog 8 producers         receive    0.22s    4470051/s   receive_all    0.29s    3416123/s   x0.76
backlog 4 producers shared  receive    0.07s   13843510/s   receive_all    0.05s   18782983/s   x1.36
backlog 8 producers shared  receive    0.08s   12997252/s   receive_all    0.06s   16489434/s   x1.27

The performance degradation in some cases (e.g., strings, multiple producers) is due to GC overhead: frequent array allocations interfere with efficient GC behavior. In scenarios with shared backlogs or simpler objects (fixnums, frozen strings), receive_all shows a clear benefit.