Fiber 클래스

Fiber 클래스 (Fiber)

Fiber는 Ruby에서 가벼운 협력적 동시성(light weight cooperative concurrency)을 구현하기 위한 기본 요소예요. 기본적으로 스레드와 비슷하게 일시 정지·재개할 수 있는 코드 블록을 만드는 수단이죠. 가장 큰 차이는 절대 선점(preempt)되지 않고, 스케줄링을 VM이 아니라 프로그래머가 해야 한다는 점이에요.

다른 스택 없는(stackless) 가벼운 동시성 모델과 달리 각 파이버는 스택을 하나씩 가져요. 덕분에 파이버 블록 안의 깊게 중첩된 함수 호출에서도 파이버를 일시 정지할 수 있어요. 파이버 스택 크기 설정은 ruby(1) manpage를 참고하세요.

파이버를 만들면 자동으로 실행되지 않아요. 반드시 Fiber#resume 메서드로 명시적으로 실행해야 해요. 파이버 안에서 실행 중인 코드는 Fiber.yield를 호출해 제어를 호출자(Fiber#resume을 호출한 쪽)에게 돌려줄 수 있어요. yield하거나 종료할 때 파이버는 마지막으로 실행된 표현식의 값을 돌려줘요.

fiber = Fiber.new do
  Fiber.yield 1
  2
end

puts fiber.resume
puts fiber.resume
puts fiber.resume

출력:

1
2
FiberError: dead fiber called

Fiber#resume 메서드는 임의 개수의 파라미터를 받아요. 첫 resume 호출이면 블록 인자로 전달되고, 그 외에는 Fiber.yield 호출의 반환값이 돼요.

fiber = Fiber.new do |first|
  second = Fiber.yield first + 2
end

puts fiber.resume 10
puts fiber.resume 1_000_000
puts fiber.resume "The fiber will be dead before I can cause trouble"

출력:

12
1000000
FiberError: dead fiber called

출처: Ruby 3.3 API

본문

비블로킹 파이버 (Non-blocking Fibers)

비블로킹 파이버의 개념은 Ruby 3.0에서 도입됐어요. 비블로킹 파이버는 보통 파이버를 블로킹하는 연산(sleep이나 다른 프로세스/I/O 대기)에 도달하면 다른 파이버에 제어를 넘기고, 스케줄러가 블로킹 처리와 진행 가능할 때 이 파이버를 깨워 재개(resume)하게 해요.

파이버가 비블로킹으로 동작하려면 Fiber.new에 blocking: false(기본값)로 생성돼야 하고, Fiber.set_scheduler로 Fiber.scheduler가 설정돼야 해요. 현재 스레드에 Fiber.scheduler가 없으면 블로킹·비블로킹 파이버의 동작은 동일해요.

Ruby는 스케줄러 클래스를 제공하지 않아요. 사용자가 Fiber::Scheduler에 대응하는 클래스를 구현해야 해요. Fiber.schedule 메서드도 있는데, 주어진 블록을 비블로킹 방식으로 즉시 수행할 것으로 기대돼요. 실제 구현은 스케줄러에 달려 있어요.

::[] (Public Class Method)

Fiber[key] → value — key가 식별하는 파이버 스토리지 변수의 값을 돌려줘요. key는 반드시 심볼이어야 하고, 값은 Fiber#[]= 또는 Fiber#store로 설정돼요. (Fiber::[]= 참고)

::[]= (Public Class Method)

Fiber[key] = value — key가 식별하는 파이버 스토리지 변수에 값을 할당해요. 변수가 없으면 생성돼요. key는 Symbol이어야 하며, 아니면 TypeError가 발생해요. value가 nil이면 그 키를 삭제해요.

::blocking (Public Class Method)

blocking{|fiber| ...} → result — 블록이 지속되는 동안 파이버를 강제로 블로킹하게 만들어요. 블록의 결과를 돌려줘요.

::blocking? (Public Class Method)

blocking? → false 또는 1 — 현재 파이버가 비블로킹이면 false를 돌려줘요. 파이버는 Fiber.new에 blocking: false를 주거나 Fiber.schedule로 만들면 비블로킹이에요. 현재 파이버가 블로킹이면 메서드는 1을 돌려줘요. 미래에는 더 큰 정수가 반환될 수 있어요. false를 돌려줘도 현재 스레드에 Fiber.scheduler가 설정돼 있을 때만 동작이 달라져요.

::current (Public Class Method)

current → fiber — 현재 파이버를 돌려줘요. 파이버 컨텍스트에서 실행 중이 아니면 루트 파이버를 돌려줘요.

::current_scheduler (Public Class Method)

current_scheduler → obj 또는 nil — 현재 파이버가 비블로킹일 때만, Fiber.set_scheduler로 현재 스레드에 마지막으로 설정된 파이버 스케줄러를 돌려줘요.

::new (Public Class Method)

new(blocking: false, storage: true) { |*args| ... } → fiber — 새 Fiber를 만들어요. 처음엔 실행되지 않고 resume으로 재개할 수 있어요. 첫 resume의 인자는 블록에 전달돼요.

f = Fiber.new do |initial|
  current = initial
  loop do
    puts "current: #{current.inspect}"
    current = Fiber.yield
  end
end
f.resume(100)     # prints: current: 100
f.resume(1, 2, 3) # prints: current: [1, 2, 3]
f.resume          # prints: current: nil
# ... and so on ...

Fiber.new에 blocking: false를 주고 현재 스레드에 Fiber.scheduler가 정의돼 있으면 파이버는 비블로킹이 돼요.

storage를 지정하지 않으면 기본값은 현재 파이버의 스토리지 사본을 상속하는 것이고, 이는 storage: true와 같아요.

Fiber[:x] = 1
Fiber.new do
  Fiber[:x] # => 1
  Fiber[:x] = 2
end.resume
Fiber[:x] # => 1

주어진 storage가 nil이면 내부 스토리지를 지연 초기화해 빈 해시로 시작해요.

Fiber[:x] = "Hello World"
Fiber.new(storage: nil) do
  Fiber[:x] # nil
end

그 외에는 주어진 storage가 새 파이버의 스토리지로 사용되며, Hash 인스턴스여야 해요. storage: true를 명시하는 것은 현재 실험적이라 미래에 바뀔 수 있어요.

::schedule (Public Class Method)

schedule { |*args| ... } → fiber — 주어진 코드 블록을 별도의 비블로킹 파이버에서 즉시 실행할 것으로 기대되는 메서드예요.

puts "Go to sleep!"

Fiber.set_scheduler(MyScheduler.new)

Fiber.schedule do
  puts "Going to sleep"
  sleep(1)
  puts "I slept well"
end

puts "Wakey-wakey, sleepyhead"

MyScheduler가 제대로 구현돼 있다면 이 프로그램은 이렇게 출력돼요.

Go to sleep!
Going to sleep
Wakey-wakey, sleepyhead
...1 sec pause here...
I slept well

즉 파이버 안의 첫 블로킹 연산(sleep(1))에서 제어가 바깥 코드(메인 파이버)로 넘어가고, 그 실행이 끝나면 스케줄러가 블로킹된 모든 파이버를 제대로 재개해요. 위 설명은 기대 동작일 뿐, 실제 동작은 현재 스케줄러의 Fiber::Scheduler#fiber 구현에 달려 있어요. Ruby는 이 메서드의 특정 동작을 강제하지 않아요. 스케줄러가 설정돼 있지 않으면 RuntimeError(No scheduler is available!)를 발생시켜요.

::scheduler (Public Class Method)

scheduler → obj 또는 nil — Fiber.set_scheduler로 현재 스레드에 마지막으로 설정된 파이버 스케줄러를 돌려줘요. 설정돼 있지 않으면 nil(기본값)을 돌려주고, 비블로킹 파이버의 동작은 블로킹과 같아져요.

::set_scheduler (Public Class Method)

set_scheduler(scheduler) → scheduler — 현재 스레드의 파이버 스케줄러를 설정해요. 설정되면 비블로킹 파이버(Fiber.new에 blocking: false 또는 Fiber.schedule로 만든 것)는 블로킹 가능한 연산에서 해당 스케줄러의 훅 메서드를 호출하고, 현재 스레드는 종료 시 스케줄러의 close 메서드를 호출해요. scheduler는 Fiber::Scheduler에 대응하는 어떤 클래스의 객체여도 되며 구현은 사용자 몫이에요.

::yield (Public Class Method)

yield(args, ...) → obj — 파이버를 재개한 컨텍스트로 제어를 돌려주고, 전달된 인자들을 함께 넘겨요. 파이버는 다음 resume이 호출될 때 이 지점부터 처리를 재개해요. 다음 resume에 전달되는 인자들은 이 Fiber.yield 표현식이 평가되는 값이 돼요.

#alive? (Public Instance Method)

alive? → true 또는 false — 파이버가 여전히 재개(또는 transfer)될 수 있으면 true를 돌려줘요. 파이버 블록 실행이 끝난 후에는 항상 false를 돌려줘요.

#backtrace (Public Instance Method)

backtrace → array, backtrace(start) → array, backtrace(start, count) → array, backtrace(start..end) → array — 파이버의 현재 실행 스택을 돌려줘요. start, count, end로 백트레이스의 일부만 선택할 수 있어요.

def level3
  Fiber.yield
end

def level2
  level3
end

def level1
  level2
end

f = Fiber.new { level1 }

# It is empty before the fiber started
f.backtrace
#=> []

f.resume

f.backtrace
#=> ["test.rb:2:in `yield'", "test.rb:2:in `level3'", "test.rb:6:in `level2'", "test.rb:10:in `level1'", "test.rb:13:in `block in <main>'"]
p f.backtrace(1) # start from the item 1
#=> ["test.rb:2:in `level3'", "test.rb:6:in `level2'", "test.rb:10:in `level1'", "test.rb:13:in `block in <main>'"]
p f.backtrace(2, 2) # start from item 2, take 2
#=> ["test.rb:6:in `level2'", "test.rb:10:in `level1'"]
p f.backtrace(1..3) # take items from 1 to 3
#=> ["test.rb:2:in `level3'", "test.rb:6:in `level2'", "test.rb:10:in `level1'"]

f.resume

# It is nil after the fiber is finished
f.backtrace
#=> nil

#backtrace_locations (Public Instance Method)

backtrace_locations → array (+ backtrace_locations(start), backtrace_locations(start, count), backtrace_locations(start..end)) — backtrace와 같지만 실행 스택의 각 줄을 Thread::Backtrace::Location으로 돌려줘요. backtrace와 같은 인자를 받아요.

f = Fiber.new { Fiber.yield }
f.resume
loc = f.backtrace_locations.first
loc.label  #=> "yield"
loc.path   #=> "test.rb"
loc.lineno #=> 1

#blocking? (Public Instance Method)

blocking? → true 또는 false — 파이버가 블로킹이면 true, 아니면 false를 돌려줘요. false여도 현재 스레드에 Fiber.scheduler가 설정돼 있을 때만 동작이 달라져요.

#inspect

inspect() — to_s의 별칭이에요.

#kill (Public Instance Method)

kill → nil — 잡을 수 없는 예외를 발생시켜 파이버를 종료해요. 그 파이버만 종료하고 다른 파이버는 종료하지 않으며, 그 파이버가 resume이나 transfer를 호출 중이었다면 다른 파이버에 nil을 돌려줘요. Fiber#kill은 파이버가 Fiber.yield에 있을 때만 다른 파이버를 중단해요. 현재 파이버에 호출하면 Fiber#kill 호출 지점에서 그 예외를 발생시켜요. 아직 시작되지 않은 파이버면 종료 상태로 바로 전환해요. 이미 종료된 파이버면 아무 것도 하지 않아요. 다른 스레드에 속한 파이버를 호출하면 FiberError를 발생시켜요.

#raise (Public Instance Method)

raise → obj, raise(string) → obj, raise(exception [, string [, array]]) → obj — 마지막 Fiber.yield가 호출된 지점에서 파이버에 예외를 발생시켜요. 파이버가 아직 시작되지 않았거나 이미 끝까지 실행됐으면 FiberError를 발생시켜요. 파이버가 yield 중이면 재개돼요. transfer 중이면 transfer로 들어가요. 하지만 resume 중이면 FiberError를 발생시켜요.

인자가 없으면 RuntimeError, 단일 String 인자면 그 문자열을 메시지로 한 RuntimeError를 발생시켜요. 그 외에는 첫 파라미터가 Exception 클래스 이름(또는 exception 메시지를 받으면 Exception 객체를 돌려주는 객체)이어야 해요. 선택 두 번째 파라미터는 예외 메시지, 세 번째는 콜백 정보 배열이에요. 예외는 begin...end 블록의 rescue 절이 잡아요. 다른 Thread에 속한 파이버를 호출하면 FiberError.

#resume (Public Instance Method)

resume(args, ...) → obj — 마지막 Fiber.yield가 호출된 지점부터 파이버를 재개하거나, 첫 resume 호출이면 실행을 시작해요. resume에 전달된 인자는 Fiber.yield 표현식의 값이 되거나 첫 재개면 파이버 블록의 파라미터로 전달돼요. 반대로 resume이 호출됐을 때는 파이버 블록 안의 다음 Fiber.yield 문에 전달된 인자들 또는(Fiber.yield 없이 끝까지 실행되면) 블록 값으로 평가돼요.

#storage (Public Instance Method)

storage → hash (dup) — 파이버의 스토리지 해시의 사본을 돌려줘요. Fiber.current에서만 호출할 수 있어요.

#storage= (Public Instance Method)

storage = hash — 파이버의 스토리지 해시를 설정해요. 실험적 기능이라 미래에 바뀔 수 있어요. Fiber.current에서만 호출할 수 있어요. 중요한 파이버-스토리지 상태를 실수로 지울 수 있으니 주의해서 사용해야 해요. 특정 키를 할당하려면 Fiber::[]=를 쓰는 게 좋아요. Fiber.new(storage: nil)로 빈 스토리지의 파이버를 만들 수도 있어요.

while request = request_queue.pop
  # Reset the per-request state:
  Fiber.current.storage = nil
  handle_request(request)
end

#to_s

to_s() — 파이버를 나타내는 문자열. inspect의 별칭이에요.

#transfer (Public Instance Method)

transfer(args, ...) → obj — 다른 파이버로 제어를 넘겨요. 그 파이버가 마지막으로 멈춘 지점부터 재개하거나, 이전에 재개된 적 없으면 시작해요. 호출한 파이버는 Fiber.yield 호출 때처럼 일시 정지돼요. transfer를 받는 파이버는 resume 호출처럼 취급해요.

파이버로/부터 제어를 넘기는 두 가지 스타일(resume·Fiber::yield와 transfer·transfer)은 자유롭게 섞을 수 없어요. transfer로 생애주기가 시작된 파이버는 결코 yield하거나 resume 제어를 받을 수 없고, 끝내거나 transfer로 돌려보낼 수만 있어요. resume으로 시작된 파이버는 yield하거나 다른 파이버로 transfer할 수 있지만, 제어를 준 방식과 호환되는 방식으로만 돌려받을 수 있어요. 규칙을 어기면 FiberError가 발생해요.

개별 파이버 설계에는 yield/resume이 더 쉽지만, transfer는 복잡한 경우에 더 유연해서 서로 의존하는 파이버의 임의 그래프를 만들 수 있어요.

manager = nil # For local var to be visible inside worker block

# This fiber would be started with transfer
# It can't yield, and can't be resumed
worker = Fiber.new { |work|
  puts "Worker: starts"
  puts "Worker: Performed #{work.inspect}, transferring back"
  # Fiber.yield     # this would raise FiberError: attempt to yield on a not resumed fiber
  # manager.resume  # this would raise FiberError: attempt to resume a resumed fiber (double resume)
  manager.transfer(work.capitalize)
}

# This fiber would be started with resume
# It can yield or transfer, and can be transferred
# back or resumed
manager = Fiber.new {
  puts "Manager: starts"
  puts "Manager: transferring 'something' to worker"
  result = worker.transfer('something')
  puts "Manager: worker returned #{result.inspect}"
  # worker.resume    # this would raise FiberError: attempt to resume a transferring fiber
  Fiber.yield        # this is OK, the fiber transferred from and to, now it can yield
  puts "Manager: finished"
}

puts "Starting the manager"
manager.resume
puts "Resuming the manager"
# manager.transfer  # this would raise FiberError: attempt to transfer to a yielding fiber
manager.resume

출력:

Starting the manager
Manager: starts
Manager: transferring 'something' to worker
Worker: starts
Worker: Performed "something", transferring back
Manager: worker returned "Something"
Resuming the manager
Manager: finished