класс Ractor

# File ractor.rb, line 287
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 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

Получает входящее сообщение из очереди текущего порта Ractor, которое было отправлено туда методом 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{
    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 # 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

Если в ракторе возникает необработанное исключение, оно передаётся при вызове 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