класс Ractor

# File ractor.rb, line 302
def self.count
  __builtin_cexpr! %q{
    ULONG2NUM(GET_VM()->ractor.cnt);
  }
end

Возвращает количество текущих работающих или заблокированных (ожидающих) Ractor.

Ractor.count                   #=> 1
r = Ractor.new(name: 'example') { Ractor.yield(1) }
Ractor.count                   #=> 2 (main + example ractor)
r.take                         # wait for Ractor.yield(1)
r.take                         # wait until r will finish
Ractor.count                   #=> 1

# File ractor.rb, line 288
def self.current
  __builtin_cexpr! %q{
    rb_ractor_self(rb_ec_ractor_ptr(ec));
  }
end

Возвращает текущий выполняемый Ractor.

Ractor.current #=> #<Ractor:#1 running>

# File ractor.rb, line 848
def self.main
  __builtin_cexpr! %q{
    rb_ractor_self(GET_VM()->ractor.main_ractor);
  }
end

возвращает основной рактор

# File ractor.rb, line 825
def self.make_shareable obj, copy: false
  if copy
    __builtin_cexpr! %q{
      rb_ractor_make_shareable_copy(obj);
    }
  else
    __builtin_cexpr! %q{
      rb_ractor_make_shareable(obj);
    }
  end
end

Сделать obj доступным для совместного использования между ракторами.

obj и все объекты, на которые он ссылается, будут заморожены, если они ещё не являются разделяемыми.

Если copy ключевое слово равно true, он скопирует объекты перед заморозкой и не будет изменять obj или его внутренние объекты.

Обратите внимание, что спецификация и реализация этого метода не доработаны и могут быть изменены в будущем.

obj = ['test']
Ractor.shareable?(obj)     #=> false
Ractor.make_shareable(obj) #=> ["test"]
Ractor.shareable?(obj)     #=> true
obj.frozen?                #=> true
obj[0].frozen?             #=> true

# Copy vs non-copy versions:
obj1 = ['test']
obj1s = Ractor.make_shareable(obj1)
obj1.frozen?                        #=> true
obj1s.object_id == obj1.object_id   #=> true
obj2 = ['test']
obj2s = Ractor.make_shareable(obj2, copy: true)
obj2.frozen?                        #=> false
obj2s.frozen?                       #=> true
obj2s.object_id == obj2.object_id   #=> false
obj2s[0].object_id == obj2[0].object_id #=> false

См. также раздел «Разделяемые и неразделяемые объекты» в документации класса Ractor.

# File ractor.rb, line 273
def self.new(*args, name: nil, &block)
  b = block # TODO: builtin bug
  raise ArgumentError, "must be called with a block" unless block
  if __builtin_cexpr!("RBOOL(ruby_single_main_ractor)")
    warn("Ractor is experimental, and the behavior may change in future versions of Ruby! " \
         "Also there are many implementation issues.", uplevel: 0, category: :experimental)
  end
  loc = caller_locations(1, 1).first
  loc = "#{loc.path}:#{loc.lineno}"
  __builtin_ractor_create(loc, name, args, b)
end

Создать новый Ractor с аргументами и блоком.

Указанный блок (Proc) будет изолирован (не может получить доступ к внешним переменным). self внутри блока будет ссылаться на текущий Ractor.

r = Ractor.new { puts "Hi, I am #{self.inspect}" }
r.take
# Prints "Hi, I am #<Ractor:#2 test.rb:1 running>"

Любые args передаваемые в функцию, передаются в аргументы блока по тем же правилам, что и объекты, отправляемые через send/Ractor.receive. Если аргумент в args не является разделяемым, он будет скопирован (с помощью глубокого клонирования, что может быть неэффективно).

arg = [1, 2, 3]
puts "Passing: #{arg} (##{arg.object_id})"
r = Ractor.new(arg) {|received_arg|
  puts "Received: #{received_arg} (##{received_arg.object_id})"
}
r.take
# Prints:
#   Passing: [1, 2, 3] (#280)
#   Received: [1, 2, 3] (#300)

name Ractor можно задать для отладки:

r = Ractor.new(name: 'my ractor') {}; r.take
p r
#=> #<Ractor:#3 my ractor test.rb:1 terminated>

# File ractor.rb, line 430
def self.receive
  __builtin_cexpr! %q{
    ractor_receive(ec, rb_ec_ractor_ptr(ec))
  }
end

Получить сообщение из входного порта текущего рактора (которое было отправлено туда с помощью send из другого рактора).

r = Ractor.new do
  v1 = Ractor.receive
  puts "Received: #{v1}"
end
r.send('message1')
r.take
# Here will be printed: "Received: message1"

В качестве альтернативы можно использовать частный метод экземпляра receive:

r = Ractor.new do
  v1 = receive
  puts "Received: #{v1}"
end
r.send('message1')
r.take
# This prints: "Received: message1"

Метод блокируется, если очередь пуста.

r = Ractor.new do
  puts "Before first receive"
  v1 = Ractor.receive
  puts "Received: #{v1}"
  v2 = Ractor.receive
  puts "Received: #{v2}"
end
wait
puts "Still not received"
r.send('message1')
wait
puts "Still received only one"
r.send('message2')
r.take

Вывод:

Before first receive
Still not received
Received: message1
Still received only one
Received: message2

Если для рактора был вызван close_incoming, метод вызывает Ractor::ClosedError, если больше нет сообщений в очереди ввода:

Ractor.new do
  close_incoming
  receive
end
wait
# in `receive': The incoming port is already closed => #<Ractor:#2 test.rb:1 running> (Ractor::ClosedError)

# File ractor.rb, line 509
def self.receive_if &b
  Primitive.ractor_receive_if b
end

Получить только определённое сообщение.

Вместо Ractor.receive, Ractor.receive_if может получить шаблон (или любой фильтр) в блоке и вы можете выбрать сообщения для принятия, которые доступны в очереди входящих сообщений рактора.

r = Ractor.new do
  p Ractor.receive_if{|msg| msg.match?(/foo/)} #=> "foo3"
  p Ractor.receive_if{|msg| msg.match?(/bar/)} #=> "bar1"
  p Ractor.receive_if{|msg| msg.match?(/baz/)} #=> "baz2"
end
r << "bar1"
r << "baz2"
r << "foo3"
r.take

Этот вывод:

foo3
bar1
baz2

Если блок возвращает истинное значение, сообщение удаляется из очереди входящих сообщений и возвращается. В противном случае сообщение остается в очереди входящих сообщений, и проверяются следующие сообщения с помощью данного блока.

Если сообщений в очереди входящих сообщений больше нет, метод будет заблокирован до появления новых сообщений.

Если блок прерывается break/return/исключением/броском, сообщение удаляется из очереди входящих сообщений так, как если бы было возвращено истинное значение.

r = Ractor.new do
  val = Ractor.receive_if{|msg| msg.is_a?(Array)}
  puts "Received successfully: #{val}"
end

r.send(1)
r.send('test')
wait
puts "2 non-matching sent, nothing received"
r.send([1, 2, 3])
wait

Выводит:

2 non-matching sent, nothing received
Received successfully: [1, 2, 3]

Обратите внимание, что вы не можете вызвать receive/receive_if рекурсивно в данном блоке. Вы не должны выполнять какие-либо задачи в блоке, кроме фильтрации сообщений.

Ractor.current << true
Ractor.receive_if{|msg| Ractor.receive}
#=> `receive': can not call receive/receive_if recursively (Ractor::Error)

# File ractor.rb, line 358
def self.select(*ractors, yield_value: yield_unspecified = true, move: false)
  raise ArgumentError, 'specify at least one ractor or `yield_value`' if yield_unspecified && ractors.empty?

  if ractors.delete Ractor.current
    do_receive = true
  else
    do_receive = false
  end

  __builtin_ractor_select_internal ractors, do_receive, !yield_unspecified, yield_value, move
end

Ожидание, пока любой рактор получит что-то в свой исходящий порт, чтение из этого рактора, а затем возвращение этого рактора и полученного объекта.

r1 = Ractor.new {Ractor.yield 'from 1'}
r2 = Ractor.new {Ractor.yield 'from 2'}

r, obj = Ractor.select(r1, r2)

puts "received #{obj.inspect} from #{r.inspect}"
# Prints: received "from 1" from #<Ractor:#2 test.rb:1 running>
# But could just as well print "from r2" here, either prints could be first.

Если один из переданных ракторов является текущим рактором и он выбран, r будет содержать символ :receive вместо объекта рактора.

r1 = Ractor.new(Ractor.current) do |main|
  main.send 'to main'
  Ractor.yield 'from 1'
end
r2 = Ractor.new do
  Ractor.yield 'from 2'
end

r, obj = Ractor.select(r1, r2, Ractor.current)
puts "received #{obj.inspect} from #{r.inspect}"
# Could print: received "to main" from :receive

Если yield_value указано, это значение может быть передано, если другой рактор вызывает take. В этом случае возвращается пара [:yield, nil]:

r1 = Ractor.new(Ractor.current) do |main|
  puts "Received from main: #{main.take}"
end

puts "Trying to select"
r, obj = Ractor.select(r1, Ractor.current, yield_value: 123)
wait
puts "Received #{obj.inspect} from #{r.inspect}"

Это выведет:

Trying to select
Received from main: 123
Received nil from :yield

move логический флаг определяет, будет ли копироваться возвращаемое значение (по умолчанию) или перемещаться.

# File ractor.rb, line 784
def self.shareable? obj
  __builtin_cexpr! %q{
    RBOOL(rb_ractor_shareable_p(obj));
  }
end

Проверка, является ли объект разделяемым ракторами.

Ractor.shareable?(1)            #=> true -- numbers and other immutable basic values are frozen
Ractor.shareable?('foo')        #=> false, unless the string is frozen due to # frozen_string_literal: true
Ractor.shareable?('foo'.freeze) #=> true

См. также раздел «Разделяемые и неразделяемые объекты» в документации класса Ractor.

# File ractor.rb, line 643
def self.yield(obj, move: false)
  __builtin_cexpr! %q{
    ractor_yield(ec, rb_ec_ractor_ptr(ec), obj, move)
  }
end

Отправить сообщение в исходящий порт текущего рактора для приема take.

r = Ractor.new {Ractor.yield 'Hello from ractor'}
puts r.take
# Prints: "Hello from ractor"

Этот метод блокирующий и вернет только когда кто-то получит отправленное сообщение.

r = Ractor.new do
  Ractor.yield 'Hello from ractor'
  puts "Ractor: after yield"
end
wait
puts "Still not taken"
puts r.take

Это выведет:

Still not taken
Hello from ractor
Ractor: after yield

Если исходящий порт был закрыт с помощью close_outgoing, метод вызовет:

r = Ractor.new do
  close_outgoing
  Ractor.yield 'Hello from ractor'
end
wait
# `yield': The outgoing-port is already closed (Ractor::ClosedError)

Значение аргумента move такое же, как и для send.

# File ractor.rb, line 838
def [](sym)
  Primitive.ractor_local_value(sym)
end

Получить значение из локального хранилища рактора.

# File ractor.rb, line 843
def []=(sym, val)
  Primitive.ractor_local_value_set(sym, val)
end

Установить значение в локальном хранилище рактора.

# File ractor.rb, line 749
def close_incoming
  __builtin_cexpr! %q{
    ractor_close_incoming(ec, RACTOR_PTR(self));
  }
end

Закрывает входящий порт и возвращает значение, указывающее, был ли он уже закрыт. Все последующие попытки Ractor.receive в ракторе и send в рактор завершатся ошибкой Ractor::ClosedError.

r = Ractor.new {sleep(500)}
r.close_incoming  #=> false
r.close_incoming  #=> true
r.send('test')
# Ractor::ClosedError (The incoming-port is already closed)

# File ractor.rb, line 767
def close_outgoing
  __builtin_cexpr! %q{
    ractor_close_outgoing(ec, RACTOR_PTR(self));
  }
end

Закрывает исходящий порт и возвращает значение, указывающее, был ли он уже закрыт. Все последующие попытки Ractor.yield в ракторе и take из рактора завершатся ошибкой Ractor::ClosedError.

r = Ractor.new {sleep(500)}
r.close_outgoing  #=> false
r.close_outgoing  #=> true
r.take
# Ractor::ClosedError (The outgoing-port is already closed)

# File ractor.rb, line 729
def name
  __builtin_cexpr! %q{RACTOR_PTR(self)->name}
end

Имя, заданное в Ractor.new, или nil.

# File ractor.rb, line 599
def send(obj, move: false)
  __builtin_cexpr! %q{
    ractor_send(ec, RACTOR_PTR(self), obj, move)
  }
end

Отправить сообщение в очередь входящих сообщений рактора для обработки методом Ractor.receive.

r = Ractor.new do
  value = Ractor.receive
  puts "Received #{value}"
end
r.send 'message'
# Prints: "Received: message"

Метод является неблокирующим (возвращает немедленно, даже если рактор не готов к приёму):

r = Ractor.new {sleep(5)}
r.send('test')
puts "Sent successfully"
# Prints: "Sent successfully" immediately

Попытка отправить сообщение рактору, который уже завершил свою работу, вызовет ошибку Ractor::ClosedError.

r = Ractor.new {}
r.take
p r
# "#<Ractor:#6 (irb):23 terminated>"
r.send('test')
# Ractor::ClosedError (The incoming-port is already closed)

Если для рактора был вызван метод close_incoming, метод также вызовет ошибку Ractor::ClosedError.

r =  Ractor.new do
  sleep(500)
  receive
end
r.close_incoming
r.send('test')
# Ractor::ClosedError (The incoming-port is already closed)
# The error is raised immediately, not when the ractor tries to receive

Если obj является неразделяемым, по умолчанию он будет скопирован в принимающий рактор путём глубокого клонирования. Если параметр move: true передан, объект переместится в принимающий рактор и станет недоступным для отправителя.

r = Ractor.new {puts "Received: #{receive}"}
msg = 'message'
r.send(msg, move: true)
r.take
p msg

Это выведет:

Received: message
in `p': undefined method `inspect' for #<Ractor::MovedObject:0x000055c99b9b69b8>

Все ссылки на объект и его части станут недействительными для отправителя.

r = Ractor.new {puts "Received: #{receive}"}
s = 'message'
ary = [s]
copy = ary.dup
r.send(ary, move: true)

s.inspect
# Ractor::MovedError (can not send any methods to a moved object)
ary.class
# Ractor::MovedError (can not send any methods to a moved object)
copy.class
# => Array, it is different object
copy[0].inspect
# Ractor::MovedError (can not send any methods to a moved object)
# ...but its item was still a reference to `s`, which was moved

Если объект разделяемый, move: true не повлияет на него:

r = Ractor.new {puts "Received: #{receive}"}
s = 'message'.freeze
r.send(s, move: true)
s.inspect #=> "message", still available

# File ractor.rb, line 710
def take
  __builtin_cexpr! %q{
    ractor_take(ec, RACTOR_PTR(self))
  }
end

Получить сообщение из исходящего порта рактора, помещенное туда методом Ractor.yield или при завершении работы рактора.

r = Ractor.new do
  Ractor.yield 'explicit yield'
  'last value'
end
puts r.take #=> 'explicit yield'
puts r.take #=> 'last value'
puts r.take # Ractor::ClosedError (The outgoing-port is already closed)

Тот факт, что последнее значение также отправляется в исходящий порт, означает, что take может использоваться в качестве аналога метода Thread#join («просто подождите, пока рактор завершит работу»). Однако он вызовет ошибку, если кто-то уже получил это сообщение.

Если исходящий порт был закрыт с помощью close_outgoing, метод вызовет ошибку Ractor::ClosedError.

r = Ractor.new do
  sleep(500)
  Ractor.yield 'Hello from ractor'
end
r.close_outgoing
r.take
# Ractor::ClosedError (The outgoing-port is already closed)
# The error would be raised immediately, not when ractor will try to receive

Если в Ractor произошла незахваченная ошибка, она распространяется методом take как Ractor::RemoteError.

r = Ractor.new {raise "Something weird happened"}

begin
  r.take
rescue => e
  p e              #  => #<Ractor::RemoteError: thrown by remote Ractor.>
  p e.ractor == r  # => true
  p e.cause        # => #<RuntimeError: Something weird happened>
end

Ractor::ClosedError является потомком StopIteration, поэтому завершение работы рактора прервёт любые циклы, получающие это сообщение, без распространения ошибки:

r = Ractor.new do
  3.times {|i| Ractor.yield "message #{i}"}
  "finishing"
end

loop {puts "Received: " + r.take}
puts "Continue successfully"

Это выведет:

Received: message 0
Received: message 1
Received: message 2
Received: finishing
Continue successfully

# File ractor.rb, line 441
        def receive
  __builtin_cexpr! %q{
    ractor_receive(ec, rb_ec_ractor_ptr(ec))
  }
end

то же, что и Ractor.receive

# File ractor.rb, line 514
        def receive_if &b
  Primitive.ractor_receive_if b
end

то же, что и Ractor.receive_if