класс Ractor

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

Возвращает общее количество активных ракторов.

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 till r will finish
Ractor.count                   #=> 1

# File ractor.rb, line 273
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 833
def self.main
  __builtin_cexpr! %q{
    rb_ractor_self(GET_VM()->ractor.main_ractor);
  }
end

Возвращает главный рактор.

# File ractor.rb, line 810
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 = ['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 262
def self.new(*args, name: nil, &block)
  b = block # TODO: builtin bug
  raise ArgumentError, "must be called with a block" unless block
  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 рактора может быть установлено для отладки:

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

# File ractor.rb, line 415
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
# Here will be printed: "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 493
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/исключения/throw, сообщение удаляется из очереди входящих сообщений так же, как если бы было возвращено истинное значение.

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 342
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?

  __builtin_cstmt! %q{
    const VALUE *rs = RARRAY_CONST_PTR_TRANSIENT(ractors);
    VALUE rv;
    VALUE v = ractor_select(ec, rs, RARRAY_LENINT(ractors),
                            yield_unspecified == Qtrue ? Qundef : yield_value,
                            (bool)RTEST(move) ? true : false, &rv);
    return rb_ary_new_from_args(2, rv, v);
  }
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>

Если один из заданных ракторов — текущий рактор и он был бы выбран, 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}"
# Prints: received "to main" from :receive

Если yield_value указан, это значение может быть возвращено, если другой Ractor вызывает 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 769
def self.shareable? obj
  __builtin_cexpr! %q{
    rb_ractor_shareable_p(obj) ? Qtrue : Qfalse;
  }
end

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

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

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

# File ractor.rb, line 626
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 823
def [](sym)
  Primitive.ractor_local_value(sym)
end

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

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

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

# File ractor.rb, line 733
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 752
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 712
def name
  __builtin_cexpr! %q{RACTOR_PTR(self)->name}
end

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

# File ractor.rb, line 582
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 would be raised immediately, not when ractor will try 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 693
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 426
        def receive
  __builtin_cexpr! %q{
    ractor_receive(ec, rb_ec_ractor_ptr(ec))
  }
end

также как Ractor.receive