класс Ractor

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

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

# File ractor.rb, line 814
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
  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') {}
p r
#=> #<Ractor:#3 my ractor test.rb:1 terminated>

# File ractor.rb, line 419
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 497
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 346
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 773
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 630
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 827
def [](sym)
  Primitive.ractor_local_value(sym)
end

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

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

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

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

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

# File ractor.rb, line 586
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`, объект перемещается в рактор и становится недоступным для отправителя.

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` не повлияет на него:

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

# File ractor.rb, line 697
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 430
        def receive
  __builtin_cexpr! %q{
    ractor_receive(ec, rb_ec_ractor_ptr(ec))
  }
end

то же самое, что и Ractor.receive