Class: CarbonFiber::Scheduler

Inherits:
Object
  • Object
show all
Defined in:
lib/carbon_fiber/scheduler.rb

Overview

Implements the Ruby Fiber Scheduler interface.

Delegates I/O and timer operations to a native Zig selector (io_uring on Linux, kqueue on macOS). Operations the native layer doesn't cover (DNS, process_wait) run on background threads.

Examples:

scheduler = CarbonFiber::Scheduler.new
Fiber.set_scheduler(scheduler)
Fiber.schedule { sleep 1; puts "done" }
scheduler.run
Fiber.set_scheduler(nil)

Instance Method Summary collapse

Constructor Details

#initialize(root_fiber = Fiber.current, selector: CarbonFiber::Native::Selector, io_contract_v4: CarbonFiber.io_contract_v4?) ⇒ Scheduler

Returns a new instance of Scheduler.

Parameters:

  • root_fiber (Fiber) (defaults to: Fiber.current) —

    the event loop fiber (defaults to current)

  • selector (Class) (defaults to: CarbonFiber::Native::Selector) —

    native selector class to instantiate

  • io_contract_v4 (Boolean) (defaults to: CarbonFiber.io_contract_v4?) —

    use the Ruby 4.1+ single-transfer contract in the native I/O paths (defaults to what the running Ruby speaks; override only for testing)



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/carbon_fiber/scheduler.rb', line 49

def initialize(root_fiber = Fiber.current, selector: CarbonFiber::Native::Selector,
  io_contract_v4: CarbonFiber.io_contract_v4?)
  @root_fiber = root_fiber
  @scheduler_thread = Thread.current
  @main_ractor = !defined?(Ractor) || Ractor.current == Ractor.main
  # Resolv keeps state Ractor isolation forbids, and only the main
  # Ractor resolves through it (see #address_resolve), so load it here
  # rather than at require time; a worker Ractor never loads it.
  require "resolv" if @main_ractor
  @selector = selector.new(root_fiber)
  @selector.io_contract_v4 = io_contract_v4
  @active_fibers = 0
  @background_count = 0
  @closed = false
  @closing = false
end

Instance Method Details

#address_resolve(hostname) ⇒ Array<String>

Resolve a hostname to addresses.

The main Ractor uses Resolv: it speaks DNS over sockets the scheduler already multiplexes, so lookups never block the loop. Resolv keeps class-level state that Ractor isolation forbids, so a scheduler running in another Ractor asks the platform resolver instead. That call blocks its thread, and from a scheduled fiber it would re-enter this very hook, so it runs on a background thread like process_wait does. Unknown hosts yield an empty list on both paths.

Parameters:

  • hostname (String)

Returns:

  • (Array<String>) 


351
352
353
354
355
356
357
358
359
360
361
362
 
# File 'lib/carbon_fiber/scheduler.rb', line 351

def address_resolve(hostname)
  if hostname.include?("%")
    hostname = hostname.split("%", 2).first
  end
  return Resolv.getaddresses(hostname) if @main_ractor

  await_background_operation do
    Addrinfo.getaddrinfo(hostname, nil, nil, :STREAM).map(&:ip_address)
  rescue SocketError
    []
  end
end

#block(_blocker, timeout = nil) ⇒ Object

Suspend the current fiber until unblocked or timed out.

Parameters:

  • _blocker (Object) —

    unused, required by the protocol

  • timeout (Float, nil) (defaults to: nil) —

    seconds before automatic resume



167
168
169
 
# File 'lib/carbon_fiber/scheduler.rb', line 167

def block(_blocker, timeout = nil)
  @selector.block(Fiber.current, timeout)
end

#blocking_operation_wait(work) ⇒ Object

Run an arbitrary callable on a background thread.

Parameters:

  • work (#call) 


366
367
368
369
370
 
# File 'lib/carbon_fiber/scheduler.rb', line 366

def blocking_operation_wait(work)
  await_background_operation do
    work.call
  end
end

#close(internal = false) ⇒ Object

Drain pending work and release the native selector.



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/carbon_fiber/scheduler.rb', line 72

def close(internal = false)
  return true if @closed || @closing

  unless internal
    return Fiber.set_scheduler(nil) if Fiber.scheduler == self
  end

  @closing = true
  run
  true
ensure
  unless @closed
    @selector&.destroy
    @closed = true
    @closing = false
    freeze
  end
end

#closed? ⇒ Boolean

Returns whether the scheduler has been closed.

Returns:

  • (Boolean) —

    whether the scheduler has been closed



92
93
94
 
# File 'lib/carbon_fiber/scheduler.rb', line 92

def closed?
  @closed
end

#current_time ⇒ Float

Monotonic clock used by the scheduler for timers.

Returns:

  • (Float) —

    seconds since an arbitrary epoch



98
99
100
 
# File 'lib/carbon_fiber/scheduler.rb', line 98

def current_time
  Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

#fiber { ... } ⇒ Fiber

Create and schedule a non-blocking fiber.

Yields:

  • the block to run inside the fiber

Returns:

  • (Fiber) 


105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/carbon_fiber/scheduler.rb', line 105

def fiber(&block)
  fiber = Fiber.new(blocking: false) do
    block.call
  ensure
    fiber_done
  end

  @active_fibers += 1
  @selector.push(fiber)
  @selector.wakeup unless Thread.current.equal?(@scheduler_thread)

  fiber
end

#fiber_interrupt(fiber, exception) ⇒ Object

Deliver an exception to a fiber from another fiber.

Parameters:

  • fiber (Fiber)
  • exception (Exception) 


375
376
377
378
379
 
# File 'lib/carbon_fiber/scheduler.rb', line 375

def fiber_interrupt(fiber, exception)
  @selector.raise(fiber, exception)
  @selector.wakeup
  true
end

#io_close(descriptor) ⇒ Object

Cancel pending waiters on a descriptor and close it.

Parameters:

  • descriptor (Integer) 


311
312
313
314
315
316
317
318
319
320
 
# File 'lib/carbon_fiber/scheduler.rb', line 311

def io_close(descriptor)
  @selector.io_close(descriptor, IOError.new("stream closed while waiting"))

  Fiber.blocking do
    io = IO.for_fd(descriptor)
    io.close unless io.closed?
  end

  true
end

#io_read_v3(io, buffer, length, offset = 0) ⇒ Integer Also known as: io_read

Read from an IO into a buffer via the native selector (legacy contract, Ruby 3.4 through 4.0: length before offset, minimum-progress semantics). Falls back to a background thread for non-socket descriptors.

Parameters:

  • io (IO)
  • buffer (IO::Buffer)
  • length (Integer)
  • offset (Integer) (defaults to: 0)

Returns:

  • (Integer) —

    bytes read, or negative errno



210
211
212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/carbon_fiber/scheduler.rb', line 210

def io_read_v3(io, buffer, length, offset = 0)
  # Native io_read_object extracts the descriptor in Zig, skipping a
  # `respond_to?(:fileno)` + `io.fileno` method-send pair per call.
  native_result = @selector.io_read_object(io, buffer, length, offset)
  return native_result unless native_result.nil?

  await_background_operation do
    Fiber.blocking { buffer.read(io, length, offset) }
  end
rescue NoMethodError, TypeError
  await_background_operation do
    Fiber.blocking { buffer.read(io, length, offset) }
  end
end

#io_read_v4(io, buffer, offset, length) ⇒ Integer

Read from an IO into a buffer via the native selector (Ruby 4.1+ contract: offset before length, single transfer). One nonblocking attempt; short results and -EAGAIN are returned directly, and Ruby's own read loop composes retries via io_wait.

Parameters:

  • io (IO)
  • buffer (IO::Buffer)
  • offset (Integer)
  • length (Integer) —

    maximum bytes for this transfer

Returns:

  • (Integer) —

    bytes read, or negative errno



234
235
236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/carbon_fiber/scheduler.rb', line 234

def io_read_v4(io, buffer, offset, length)
  return 0 if length.zero?

  native_result = @selector.io_read_object(io, buffer, length, offset)
  return native_result unless native_result.nil?

  await_background_operation do
    Fiber.blocking { buffer.read(io, offset, length) }
  end
rescue NoMethodError, TypeError
  await_background_operation do
    Fiber.blocking { buffer.read(io, offset, length) }
  end
end

#io_select ⇒ Object

Blocking IO.select on a background thread.



303
304
305
306
307
 
# File 'lib/carbon_fiber/scheduler.rb', line 303

def io_select(...)
  await_background_operation do
    Fiber.blocking { IO.select(...) }
  end
end

#io_wait(io, events, timeout = nil) ⇒ Integer, false

Wait for I/O readiness on a file descriptor.

Parameters:

  • io (IO)
  • events (Integer) —

    bitmask of IO::READABLE, IO::WRITABLE

  • timeout (Float, nil) (defaults to: nil)

Returns:

  • (Integer, false) —

    readiness bitmask, or false on timeout



190
191
192
193
194
195
196
197
198
199
200
 
# File 'lib/carbon_fiber/scheduler.rb', line 190

def io_wait(io, events, timeout = nil)
  return poll_io_now(io, events) if timeout == 0

  # Native io_wait_object handles fileno extraction, Fiber.current,
  # and nil/numeric timeout in Zig — skipping a Ruby frame + branch
  # per call on Net::HTTP's hot read/write loop.
  result = @selector.io_wait_object(io, events, timeout)
  result.nil? ? await_background_operation { io_select_readiness(io, events, timeout) } : result
rescue NoMethodError, TypeError
  await_background_operation { io_select_readiness(io, events, timeout) }
end

#io_write_v3(io, buffer, length, offset = 0) ⇒ Integer Also known as: io_write

Write from a buffer to an IO via the native selector (legacy contract, Ruby 3.4 through 4.0: length before offset, minimum-progress semantics). Falls back to a background thread for non-socket descriptors.

Parameters:

  • io (IO)
  • buffer (IO::Buffer)
  • length (Integer)
  • offset (Integer) (defaults to: 0)

Returns:

  • (Integer) —

    bytes written, or negative errno



257
258
259
260
261
262
263
264
265
266
267
268
 
# File 'lib/carbon_fiber/scheduler.rb', line 257

def io_write_v3(io, buffer, length, offset = 0)
  native_result = @selector.io_write_object(io, buffer, length, offset)
  return native_result unless native_result.nil?

  await_background_operation do
    Fiber.blocking { buffer.write(io, length, offset) }
  end
rescue NoMethodError, TypeError
  await_background_operation do
    Fiber.blocking { buffer.write(io, length, offset) }
  end
end

#io_write_v4(io, buffer, offset, length) ⇒ Integer

Write from a buffer to an IO via the native selector (Ruby 4.1+ contract: offset before length, single transfer).

Parameters:

  • io (IO)
  • buffer (IO::Buffer)
  • offset (Integer)
  • length (Integer) —

    maximum bytes for this transfer

Returns:

  • (Integer) —

    bytes written, or negative errno



277
278
279
280
281
282
283
284
285
286
287
288
289
290
 
# File 'lib/carbon_fiber/scheduler.rb', line 277

def io_write_v4(io, buffer, offset, length)
  return 0 if length.zero?

  native_result = @selector.io_write_object(io, buffer, length, offset)
  return native_result unless native_result.nil?

  await_background_operation do
    Fiber.blocking { buffer.write(io, offset, length) }
  end
rescue NoMethodError, TypeError
  await_background_operation do
    Fiber.blocking { buffer.write(io, offset, length) }
  end
end

#kernel_sleep(duration = nil) ⇒ Object

Intercept Kernel#sleep. Parks the fiber on a native timer.

Parameters:

  • duration (Float, nil) (defaults to: nil) —

    seconds to sleep; nil sleeps forever



181
182
183
 
# File 'lib/carbon_fiber/scheduler.rb', line 181

def kernel_sleep(duration = nil)
  @selector.kernel_sleep(duration)
end

#process_wait(pid, flags) ⇒ Process::Status

Wait for a child process on a background thread.

Parameters:

  • pid (Integer)
  • flags (Integer) —

    waitpid flags

Returns:

  • (Process::Status) 


326
327
328
329
330
331
332
333
334
335
336
337
338
 
# File 'lib/carbon_fiber/scheduler.rb', line 326

def process_wait(pid, flags)
  # Ruby 4.0 bug: rb_process_status_wait re-enters the scheduler hook,
  # so native process_wait produces an incorrect status. Background-thread
  # waitpid avoids this because new threads have no scheduler installed.
  await_background_operation do
    if flags.zero?
      Process::Status.wait(pid, flags)
    else
      _waited_pid, status = Process.waitpid2(pid, flags)
      status
    end
  end
end

#push(fiber) ⇒ Object

Enqueue a fiber into the ready queue.

Parameters:

  • fiber (Fiber) 


131
132
133
 
# File 'lib/carbon_fiber/scheduler.rb', line 131

def push(fiber)
  @selector.push(fiber)
end

#raise(fiber, exception) ⇒ Object

Deliver an exception to a suspended fiber.

Parameters:

  • fiber (Fiber)
  • exception (Exception) 


149
150
151
 
# File 'lib/carbon_fiber/scheduler.rb', line 149

def raise(fiber, exception)
  @selector.raise(fiber, exception)
end

#resume(fiber, *arguments) ⇒ Object

Resume a fiber, optionally passing a value.

Parameters:

  • fiber (Fiber)
  • arguments (Array) —

    at most one value to pass to the fiber



138
139
140
141
142
143
144
 
# File 'lib/carbon_fiber/scheduler.rb', line 138

def resume(fiber, *arguments)
  if arguments.empty?
    @selector.push(fiber)
  else
    @selector.resume(fiber, arguments.first)
  end
end

#run ⇒ Object

Run the event loop until all fibers and background operations complete.



399
400
401
402
403
404
 
# File 'lib/carbon_fiber/scheduler.rb', line 399

def run
  Kernel.raise RuntimeError, "Scheduler has been closed" if closed?

  run_once until idle?
  true
end

#run_once(timeout = nil) ⇒ Object

Run one event loop iteration. Alias for #select.



394
395
396
 
# File 'lib/carbon_fiber/scheduler.rb', line 394

def run_once(timeout = nil)
  @selector.select(timeout)
end

#scheduler_close ⇒ Object

Called by Ruby when Fiber.set_scheduler(nil) is invoked.



67
68
69
 
# File 'lib/carbon_fiber/scheduler.rb', line 67

def scheduler_close
  close(true)
end

#select(timeout = nil) ⇒ Object

Run one iteration of the event loop.

Parameters:

  • timeout (Float, nil) (defaults to: nil) —

    maximum seconds to wait



160
161
162
 
# File 'lib/carbon_fiber/scheduler.rb', line 160

def select(timeout = nil)
  @selector.select(timeout)
end

#timeout_after(duration, klass = Timeout::Error, message = "execution expired", &block) ⇒ Object

Run a block with a timeout, raising an exception if it expires.

Parameters:

  • duration (Float) —

    seconds

  • klass (Class, Exception) (defaults to: Timeout::Error) —

    exception class or instance

  • message (String) (defaults to: "execution expired") 


385
386
387
388
389
390
391
 
# File 'lib/carbon_fiber/scheduler.rb', line 385

def timeout_after(duration, klass = Timeout::Error, message = "execution expired", &block)
  exc = klass.is_a?(Class) ? klass.new(message) : klass
  token = @selector.raise_after(Fiber.current, exc, duration)
  block.call(duration)
ensure
  @selector.cancel_timer(token) if token
end

#transfer ⇒ Object

Transfer control to the next ready fiber or the event loop.



120
121
122
 
# File 'lib/carbon_fiber/scheduler.rb', line 120

def transfer
  @selector.transfer
end

#unblock(_blocker, fiber) ⇒ Object

Resume a fiber previously suspended by #block.

Parameters:

  • _blocker (Object) —

    unused, required by the protocol

  • fiber (Fiber) 


174
175
176
177
 
# File 'lib/carbon_fiber/scheduler.rb', line 174

def unblock(_blocker, fiber)
  @selector.unblock(fiber)
  true
end

#wakeup ⇒ Object

Wake the event loop (thread-safe).



154
155
156
 
# File 'lib/carbon_fiber/scheduler.rb', line 154

def wakeup
  @selector.wakeup
end

#yield ⇒ Object

Re-enqueue the current fiber and transfer to the event loop.



125
126
127
 
# File 'lib/carbon_fiber/scheduler.rb', line 125

def yield
  @selector.yield
end