Ractor 클래스

Ractor는 Ruby를 위한 액터 모델(Actor-model) 추상화로, 스레드 안전한 병렬 실행(parallel execution)을 제공해요. Ractor.new로 새 Ractor를 만들면 병렬로 실행될 수 있어요.

출처: Ruby 3.3 API

본문

기본 예제

# The simplest ractor
r = Ractor.new {puts "I am in Ractor!"}
r.take # wait for it to finish
# Here, "I am in Ractor!" is printed

Ractor들은 서로 모든 객체를 공유하지 않아요. 여기에는 두 가지 큰 장점이 있어요. 첫째, Ractor들 사이에서는 데이터 레이스(data-races)나 레이스 컨디션(race-conditions) 같은 스레드 안전 문제가 생길 수 없어요. 둘째는 병렬성(parallelism)이에요.

이를 위해 Ractor 간의 객체 공유를 제한해요. 예를 들어 스레드와 달리 Ractor는 다른 Ractor에 있는 모든 객체에 접근할 수 없어요. 바깥 스코프의 변수로 평소 접근 가능한 객체조차 Ractor 간에 사용하는 게 금지돼요.

a = 1
r = Ractor.new {puts "I am in Ractor! a=#{a}"}
# fails immediately with
# ArgumentError (can not isolate a Proc because it accesses outer variables (a).)

객체를 명시적으로 공유해야 해요:

a = 1
r = Ractor.new(a) { |a1| puts "I am in Ractor! a=#{a1}"}

CRuby(기본 구현)에서는 GVL(Global Virtual Machine Lock)이 Ractor마다 따로 잡혀서, Ractor들은 서로를 막지 않고 병렬로 실행될 수 있어요. 이는 CRuby에서 스레드가 그런 상황과는 달라요.

공유 상태에 접근하는 대신, 객체를 메시지로 보내고 받으면서 Ractor 안팎으로 전달해야 해요.

a = 1
r = Ractor.new do
  a_in_ractor = receive # receive blocks until somebody passes a message
  puts "I am in Ractor! a=#{a_in_ractor}"
end
r.send(a)  # pass it
r.take
# Here, "I am in Ractor! a=1" is printed

메시지를 주고받는 메서드에는 두 쌍이 있어요:

Ractor#send와 Ractor.receive — 송신자가 수신자를 아는 경우(밀어 넣기, push)
Ractor.yield와 Ractor#take — 수신자가 송신자를 아는 경우(당기기, pull)

추가로, Ractor.new에 넘긴 인자들은 블록으로 전달되어 Ractor.receive로 받은 것처럼 쓸 수 있고, 블록의 마지막 값은 Ractor.yield로 보낸 것처럼 Ractor 밖으로 나가요.

Ractor들은 incoming port(들어오는 포트)로 메시지를 받고, outgoing port(나가는 포트)로 메시지를 보내요. 각 포트는 Ractor#close_incoming과 Ractor#close_outgoing으로 끌 수 있고, Ractor가 종료되면 포트는 자동으로 닫혀요.

고전적인 ping-pong 예제:

server = Ractor.new(name: "server") do
  puts "Server starts: #{self.inspect}"
  puts "Server sends: ping"
  Ractor.yield 'ping'                       # The server doesn't know the receiver and sends to whoever interested
  received = Ractor.receive                 # The server doesn't know the sender and receives from whoever sent
  puts "Server received: #{received}"
end

client = Ractor.new(server) do |srv|        # The server is sent to the client, and available as srv
  puts "Client starts: #{self.inspect}"
  received = srv.take                       # The client takes a message from the server
  puts "Client received from #{srv.inspect}: #{received}"
  puts "Client sends to #{srv.inspect}: pong"
  srv.send 'pong'                           # The client sends a message to the server
end

[client, server].each(&:take)               # Wait until they both finish

공유 가능한 객체와 불가능한 객체 (Shareable and unshareable objects)

객체를 Ractor로 보내고 받을 때 그 객체가 공유 가능(shareable)한지 불가능(unshareable)한지 이해하는 게 중요해요. 대부분의 Ruby 객체는 공유 불가능한 객체예요. 동결(frozen)된 객체라도 그 인스턴스 변수에 동결되지 않은 객체를 담고 있으면 공유 불가능할 수 있어요.

공유 가능한 객체는 숫자, true, false처럼 여러 스레드가 스레드 안전을 해치지 않고 쓸 수 있는 객체예요. Ractor.shareable?로 확인할 수 있고, Ractor.make_shareable은 이미 공유 가능하지 않으면 그 객체를 공유 가능하게 만들려고 시도하며, 할 수 없으면 오류를 줘요.

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

ary = ['hello', 'world']
ary.frozen?                 #=> false
ary[0].frozen?              #=> false
Ractor.make_shareable(ary)
ary.frozen?                 #=> true
ary[0].frozen?              #=> true
ary[1].frozen?              #=> true

공유 가능한 객체를 보낼 때(send 또는 Ractor.yield)는 추가 처리가 없어요. 그냥 두 Ractor 모두에서 쓸 수 있게 되죠. 공유 불가능한 객체를 보낼 때는 *복사(copied)*되거나 *이동(moved)*될 수 있어요. 전자가 기본값이고, 구조의 비공유 부분을 깊은 복제(Object#clone)로 완전히 복사해요.

data = ['foo', 'bar'.freeze]
r = Ractor.new do
  data2 = Ractor.receive
  puts "In ractor: #{data2.object_id}, #{data2[0].object_id}, #{data2[1].object_id}"
end
r.send(data)
r.take
puts "Outside  : #{data.object_id}, #{data[0].object_id}, #{data[1].object_id}"

객체를 깊게 복제하는 건 느릴 수 있고 가끔은 불가능할 수도 있어요. 대신 보낼 때 move: true를 쓸 수 있는데, 이 경우 공유 불가능한 객체를 받는 Ractor로 *이동(move)*시켜서 보낸 Ractor에서는 접근할 수 없게 해요.

data = ['foo', 'bar']
r = Ractor.new do
  data_in_ractor = Ractor.receive
  puts "In ractor: #{data_in_ractor.object_id}, #{data_in_ractor[0].object_id}"
end
r.send(data, move: true)
r.take
puts "Outside: moved? #{Ractor::MovedObject === data}"
puts "Outside: #{data.inspect}"  # raises Ractor::MovedError

이동된 객체에는 inspect(그리고 __id__ 같은 더 기본적인 메서드)조차 접근할 수 없어요.

Class와 Module 객체는 공유 가능해서 클래스/모듈 정의는 Ractor 간에 공유돼요. Ractor 객체도 공유 가능해요. 공유 가능한 객체에 대한 모든 연산은 스레드 안전해서 그 성질이 유지돼요. Ruby에서는 가변(mutable) 공유 가능 객체를 정의할 수 없지만, C 확장은 도입할 수 있어요.

다른 Ractor에서 공유 가능한 객체의 인스턴스 변수 값이 공유 가능하지 않다면 그 변수에 접근(가져오기)하는 건 금지돼요. 모듈/클래스는 공유 가능하지만 그 인스턴스 변수 값은 공유 가능하지 않을 수 있기 때문이에요. 메인이 아닌 Ractor에서는 클래스/모듈에 인스턴스 변수를 설정하는 것도 금지돼요(값이 공유 가능해도).

class C
  class << self
    attr_accessor :tricky
  end
end

C.tricky = "unshareable".dup

r = Ractor.new(C) do |cls|
  puts "I see #{cls}"
  puts "I can't see #{cls.tricky}"
  cls.tricky = true # doesn't get here, but this would also raise an error
end
r.take
# I see C
# can not access instance variables of classes/modules from non-main Ractors (RuntimeError)

Ractor는 공유 가능한 상수(constant)에는 접근할 수 있어요. 메인 Ractor만 비공유 상수에 접근할 수 있어요.

GOOD = 'good'.freeze
BAD = 'bad'.dup

r = Ractor.new do
  puts "GOOD=#{GOOD}"
  puts "BAD=#{BAD}"
end
r.take
# GOOD=good
# can not access non-shareable objects in constant Object::BAD by non-main Ractor. (NameError)

Comments 문법 설명의 # shareable_constant_value 프라그마 설명도 함께 참고하세요.

Ractor vs 스레드 (Ractors vs threads)

각 Ractor는 자신만의 메인 Thread를 가져요. Ractor 안에서 새 스레드를 만들 수 있고(CRuby에서는 이 Ractor의 다른 스레드들과 GVL을 공유해요), 그 스레드는 Ractor의 지역 변수에 접근할 수 있어요.

r = Ractor.new do
  a = 1
  Thread.new {puts "Thread in ractor: a=#{a}"}.join
end
r.take

코드 예제에 대한 참고 (Note on code examples)

아래 예제들에서는 현재 막히지 않은 Ractor들이 끝나길(또는 진행하기를) 기다리기 위해 가끔 이런 메서드를 써요. sleep(0.1)을 하는 wait 같은 것이요. 이는 오직 데모용이고 실제 코드에 쓰면 안 돼요. 대부분의 경우 Ractor가 끝나기를 기다리는 데 take를 써요.

Public Class Methods

count()

현재 실행 중이거나 블로킹(대기) 중인 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

current()

현재 실행 중인 Ractor를 반환해요.

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

main()

메인 Ractor를 반환해요.

make_shareable(obj, copy: false) → shareable_obj

obj를 Ractor 간에 공유 가능하게 만들어요. obj와 그것이 참조하는 모든 객체는 이미 공유 가능하지 않다면 동결(frozen)돼요. 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

new(args, name: nil) {|args| block } → ractor

args와 블록으로 새 Ractor를 만들어요. 주어진 블록(Proc)은 격리(isolate)돼서(바깥 변수에 접근할 수 없어요) 블록 안의 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로 보내는 객체와 같은 규칙으로 블록 인자에 전달돼요. 공유 불가능한 인자는 복사돼요(깊은 복제라 비효율적일 수 있어요).

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

Ractor의 name은 디버깅 목적으로 설정할 수 있어요.

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

receive → msg

현재 Ractor의 incoming port에서 메시지를 받아요(다른 Ractor가 send로 보낸 것). 큐가 비어 있으면 이 메서드는 블로킹돼요. 또한 private 인스턴스 메서드 receive로도 쓸 수 있어요.

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

Ractor에 close_incoming이 호출됐고 incoming 큐에 메시지가 더 없다면 Ractor::ClosedError를 발생시켜요. recv로도 불려요.

receive_if {|msg| block } → msg

특정 메시지만 받아요. 블록에 패턴(또는 어떤 필터)을 주고, incoming 큐에 있는 메시지 중 받아들일 걸 고를 수 있어요.

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

블록이 truthy 값을 반환하면 큐에서 메시지를 제거하고 반환해요. 그렇지 않으면 메시지는 큐에 남고 다음 메시지를 블록으로 검사해요. 큐에 메시지가 없으면 새 메시지가 올 때까지 블로킹돼요. 블록을 break/return/exception/throw로 이탈하면 truthy 값을 반환한 것처럼 메시지를 큐에서 제거해요. 블록 안에서 재귀적으로 receive/receive_if를 호출할 수 없고, 블록에서는 메시지 필터링 외의 작업을 하면 안 돼요.

recv

receive의 별칭이에요.

select(*ractors, [yield_value:, move: false]) → [ractor or symbol, obj]

아무 Ractor의 outgoing port에 뭔가가 있을 때까지 기다렸다가, 그 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}"

주어진 Ractor 중에 현재 Ractor가 있고 그것이 선택되면, r에는 Ractor 객체 대신 :receive 심볼이 들어가요. yield_value가 주어지고 다른 Ractor가 take를 호출하면 그 값이 yield될 수 있는데, 이 경우 [:yield, nil] 쌍을 반환해요. move 불리언 플래그는 yield된 값을 복사(기본)할지 이동할지 정해요.

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}"
# Received from main: 123
# Received nil from :yield

shareable?(obj) → true | false

객체가 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

yield(msg, move: false) → nil

현재 Ractor의 outgoing port에 메시지를 보내서 take가 받아가게 해요. 이 메서드는 블로킹이라, 누군가 보낸 메시지를 소비할 때까지 반환하지 않아요.

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

close_outgoing으로 outgoing port가 닫혀 있으면 Ractor::ClosedError를 발생시켜요. move 인자의 의미는 send와 같아요.

Public Instance Methods

<<

send의 별칭이에요.

ractor-로컬 저장소에서 값을 가져와요.

[]=(sym, val)

ractor-로컬 저장소에 값을 설정해요.

close_incoming → true | false

incoming port를 닫고 이미 닫혀 있었는지 반환해요. 이후 그 Ractor에서 Ractor.receive를 하거나 그 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)

close_outgoing → true | false

outgoing port를 닫고 이미 닫혀 있었는지 반환해요. 이후 그 Ractor에서 Ractor.yield를 하거나 그 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)

inspect() / to_s()

Ractor를 설명하는 문자열을 반환해요. to_s는 inspect의 별칭이에요.

name()

Ractor.new에서 설정한 이름, 또는 nil을 반환해요.

recv()

receive의 별칭이에요.

send(msg, move: false) → self / <<

Ractor의 incoming 큐에 메시지를 보내서 Ractor.receive가 받아가게 해요. 이 메서드는 논블로킹이에요(Ractor가 받을 준비가 안 되어 있어도 즉시 반환해요).

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

이미 실행을 마친 Ractor에 보내려 하면 Ractor::ClosedError가 발생해요. close_incoming이 호출된 Ractor에 보내도 Ractor::ClosedError가 발생해요.

obj가 공유 불가능하면 기본적으로 깊은 복제로 받는 Ractor에 복사돼요. move: true를 넘기면 객체가 받는 Ractor로 이동되고 송신자에게는 접근 불가능해져요. 이동된 객체와 그 일부에 대한 모든 참조는 송신자에게 무효가 돼요. 객체가 공유 가능하면 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:...> (Ractor::MovedError)

take → msg

Ractor의 outgoing port에서 메시지를 가져와요(Ractor.yield나 Ractor 종료 시 그곳에 놓인 것). 마지막 값도 outgoing port로 보내지므로, take는 Thread#join의 유사체("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)

close_outgoing으로 포트가 닫혀 있으면 Ractor::ClosedError를 발생시켜요. 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"

Private Instance Methods

receive()

Ractor.receive와 같아요. recv로도 불려요.

receive_if(&b)

Ractor.receive_if와 같아요.

Ractor 클래스