class Ractor

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

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

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

устанавливает значение в локальном хранилище текущего 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 879
def self.main
  __builtin_cexpr! %q{
    rb_ractor_self(GET_VM()->ractor.main_ractor);
  }
end

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

# File ractor.rb, line 886
def self.main?
  __builtin_cexpr! %q{
    RBOOL(GET_VM()->ractor.main_ractor == rb_ec_ractor_ptr(ec))
  }
end

возвращает true, если текущий Ractor является основным

# File ractor.rb, line 826
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 совместно используемым между Ractor.

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 с аргументами args и блоком.

Переданный блок (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

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

Если для Ractor был вызван 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 можно передать блок с шаблоном (или любым фильтром), чтобы выбрать сообщения из входящей очереди вашего Ractor.

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

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

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.

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

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, это значение может быть передано, если другой 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 785
def self.shareable? obj
  __builtin_cexpr! %q{
    RBOOL(rb_ractor_shareable_p(obj));
  }
end

Проверяет, является ли объект совместно используемым для Ractor.

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 874
def self.store_if_absent(sym)
  Primitive.ractor_local_value_store_if_absent(sym)
end

Если соответствующее значение не задано, получает значение из init_block и сохраняет его потокобезопасным способом. Метод возвращает соответствующее сохранённое значение.

(1..10).map{
  Thread.new(it){|i|
    Ractor.store_if_absent(:s){ f(); i }
    #=> return stored value of key :s
  }
}.map(&:value).uniq.size #=> 1 and f() is called only once

# 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

Отправляет сообщение в исходящий порт текущего Ractor, откуда его можно получить с помощью 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 840
def [](sym)
  Primitive.ractor_local_value(sym)
end

получает значение из локального хранилища текущего Ractor. Устарел; вместо него используйте Ractor.[].

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

устанавливает значение в локальном хранилище текущего Ractor. Устарел; вместо него используйте Ractor.[]=.

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

Закрывает входящий порт и возвращает значение, указывающее, был ли он уже закрыт. Все последующие попытки вызвать Ractor.receive в Ractor или отправить сообщение в Ractor с помощью 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 768
def close_outgoing
  __builtin_cexpr! %q{
    ractor_close_outgoing(ec, RACTOR_PTR(self));
  }
end

Закрывает исходящий порт и возвращает значение, указывающее, был ли он уже закрыт. Все последующие попытки вызвать Ractor.yield в Ractor или получить сообщение из Ractor с помощью 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, откуда его можно получить с помощью Ractor.receive.

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

Метод неблокирующий (возвращает управление сразу, даже если Ractor не готов что-либо принять):

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

Попытка отправить сообщение в Ractor, выполнение которого уже завершилось, вызовет исключение 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)

Если для Ractor был вызван 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 является несовместно используемым, по умолчанию он будет скопирован в принимающий Ractor посредством глубокого клонирования. Если передан move: true, объект будет перемещён в принимающий Ractor и станет недоступен отправителю.

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, куда оно было помещено с помощью Ractor.yield или при завершении Ractor.

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 («просто подождать завершения Ractor»). Однако метод вызовет исключение, если это сообщение уже было получено кем-то другим.

Если исходящий порт был закрыт с помощью 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, поэтому завершение Ractor прерывает циклы, получающие это сообщение, не передавая исключение дальше:

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