Class: Rage::FiberScheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/rage/fiber_scheduler.rb

Defined Under Namespace

Modules: BlockingOperationWait

Instance Method Summary collapse

Constructor Details

#initialize ⇒ FiberScheduler

Initialize the scheduler, storing the root fiber and an empty DNS cache.



16
17
18
19
20
 
# File 'lib/rage/fiber_scheduler.rb', line 16

def initialize
  @root_fiber = Fiber.current
  @dns_cache = {}
  @blocked = {}
end

Instance Method Details

#address_resolve(hostname) ⇒ Object

Resolve a hostname to IP addresses, caching results for 60 seconds.



112
113
114
115
116
117
118
119
120
 
# File 'lib/rage/fiber_scheduler.rb', line 112

def address_resolve(hostname)
  @dns_cache[hostname] ||= begin
    ::Iodine.run_after(60_000) do
      @dns_cache[hostname] = nil
    end

    Resolv.getaddresses(hostname)
  end
end

#block(_blocker, timeout = nil) ⇒ Object

Block the current fiber until unblocked or timeout.



123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/rage/fiber_scheduler.rb', line 123

def block(_blocker, timeout = nil)
  f = Fiber.current
  gen = (f.__wait_generation += 1)
  @blocked[f] = gen

  if timeout
    ::Iodine.run_after((timeout * 1000).to_i) { unblock(nil, f) if gen == f.__wait_generation }
  end

  Fiber.yield
ensure
  @blocked.delete(f)
end

#close ⇒ Object

Clean up by closing the worker pool and Iodine scheduler.



211
212
213
214
 
# File 'lib/rage/fiber_scheduler.rb', line 211

def close
  @worker_pool&.close
  ::Iodine::Scheduler.close
end

#fiber(&block) ⇒ Object

Create and schedule a new non-blocking fiber, handling request and user-spawned fibers differently.



179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/rage/fiber_scheduler.rb', line 179

def fiber(&block)
  parent = Fiber.current

  fiber = if parent == @root_fiber
    # the fiber to wrap a request in
    Fiber.new(blocking: false) do
      Fiber.current.__set_id
      Rage::Telemetry.tracer.span_core_fiber_dispatch do
        Fiber.current.__set_result(block.call)
      end
    end
  else
    # the fiber was created in the user code
    Fiber.new(blocking: false) do
      Rage::Telemetry.tracer.span_core_fiber_spawn(parent:) do
        Fiber.current.__set_result(block.call)
      end
      # send a message for `Fiber.await` to work
      Iodine.publish(parent.__await_channel, "", Iodine::PubSub::PROCESS) if parent.__await_channel
    rescue Exception => e
      Fiber.current.__set_err(e)
      Iodine.publish(parent.__await_channel, Fiber::AWAIT_ERROR_MESSAGE, Iodine::PubSub::PROCESS) if parent.__await_channel
    end
  end

  fiber.__wait_generation = 0
  fiber.resume

  fiber
end

#fiber_interrupt(fiber, exception) ⇒ Object

Interrupt a fiber by incrementing its generation and raising an exception.



148
149
150
151
 
# File 'lib/rage/fiber_scheduler.rb', line 148

def fiber_interrupt(fiber, exception)
  fiber.__wait_generation += 1
  fiber.raise(exception)
end

#io_read(io, buffer, length, offset = 0) ⇒ Object 



41
42
43
 
# File 'lib/rage/fiber_scheduler.rb', line 41

def io_read(io, buffer, offset, length)
  ::Iodine::Scheduler.read(io.fileno, buffer, length, offset)
end

#io_wait(io, events, timeout = nil) ⇒ Object

Wait for I/O events on a file descriptor, yielding the fiber until ready or timeout.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/rage/fiber_scheduler.rb', line 23

def io_wait(io, events, timeout = nil)
  f = Fiber.current
  gen = (f.__wait_generation += 1)

  ::Iodine::Scheduler.attach(io.fileno, events, timeout&.ceil) do |err|
    f.resume(err) if f.alive? && gen == f.__wait_generation
  end

  err = Fiber.defer(io.fileno)
  if err == false || (err && err < 0)
    err
  else
    events
  end
end

#io_write(io, buffer, length, offset = 0) ⇒ Object 



74
75
76
 
# File 'lib/rage/fiber_scheduler.rb', line 74

def io_write(io, buffer, offset, length)
  ::Iodine::Scheduler.write(io.fileno, buffer, length, offset)
end

#kernel_sleep(duration = nil) ⇒ Object

Pause the current fiber for the specified duration.



88
89
90
91
 
# File 'lib/rage/fiber_scheduler.rb', line 88

def kernel_sleep(duration = nil)
  block(nil, duration || 0)
  Fiber.pause if duration.nil? || duration < 1
end

#process_wait(pid, flags) ⇒ Object

Wait for a specified process.



174
175
176
 
# File 'lib/rage/fiber_scheduler.rb', line 174

def process_wait(pid, flags)
  Thread.new { Process::Status.wait(pid, flags) }.value
end

#timeout_after(duration, exception_class = Timeout::Error, *exception_arguments, &block) ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/rage/fiber_scheduler.rb', line 94

def timeout_after(duration, exception_class = Timeout::Error, *exception_arguments, &block)
  f, fulfilled = Fiber.current, false

  ::Iodine.run_after((duration * 1000).to_i) do
    if !fulfilled && f.alive?
      fulfilled = true
      f.__wait_generation += 1
      f.raise(exception_class, *exception_arguments)
    end
  end

  block.call
ensure
  fulfilled = true
end

#unblock(_blocker, fiber) ⇒ Object

Unblock a fiber previously blocked by #block.



138
139
140
141
142
143
144
145
 
# File 'lib/rage/fiber_scheduler.rb', line 138

def unblock(_blocker, fiber)
  gen = @blocked.delete(fiber)
  return unless gen

  ::Iodine.defer do
    fiber.resume if fiber.alive? && gen == fiber.__wait_generation
  end
end