https://ivoanjo.me/blog/2021/02/14/ractor-experiments-safe-async/

ruby ractor experiments: safe async communication

ivo anjo * february 14th, 2021 *  9 minute read

Ractors (api documentation, design documentation) are a new
concurrency abstraction for Ruby 3.0 inspired on the actor model.

From the point of view of a Ractor that wants to send some
information to another, communication can either be:

  * asynchrous (or non-blocking): a Ractor can send information to
    another using Ractor#send, placing it into an infinite queue that
    can be read by the destination Ractor with Ractor.receive

  * synchronous (or blocking): a Ractor can use Ractor.yield to block
    until another Ractor calls Ractor#take

Let's consider the async case: Let's say we want to send information
to another Ractor, but don't want to block for it to finish
processing it. What happens if the receiving Ractor is too slow to
process the data?

receiver_ractor = Ractor.new do
  loop do
    message = Ractor.receive
    sleep 1
    puts "Processed #{message}"
  end
end

counter = 0
while true
  counter += 1
  receiver_ractor.send(counter)
end

ractor unbounded memory


As expected, if the receiver cannot keep up with the sender, more and
more memory will be used, until the system memory is exhausted and
the application crashes.

Looking at the Ractor API, there's no built-in way for the sender to
check if the receiver is falling behind or not, so I came up with the
following approach:

receiver_ractor = Ractor.new do
  processing_queue = Queue.new

  Thread.new do
    sleep(1) # simulate a slow start for this thread

    loop do
      message = processing_queue.pop
      puts "Processed from queue: #{message}"
    end
  end

  loop do
    queue_size = processing_queue.size
    sender, message = Ractor.select(Ractor.current, yield_value: queue_size)

    if sender != :yield
      processing_queue << message
      puts "Added message to queue: #{message}"
    else
      puts "Sent queue status: #{queue_size}"
    end
  end
end

receiver_ractor.send(1)
receiver_ractor.send(2)
receiver_ractor.send(3)
puts "Finished submissions"

sleep(0.5)

receiver_ractor.take # force refresh status
puts "Receiver queue length: #{receiver_ractor.take}"

sleep(1)

receiver_ractor.take # force refresh status
puts "Receiver queue length: #{receiver_ractor.take}"

Here's how this looks when it executes:

<internal:ractor>:267: warning: Ractor is experimental, and the behavior may change in future versions of Ruby! Also there are many implementation issues.
Finished submissions
Added message to queue: 1
Added message to queue: 2
Added message to queue: 3
Sent queue status: 3
Sent queue status: 3
Receiver queue length: 3
Processed from queue: 1
Processed from queue: 2
Processed from queue: 3
Sent queue status: 3
Sent queue status: 0
Receiver queue length: 0

And here's the timeline:

 1. The main Ractor sends three messages to the receiver_ractor,
    containing 1, 2 and 3

 2. The main Ractor goes to sleep

 3. The receiver Ractor wakes up, reads the three messages, and
    redirects them to the processing_queue

 4. The main Ractor wakes up, forces a refresh of the queue_size (see
    below for more details)

 5. The main Ractor calls take, observes the queue_size is 3

 6. The main Ractor goes to sleep

 7. The receiver Ractor's second thread wakes up and processes the
    three messages: 1, 2 and 3

 8. The main Ractor wakes up, forces a refresh of the queue_size
    (again) -- notice that this refresh was needed, because the value
    was outdated (goes from 3 to 0)

 9. The main Ractor observes that the queue is empty!

Inside the receiver Ractor, this strategy works as follows: There are
now two threads. One of the threads uses Ractor.select to do two
things at once -- either receive new items for processing, putting
them on a regular thread-safe queue, or returning back the current
size of the queue. The second thread just processes items from the
thread-safe queue.

The sender can now either use send to send items, or call take twice
to get the size of the queue. Why twice? Because this value is only
refreshed before select gets called, if a long time passes between
select is entered and any calls to send or take, this value can
become outdated, as happened the example above. Calling take twice in
a row guarantees that we get a "fresh" value -- we know the value was
just refreshed for the second take.

Building atop this construction, we can implement a number of
strategies for better communication between two Ractors. For
instance, we can review the original example to make sure that the
sender never runs "too far ahead" the receiver:

receiver_ractor = Ractor.new do
  processing_queue = Queue.new

  Thread.new do
    loop do
      message = processing_queue.pop
      sleep(1)
      puts "Processed #{message}"
    end
  end

  loop do
    queue_size = processing_queue.size
    sender, message = Ractor.select(Ractor.current, yield_value: queue_size)

    if sender != :yield
      processing_queue << message
      puts "Added message to queue: #{message}"
    else
      puts "Sent queue status: #{queue_size}"
    end
  end
end

counter = 0
while true
  counter += 1
  receiver_ractor.send(counter)

  if counter % 10 == 0
    receiver_ractor.take # force refresh status
    queue_size = receiver_ractor.take
    if queue_size > 5
      puts "Ractor is falling behind (#{queue_size} elements unprocessed); sleeping for a while"
      sleep(1) while (receiver_ractor.take && receiver_ractor.take > 1)
    end
  end
end

And here's how it looks:

<internal:ractor>:267: warning: Ractor is experimental, and the behavior may change in future versions of Ruby! Also there are many implementation issues.
Added message to queue: 1
Added message to queue: 2
Added message to queue: 3
Added message to queue: 4
Added message to queue: 5
Added message to queue: 6
Added message to queue: 7
Added message to queue: 8
Added message to queue: 9
Added message to queue: 10
Sent queue status: 10
Sent queue status: 9
Ractor is falling behind (9 elements unprocessed); sleeping for a while
Sent queue status: 9
Sent queue status: 9
Processed 1
Sent queue status: 9
Sent queue status: 8
Processed 2
Sent queue status: 8
Sent queue status: 7
Processed 3
Sent queue status: 7
Sent queue status: 6
Processed 4
Sent queue status: 6
Sent queue status: 5
Processed 5
Sent queue status: 5
Sent queue status: 4
Processed 6
Sent queue status: 4
Sent queue status: 3
Processed 7
Sent queue status: 3
Sent queue status: 2
Processed 8
Sent queue status: 2
Sent queue status: 1
Added message to queue: 11
Added message to queue: 12
Added message to queue: 13
Added message to queue: 14
Added message to queue: 15
Added message to queue: 16
Added message to queue: 17
Added message to queue: 18
Added message to queue: 19
Added message to queue: 20
Sent queue status: 11
Sent queue status: 11
Ractor is falling behind (11 elements unprocessed); sleeping for a while

Sleeping is a very simplistic solution, but gets the job done. Other
alternatives could be to switch to synchronous communication, or to
instead submit the work to a different Ractor/code path.

That's it for my first experiment with Ractors!

talk to me about this post: ivo@ this domain / twitter
interested in my blog? get notified of new posts via email

other posts @ ivo's awfully random tech blog:

  * sunday lol: embedding images in commit logs (mar 14, 2021)
  * ruby ractor experiments: safe async communication (feb 14, 2021)
  * looking into array memory usage in ruby (feb 11, 2021)
  * what i've been reading: december+january 2021 edition (feb 6,
    2021)
  * what i've been reading: november 2020 edition (dec 2, 2020)
  * creating a newsletter! (dec 1, 2020)
  * what i've been reading: october 2020 edition (nov 3, 2020)
  * better backtraces in ruby using tracepoint (jul 19, 2020)
  * snippet: getting a dynamically-generated method name on the java
    stack using javassist (jul 12, 2020)
  * ruby experiment: include class names in backtraces (jul 5, 2020)
  * quick tip: unsafe concurrent ruby hash access (aug 26, 2019)
  * kotlin hack: transparently replace class with interface (aug 10,
    2019)
  * kotlin for rubyists (mar 5, 2019)
  * spotting unsafe ruby patterns - talk recording (oct 13, 2018)
  * writing to a java treemap concurrently can lead to an infinite
    loop during reads (jul 21, 2018)
  * til: java hides lambda frames in stack traces (jun 21, 2018)
  * my thoughts on, and how i approach code reviews (apr 8, 2018)
  * is this ok... ? or, spotting unsafe concurrent ruby patterns (jan
    31, 2018)
  * lightning talk - warm-blanket: goodbye crappy after-boot
    performance (jan 7, 2018)
  * persistent-: a new ruby gem for beautiful immutable data
    structures (jan 3, 2018)
  * asciidoc: an awesome markdown alternative (oct 22, 2017)
  * why i always use attr_reader to access instance variables (sep
    20, 2017)
  * introducing the warmblanket gem (aug 20, 2017)
  * ninjas' guide to getting started with visualvm (aug 12, 2017)
  * adopting tls (may 2, 2017)
  * rubies: a look at ruby's shiny future (apr 17, 2017)
  * quickies: heroku exec and deploying jruby (mar 19, 2017)
  * why you should be using jruby in production (mar 16, 2017)
  * benchmarking jruby invokedynamic with a production application (
    jan 29, 2017)
  * pry-debugger-jruby gem now on rubygems! (dec 2, 2016)
  * weekend hacking: enviado gem (nov 22, 2016)
  * jruby's charles nutter on the jvm as a language platform (nov 15,
    2016)
  * peek and pick at mri's heap, part 2 (nov 13, 2016)
  * finding dead ruby code with debride (oct 29, 2016)
  * psa: you can now debug with pry on jruby (oct 12, 2016)
  * peek and pick at mri's heap, part 1 (oct 10, 2016)
  * why you should regenerate your spec_helper (sep 25, 2016)
  * explaining git (sep 3, 2016)
  * another round of jruby goodness (jul 17, 2016)
  * down the jruby rabbit hole (jul 16, 2016)
  * did_you_know.ruby? (may 14, 2016)
  * asciinema: shell recording done right! (dec 1, 2015)
  * whoa! java has a repl now! (nov 28, 2015)
  * ruby meets weak memory models (nov 28, 2015)
  * ruby features i'm looking forward to (nov 13, 2015)
  * starting a new blog (nov 3, 2015)

...and a link to my homepage :)