Ractors

# frozen_string_literal: true

# The first Ractor.new call in a process prints the experimental warning.
greeter = Ractor.new { "Hello from a ractor" }
greeter.value # => "Hello from a ractor"
greeter.inspect.match?(/\A#<Ractor:#\d+ ex01_first.rb:4 terminated>\z/) # => true

Ractor.main? # => true
Ractor.count # => 1
Ractor.current.equal?(Ractor.main) # => true
Ractor.new { Ractor.main? }.value # => false

ex01_first.rb:4: warning: Ractor API is experimental and may change in future versions of Ruby.

# frozen_string_literal: true

Warning[:experimental] = false

limit = 3
begin
  Ractor.new { limit }
rescue ArgumentError => e
  e.message # => "can not isolate a Proc because it accesses outer variables (limit)."
end

# Arguments of Ractor.new become block parameters; unshareable ones are copied.
Ractor.new(limit) { |max_loans| max_loans * 2 }.value # => 6

# self inside the block is the Ractor, and a name helps debugging.
worker = Ractor.new(name: "indexer") { Ractor.current.name }
worker.value # => "indexer"
worker.name # => "indexer"
Ractor.new { equal?(Ractor.current) }.value # => true

begin
  Ractor.new
rescue ArgumentError => e
  e.message # => "must be called with a block"
end

# frozen_string_literal: true

Warning[:experimental] = false
# Without this line, an unhandled exception in a ractor is also printed to stderr
# by its main thread ("terminated with exception (report_on_exception is true)").
Thread.report_on_exception = false

failing = Ractor.new { raise ArgumentError, "invalid ISBN: 123" }
begin
  failing.value
rescue Ractor::RemoteError => e
  e.message # => "thrown by remote Ractor."
  e.cause # => #<ArgumentError: invalid ISBN: 123>
  e.ractor.equal?(failing) # => true
end

# join raises the same way; a ractor that ended normally just returns itself.
begin
  Ractor.new { raise "boom" }.join
rescue Ractor::RemoteError => e
  e.cause.message # => "boom"
end
Ractor.new { :ok }.join.class # => Ractor

# value hands the result to one Ractor only. The caller may ask again,
# but a second Ractor that asks gets an error; join has no such limit.
result = Ractor.new { [1, 2, 3] }
result.value # => [1, 2, 3]
result.value # => [1, 2, 3]
latecomer = Ractor.new(result) { |other| other.value.size }
begin
  latecomer.value
rescue Ractor::RemoteError => e
  e.cause.message # => "Only the successor ractor can take a value"
end
Ractor.new(result) { |other| other.join.class }.value # => Ractor

# frozen_string_literal: true

Warning[:experimental] = false

Ractor.shareable?(42) # => true
Ractor.shareable?(:sym) # => true
Ractor.shareable?(nil) # => true
Ractor.shareable?("literal") # => true
Ractor.shareable?(+"mutable") # => false
Ractor.shareable?(1..3) # => true
Ractor.shareable?(Object) # => true
Ractor.shareable?([1, 2].freeze) # => true
Ractor.shareable?([+"a"].freeze) # => false
Ractor.shareable?(Set[1, 2]) # => false
Ractor.shareable?(Set[1, 2].freeze) # => true
Ractor.shareable?(Ractor.current) # => true
Ractor.shareable?(Queue.new) # => false
Ractor.shareable?(proc { 1 }.freeze) # => false

Point = Data.define(:x, :y)
Ractor.shareable?(Point.new(x: 1, y: 2)) # => true
Ractor.shareable?(Point.new(x: +"1", y: 2)) # => false

# A frozen Data object can still hold an unfrozen String: String#delete returns a new one.
Isbn = Data.define(:digits)
isbn = Isbn.new(digits: "978-0-306-40615-7".delete("-"))
isbn.frozen? # => true
Ractor.shareable?(isbn) # => false
Ractor.shareable?(Ractor.make_shareable(isbn, copy: true)) # => true

# make_shareable freezes the whole graph in place ...
config = { name: +"shelf", limits: [3, 5] }
shared = Ractor.make_shareable(config)
shared.equal?(config) # => true
config[:name].frozen? # => true
config[:limits].frozen? # => true

# ... unless copy: true, which leaves the original untouched.
original = [+"fiction"]
copy = Ractor.make_shareable(original, copy: true)
[original.frozen?, copy.frozen?, copy.equal?(original)] # => [false, true, false]

begin
  Ractor.make_shareable(Thread.current)
rescue Ractor::Error => e
  e.message.sub(/0x\h+/, "0x...") # => "can not make shareable object for #<Thread:0x... run>"
end
begin
  Ractor.make_shareable(proc { 1 })
rescue Ractor::IsolationError => e
  e.message[/\A[^#]*/] # => "Proc's self is not shareable: "
end

# frozen_string_literal: true

Warning[:experimental] = false

# A port belongs to the Ractor that created it: anyone may send, only the owner receives.
inbox = Ractor::Port.new
producer = Ractor.new(inbox) do |port|
  port << "978-0-306-40615-7"
  port.send("978-1-4028-9462-6")
  :finished
end
inbox.receive # => "978-0-306-40615-7"
inbox.receive # => "978-1-4028-9462-6"
producer.value # => :finished

# Closing: sending to a closed port fails, receiving drains what is queued first.
queue = Ractor::Port.new
queue << 1
queue << 2
queue.close
queue.closed? # => true
queue.receive # => 1
queue.receive # => 2
begin
  queue.receive
rescue Ractor::ClosedError => e
  e.message # => "The port was already closed"
end
begin
  queue << 3
rescue Ractor::ClosedError => e
  e.message # => "The port was already closed"
end

# Every Ractor has a default port behind Ractor.receive, Ractor#send and Ractor#<<.
upcaser = Ractor.new do
  Ractor.receive.upcase
end
upcaser << "shelf"
upcaser.value # => "SHELF"
Ractor.current.default_port.class # => Ractor::Port

# Ractor::ClosedError inherits from StopIteration, so loop ends quietly once a closed port
# has been drained.
words = Ractor::Port.new
words << "ruby" << "ractor"
words.close
total = 0
loop { total += words.receive.size }
total.to_s # => "10"

# frozen_string_literal: true

Warning[:experimental] = false

# Copy (the default): an unshareable argument is deep-copied.
titles = [+"Dune", +"Emma"]
copied = Ractor.new(titles) { |received| received.map(&:object_id) }.value
copied.eql?(titles.map(&:object_id)) # => false
titles.first # => "Dune"

# Shareable objects are sent by reference.
frozen_title = "Dune"
Ractor.new(frozen_title) { |received| [received.object_id] }.value.eql?([frozen_title.object_id]) # => true

# Move: ownership changes, so no copy is made and the sender loses access.
draft = +"Dune"
editor = Ractor.new do
  Ractor.receive << " (1965)"
end
editor.send(draft, move: true)
editor.value # => "Dune (1965)"
[draft].grep(Ractor::MovedObject).size # => 1
begin
  draft << "!"
rescue Ractor::MovedError => e
  e.message # => "can not send any methods to a moved object"
end

# Some objects can be neither copied nor moved.
begin
  Ractor.new(Thread.current) { |t| t }
rescue TypeError => e
  e.message # => "allocator undefined for Thread"
end
begin
  Ractor.new(proc { 1 }) { |pr| pr.call + 1 }
rescue TypeError => e
  e.message # => "allocator undefined for Proc"
end

# frozen_string_literal: true

Warning[:experimental] = false
Thread.report_on_exception = false

# Ractor.select returns [port, message] for ports and [ractor, value] for ractors.
slow = Ractor.new do
  sleep 0.2
  "slow"
end
fast = Ractor.new { "fast" }
source, value = Ractor.select(slow, fast)
source.equal?(fast) # => true
value.upcase # => "FAST"
source, value = Ractor.select(slow)
[source.equal?(slow), value] # => [true, "slow"]

# Mixing ports and ractors: collect messages until every worker has finished.
results = Ractor::Port.new
workers = Array.new(3) do |i|
  Ractor.new(results, i) do |port, n|
    port << "line #{n}"
    n
  end
end
messages = []
pending = workers.dup
until pending.empty? && messages.size == 3
  source, value = Ractor.select(results, *pending)
  if source.is_a?(Ractor)
    pending.delete(source)
  else
    messages << value
  end
end
messages.sort # => ["line 0", "line 1", "line 2"]

# monitor: a port receives :exited or :aborted when a Ractor terminates.
ok = Ractor.new { :done }
ok.monitor(status = Ractor::Port.new)
status.receive # => :exited
bad = Ractor.new { raise "bad record" }
bad.monitor(status = Ractor::Port.new)
status.receive # => :aborted

begin
  Ractor.select
rescue ArgumentError => e
  e.message # => "specify at least one Ractor::Port or Ractor"
end
begin
  Ractor.select(:nope)
rescue ArgumentError => e
  e.message # => "should be Ractor::Port or Ractor"
end

# frozen_string_literal: true

Warning[:experimental] = false

module Shelf
  # shareable_constant_value: none
  PLAIN = { max_loans: 3 } # rubocop:disable Style/MutableConstant

  # shareable_constant_value: literal
  LIMITS = { max_loans: 3, grace_days: 2 }
  GENRES = %w[fiction poetry history]

  # shareable_constant_value: experimental_everything
  OPENED = Time.at(0).utc

  # shareable_constant_value: experimental_copy
  seed = [+"seed"]
  SEED = seed
end

Ractor.shareable?(Shelf::PLAIN) # => false
Ractor.shareable?(Shelf::LIMITS) # => true
Ractor.shareable?(Shelf::GENRES) # => true
Ractor.shareable?(Shelf::OPENED) # => true
Ractor.shareable?(Shelf::SEED) # => true
Ractor.new { Shelf::LIMITS[:max_loans] }.value # => 3

# frozen_string_literal: true
# shareable_constant_value: literal

def load_title = +"Dune"

# A literal is made shareable; the result of a method call is only checked.
TITLES = %w[Dune Emma]
OPENED = load_title

ex10_literal_only.rb:8:in 'RubyVM::FrozenCore.ensure_shareable': cannot assign unshareable object to OPENED (Ractor::IsolationError)
	from ex10_literal_only.rb:8:in '<main>'

# frozen_string_literal: true

Warning[:experimental] = false
Thread.report_on_exception = false

# A plain proc cannot be shared, even when frozen.
Ractor.shareable?(proc { 1 }.freeze) # => false

double = Ractor.shareable_proc { it * 2 }
Ractor.shareable?(double) # => true
Ractor.new(double) { |f| f.call(21) }.value # => 42

# self is nil unless a shareable self: is given.
Ractor.shareable_proc { self }.call # => nil
Ractor.shareable_proc(self: :catalog) { self }.call # => :catalog

square = Ractor.shareable_lambda { |n| n * n }
square.lambda? # => true
square.call(7) # => 49

# Outer variables: only never-reassigned variables holding shareable values are accepted.
limit = 3
Ractor.shareable_proc { limit }.call # => 3
name = +"mutable"
begin
  Ractor.shareable_proc { name }
rescue Ractor::IsolationError => e
  e.message # => "cannot make a shareable Proc because it can refer unshareable object \"mutable\" from variable 'name'"
end
counter = 0
counter += 1
begin
  Ractor.shareable_proc { counter }
rescue Ractor::IsolationError => e
  e.message # => "cannot make a shareable Proc because the outer variable 'counter' may be reassigned."
end

# define_method needs a shareable proc for the method to be callable from any Ractor.
class Book
  define_method :label, &Ractor.shareable_proc { "Book #{self.class}" }
  define_method :plain_label, &-> { "Book #{self.class}" }
end
Ractor.new { Book.new.label }.value # => "Book Book"
begin
  Ractor.new { Book.new.plain_label }.value
rescue Ractor::RemoteError => e
  e.cause.message # => "defined with an un-shareable Proc in a different Ractor"
end

# frozen_string_literal: true

Warning[:experimental] = false

# Ractor-local storage can hold unshareable values and is private to one Ractor.
Ractor[:request_log] = [+"main"]
Ractor[:request_log] << "more"
Ractor.new { Ractor[:request_log] }.value # => nil

worker = Ractor.new do
  Ractor[:seen] = []
  # Threads of the same Ractor share its storage and its lock (the GVL).
  threads = Array.new(3) { |i| Thread.new { Ractor[:seen] << i } }
  threads.each(&:join)
  Ractor[:seen].sort
end
worker.value # => [0, 1, 2]

# store_if_absent runs the block at most once per Ractor, even from several threads.
calls = Ractor.new do
  count = Ractor.store_if_absent(:setup_calls) { [0] }
  Array.new(8) { Thread.new { Ractor.store_if_absent(:setup_calls) { [99] } } }.each(&:join)
  count
end
calls.value # => [0]
Ractor[:request_log] # => ["main", "more"]

# frozen_string_literal: true

Warning[:experimental] = false

# The ISBN-13 check from Shelf::Isbn as a plain method; methods are shared by all Ractors.
def isbn_valid?(digits)
  return false unless digits.match?(/\A\d{13}\z/)

  sum = digits.chars.map(&:to_i).each_with_index.sum { |d, i| i.even? ? d : d * 3 }
  (sum % 10).zero?
end

ISBNS = %w[
  9780306406157 9781402894626 9780596516178 9780131103627
  9780306406158 9781402894627 9780596516179 9780131103628
  9780201633610 9780321125217 9780134757599 9781491950357
].freeze

results = Ractor::Port.new
workers = Array.new(4) do |index|
  Ractor.new(results, index) do |out, id|
    handled = 0
    while (isbn = Ractor.receive)
      out << [isbn, isbn_valid?(isbn)]
      handled += 1
    end
    [id, handled]
  end
end

ISBNS.each_with_index { |isbn, i| workers[i % workers.size] << isbn }
checked = Array.new(ISBNS.size) { results.receive }.to_h
workers.each { |worker| worker << nil }

checked.count { |_isbn, valid| valid } # => 8
checked["9780306406158"] # => false
workers.map(&:value) # => [[0, 3], [1, 3], [2, 3], [3, 3]]

# frozen_string_literal: true

Warning[:experimental] = false

# Each stage reads from its default port and forwards to the next Ractor; nil ends the stream.
def stage(next_stage, &)
  Ractor.new(next_stage, Ractor.shareable_proc(&)) do |downstream, work|
    while (item = Ractor.receive)
      downstream << work.call(item)
    end
    downstream << nil
  end
end

sink = Ractor.current
label = stage(sink) { |isbn| "ISBN #{isbn}" }
check = stage(label) { |raw| raw.delete("-") }
normalize = stage(check) { |raw| raw.strip.squeeze(" ") }

["978-0-306-40615-7 ", " 978-1-4028-9462-6"].each { |raw| normalize << raw }
normalize << nil

Ractor.receive # => "ISBN 9780306406157"
Ractor.receive # => "ISBN 9781402894626"
Ractor.receive # => nil

# frozen_string_literal: true

Warning[:experimental] = false

# CPU-bound work that allocates nothing: count valid ISBN-13 check digits in a numeric range.
def count_valid(range)
  range.count do |n|
    sum = 0
    13.times do |i|
      n, digit = n.divmod(10)
      sum += i.even? ? digit : digit * 3
    end
    (sum % 10).zero?
  end
end

def measure
  started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  result = yield
  [result, (Process.clock_gettime(Process::CLOCK_MONOTONIC) - started).round(2)]
end

WORKERS = 4
RANGE = 9_780_000_000_000...9_780_003_200_000
slices = RANGE.each_slice(RANGE.size / WORKERS).map { |chunk| chunk.first..chunk.last }

sequential, t_seq = measure { count_valid(RANGE) }
threaded, t_threads = measure { slices.map { |slice| Thread.new { count_valid(slice) } }.sum(&:value) }
ractors, t_ractors = measure do
  slices.map { |slice| Ractor.new(slice) { |part| count_valid(part) } }.sum(&:value)
end

puts "sequential: #{sequential} valid in #{t_seq}s"
puts "threads:    #{threaded} valid in #{t_threads}s"
puts "ractors:    #{ractors} valid in #{t_ractors}s"
puts "same answer: #{[sequential, threaded, ractors].uniq.size == 1}"

sequential: 320000 valid in 3.43s
threads:    320000 valid in 3.08s
ractors:    320000 valid in 1.79s
same answer: true

# frozen_string_literal: true

Warning[:experimental] = false
Thread.report_on_exception = false

# A supervisor is plain Ruby around value: start a fresh Ractor when one aborts.
attempts = 0
loaded = begin
  attempts += 1
  Ractor.new(attempts) do |n|
    raise IOError, "flaky store" if n < 3

    "loaded on attempt #{n}"
  end.value
rescue Ractor::RemoteError => e
  retry if e.cause.is_a?(IOError) && attempts < 3
  raise
end
[attempts, loaded] # => [3, "loaded on attempt 3"]