Ractor

Ractor

Ractor.new은 다른 ractor들과 병렬로 실행될 수 있는 새 Ractor를 만들어요.

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

Ractor는 서로 모든 객체를 공유하지 않아요. 여기서 두 가지 주요 이점이 나와요. 하나는 ractor들 사이에서 데이터 레이스나 레이스 컨디션 같은 스레드 안전성 우려가 불가능하다는 것이고, 다른 하나는 병렬성(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 the Thread until our default port gets sent a message
  puts "I am in Ractor! a=#{a_in_ractor}"
end
r.send(a)  # pass it
r.join
# Here, "I am in Ractor! a=1" is printed

게다가 Ractor.new에 전달된 어떤 인자든 블록에 전달되고, 마치 Ractor.receive로 받은 것처럼 거기서 사용 가능해요. 그리고 마지막 블록 값은 Ractor#value로 받을 수 있어요.

출처: Ruby 4.0 API

본문

공유 가능·불가능 객체

객체를 ractor로 보낼 때는 그 객체가 shareable(공유 가능)인지 unshareable(공유 불가)인지 이해하는 게 중요해요. 대부분의 Ruby 객체는 unshareable이에요. frozen 객체조차 (인스턴스 변수를 통해) unfrozen 객체를 포함하면 unshareable일 수 있어요.

Shareable 객체는 스레드 안전성을 훼손하지 않으면서 여러 ractor가 동시에 사용할 수 있는 객체예요. 숫자, 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

shareable 객체를 send로 보내면 추가 처리가 일어나지 않고, 두 ractor 모두가 사용할 수 있게 돼요. unshareable 객체를 보내면 복사(copied)되거나 이동(moved)될 수 있어요. 복사가 기본이고, 구조의 non-shareable 부분을 딥 클로닝(Object#clone)해서 객체를 완전히 복사해요.

data = ['foo'.dup, '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.join
puts "Outside  : #{data.object_id}, #{data[0].object_id}, #{data[1].object_id}"

이 출력은 대략 이렇게 나와요.

In ractor: 8, 16, 24
Outside  : 32, 40, 24

배열과 배열 안의 non-frozen 문자열의 객체 id가 ractor 안에서 다른 객체이기 때문에 바뀐 점에 유의하세요. 두 번째 배열 요소인 shareable frozen 문자열은 같은 객체예요.

객체 딥 클로닝은 느릴 수 있고, 때로는 불가능할 수 있어요. 대신 보낼 때 move: true를 쓸 수 있어요. 이렇게 하면 unshareable 객체를 받는 ractor로 이동시켜서, 보내는 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.join
puts "Outside: moved? #{Ractor::MovedObject === data}"
puts "Outside: #{data.inspect}"

이 출력은 이렇게 나와요.

In ractor: 100, 120
Outside: moved? true
test.rb:9:in `method_missing': can not send any methods to a moved object (Ractor::MovedError)

moved 객체에서는 inspect조차, __id__ 같은 더 기본적인 메서드조차 접근할 수 없다는 점을 알아 두세요.

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

다른 ractor에서 shareable 객체의 인스턴스 변수에 접근(가져오기)하는 것은, 그 변수의 값이 shareable이 아니면 금지돼요. 모듈/클래스는 shareable이지만 값이 shareable이 아닌 인스턴스 변수를 가질 수 있기 때문에 이런 상황이 생길 수 있어요. non-main ractor에서는 클래스/모듈에 인스턴스 변수를 설정하는 것도(값이 shareable이어도) 금지돼요.

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.join
# I see C
# can not access instance variables of classes/modules from non-main Ractors (RuntimeError)

Ractor는 shareable이면 상수에 접근할 수 있어요. main Ractor만 non-shareable 상수에 접근할 수 있어요.

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

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

# Consider the same C class from above

r = Ractor.new do
  puts "I see #{C}"
  puts "I can't see #{C.tricky}"
end
r.join
# I see C
# can not access instance variables of classes/modules from non-main Ractors (RuntimeError)

# shareable_constant_value pragma에 대한 설명은 Comments 문법 설명의 shareable_constant_value pragma 부분도 함께 참고하세요.

Ractor vs 스레드

각 ractor는 자신만의 main Thread를 가져요. ractor 내부에서 새 스레드를 만들 수 있어요(CRuby에서는 이 ractor의 다른 스레드들과 GVL을 공유해요).

r = Ractor.new do
  a = 1
  Thread.new {puts "Thread in ractor: a=#{a}"}.join
end
r.join
# Here "Thread in ractor: a=1" will be printed

코드 예제에 대한 참고

아래 예제들에서 때로는 ractor들이 진행되거나 끝나길 기다리기 위해 다음 메서드를 사용해요.

def wait
  sleep(0.1)
end

이것은 오직 시연 목적이고 실제 코드에서 써서는 안 돼요. 대부분 join이 ractor들이 끝나길 기다리는 데, Ractor.receive가 메시지를 기다리는 데 쓰여요.

참고

자세한 내용은 Ractor 설계 문서를 참고하세요.

Public Class Methods

[](sym)

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

[]=(sym, val)

현재 Ractor의 ractor-로컬 저장소에 값을 설정해요.

count()

현재 실행 중이거나 블로킹(대기) 중인 ractor의 수를 돌려줘요.

Ractor.count                   #=> 1
r = Ractor.new(name: 'example') { Ractor.receive }
Ractor.count                   #=> 2 (main + example ractor)
r << 42                        # r's Ractor.receive will resume
r.join                         # wait for r's termination
Ractor.count                   #=> 1

current()

현재 실행 중인 Ractor를 돌려줘요.

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

main()

main ractor를 돌려줘요.

main?()

현재 ractor가 main ractor이면 true를 돌려줘요.

make_shareable(obj, copy: false) → shareable_obj

obj를 ractor 간 공유 가능하게 만들어요. obj와 그것이 참조하는 모든 객체는, 이미 shareable이 아니라면 frozen돼요. copy 키워드가 true이면 객체들을 frozen하기 전에 복사하고, 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 클래스 문서의 "Shareable and unshareable objects" 섹션도 참고하세요.

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

args와 블록으로 새 Ractor를 만들어요. 주어진 블록(Proc)은 격리(isolate)돼서 바깥 변수에 접근할 수 없어요. 블록 안에서 self는 현재 Ractor를 가리켜요.

r = Ractor.new { puts "Hi, I am #{self.inspect}" }
r.join
# Prints "Hi, I am #<Ractor:#2 test.rb:1 running>"

전달된 어떤 args든 블록 인자로 전파되는데, send/Ractor.receive로 보내는 객체와 같은 규칙을 따라요. args의 인자가 shareable이 아니면 복사돼요(딥 클로닝으로, 비효율적일 수 있음).

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.join
# Prints:
#   Passing: [1, 2, 3] (#280)
#   Received: [1, 2, 3] (#300)

디버깅 목적으로 ractor 이름을 설정할 수 있어요.

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

receive → obj

현재 ractor의 기본 포트에서 메시지를 받아요.

recv

receive의 별칭이에요.

select(*ractors_or_ports) → [ractor or port, obj]

주어진 포트 중 하나가 메시지를 받을 때까지 현재 Thread를 블로킹해요. 첫 요소가 Port, 둘째가 받은 객체인 2-요소 배열을 돌려줘요. 이 메서드는 Ractor 객체 자체도 받을 수 있는데, 그 경우에는 하나가 종료될 때까지 기다렸다가 첫 요소가 ractor이고 둘째가 그 종료 값인 2-요소 배열을 돌려줘요.

p1, p2 = Ractor::Port.new, Ractor::Port.new
ps = [p1, p2]
rs = 2.times.map do |i|
  Ractor.new(ps.shift, i) do |p, i|
    sleep rand(0.99)
    p.send("r#{i}")
    sleep rand(0.99)
    "r#{i} done"
  end
end

waiting_on = [p1, p2, *rs]
until waiting_on.empty?
  received_on, obj = Ractor.select(*waiting_on)
  waiting_on.delete(received_on)
  puts obj
end

# r0
# r1
# r1 done
# r0 done

다음 예제는 ractors.map(&:value)와 거의 동등한데, 배열 요소 순서대로 종료를 기다리는 대신 어느 ractor든 종료되면 스레드가 풀린다는 점이 달라요.

values = []
until ractors.empty?
  r, val = Ractor.select(*ractors)
  ractors.delete(r)
  values << val
end

shareable?(obj) → true | false

객체가 ractor로 공유 가능한지 검사해요.

Ractor.shareable?(1)            #=> true -- numbers are shareable
Ractor.shareable?('foo')        #=> false, unless the string is frozen due to # frozen_string_literal: true
Ractor.shareable?('foo'.freeze) #=> true

Ractor 클래스 문서의 "Shareable and unshareable objects" 섹션도 참고하세요.

shareable_lambda(self: nil){} → shareable lambda

Ractor.shareable_proc과 같지만 lambda Proc을 돌려줘요.

shareable_proc(self: nil){} → shareable proc

주어진 블록의 Proc의 shareable 복사본을 돌려줘요. Proc의 self 값은 self: 키워드로 전달된 값으로 대체되고, 주어지지 않으면 nil이 돼요. shareable Proc에서는 바깥 변수 접근이 금지돼요.

a = 42
Ractor.shareable_proc{ p a }
#=> can not isolate a Proc because it accesses outer variables (a). (ArgumentError)

Proc의 self 값은 shareable 객체여야 해요.

Ractor.shareable_proc(self: self){}
#=> self should be shareable: main (Ractor::IsolationError)

store_if_absent(key){ init_block }

해당 ractor-로컬 값이 설정되지 않았으면 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

Public Instance Methods

<<

default_port로 메시지를 보내는 단축 형태(send와 동등).

[](sym)

현재 Ractor의 ractor-로컬 저장소에서 값을 가져와요. 폐기 예정입니다. Ractor.[]를 쓰세요.

[]=(sym, val)

현재 Ractor의 ractor-로컬 저장소에 값을 설정해요. 폐기 예정입니다. Ractor.[]=를 쓰세요.

close → true | false

기본 포트를 닫아요. 포트 닫기는 그 포트를 만든 ractor만 할 수 있어요. 따라서 receiver가 현재 ractor여야 해요.

default_port → port object

Ractor의 기본 포트를 돌려줘요.

inspect()

join → self

Ractor의 종료를 기다려요. Ractor가 abort됐으면(처리되지 않은 예외로 종료), 그 예외가 현재 ractor에서 발생해요.

Ractor.new{}.join #=> ractor

Ractor.new{ raise "foo" }.join
#=> raises the exception "foo (RuntimeError)"

monitor(port) → self

port를 이 ractor의 모니터링 포트로 등록해요. ractor가 종료되면 그 포트가 Symbol 객체를 받아요.

• ractor가 처리되지 않은 예외 없이 종료하면 :exited가 보내져요.
• ractor가 처리되지 않은 예외로 종료하면 :aborted가 보내져요.

r = Ractor.new { some_task() }
r.monitor(port = Ractor::Port.new)
port.receive #=> :exited and r is terminated

r = Ractor.new { raise "foo" }
r.monitor(port = Ractor::Port.new)
port.receive #=> :aborted and r is terminated by the RuntimeError "foo"

name()

Ractor.new에서 설정된 이름을 돌려주거나, 없으면 nil을 돌려줘요.

recv()

receive의 별칭이에요.

send(msg, move: false) → self

ractor의 default_port에 Port#send를 하는 것과 동등해요.

to_s()

unmonitor(port) → self

이 ractor의 모니터링 포트에서 port를 등록 해제해요.

value → obj

ractor가 완료되길 기다렸다가 그 값을 돌려주거나, ractor를 종료시킨 예외를 발생시켜요. 종료 값은 호출하는 Ractor로 이동(moved)돼요. 따라서 최대 1개의 ractor가 다른 ractor의 종료 값을 받을 수 있어요.

r = Ractor.new{ [1, 2] }
r.value #=> [1, 2] (unshareable object)

Ractor.new(r){|r| r.value} #=> Ractor::Error

Private Instance Methods

receive()

Ractor.receive와 같아요.