Class: UringMachine::FiberScheduler

Inherits:

Object

Object
UringMachine::FiberScheduler

show all

Defined in:: lib/uringmachine/fiber_scheduler.rb

Overview

Implements the ‘Fiber::Scheduler` interface for creating fiber-based concurrent applications in Ruby, in tight integration with the standard Ruby I/O and locking APIs.

Constant Summary collapse

BLOCKING_OP_SUPPORT =

DEFAULT_THREAD_POOL = The blocking operation thread pool is shared by all fiber schedulers.

BLOCKING_OP_SUPPORT && BlockingOperationThreadPool.new

Instance Attribute Summary collapse

#fiber_map ⇒ Object readonly

WeakMap holding references scheduler fibers as keys.
#machine ⇒ Object readonly

UringMachine instance associated with scheduler.

Instance Method Summary collapse

#address_resolve(hostname) ⇒ Array<Addrinfo>

Resolves an hostname.
#block(blocker, timeout = nil) ⇒ bool

Blocks the current fiber by yielding to the machine.
#blocking_operation_wait(op) ⇒ void

Runs the given operation in a separate thread, so as not to block other fibers.
#fiber(&block) ⇒ Fiber

Creates a new fiber with the given block.
#fiber_interrupt(fiber, exception) ⇒ void

Interrupts the given fiber with an exception.
#initialize(machine = nil, thread_pool = DEFAULT_THREAD_POOL) ⇒ void constructor

Instantiates a scheduler with the given UringMachine instance.
#instance_variables_to_inspect ⇒ Object

:nodoc:.
#io_close(fd) ⇒ Integer

Closes the given fd.
#io_pread(io, buffer, from, length, offset) ⇒ Integer

Reads from the given IO at the given file offset.
#io_pwrite(io, buffer, from, length, offset) ⇒ Integer

Writes to the given IO at the given file offset.
#io_read(io, buffer, length, offset) ⇒ Integer

Reads from the given IO.
#io_select(rios, wios, eios, timeout = nil) ⇒ Object

Selects the first ready IOs from the given sets of IOs.
#io_wait(io, events, timeout = nil) ⇒ Integer

Waits for the given io to become ready.
#io_write(io, buffer, length, offset) ⇒ Integer

Writes to the given IO.
#join(*fibers) ⇒ void

Waits for the given fibers to terminate.
#kernel_sleep(duration = nil) ⇒ void

Sleeps for the given duration.
#process_wait(pid, flags) ⇒ Process::Status

Waits for a process to terminate.
#scheduler_close ⇒ void

Waits for all fiber to terminate.
#timeout_after(duration, exception, message, &block) ⇒ any

Run the given block with a timeout.
#unblock(blocker, fiber) ⇒ void

Unblocks the given fiber by scheduling it.
#yield ⇒ Object

Yields to the next runnable fiber.

Constructor Details

#initialize(machine = nil, thread_pool = DEFAULT_THREAD_POOL) ⇒ `void`

Instantiates a scheduler with the given UringMachine instance.

machine = UM.new
scheduler = UM::FiberScheduler.new(machine)
Fiber.set_scheduler(scheduler)

Parameters:

machine (UringMachine, nil) (defaults to: nil) —

UringMachine instance

# File 'lib/uringmachine/fiber_scheduler.rb', line 100

def initialize(machine = nil, thread_pool = DEFAULT_THREAD_POOL)
  @machine = machine || UM.new
  @thread_pool = thread_pool
  @fiber_map = ObjectSpace::WeakMap.new
  @thread = Thread.current
end

Instance Attribute Details

#fiber_map ⇒ `Object` (readonly)

WeakMap holding references scheduler fibers as keys.



90
91
92

# File 'lib/uringmachine/fiber_scheduler.rb', line 90

def fiber_map
  @fiber_map
end

#machine ⇒ `Object` (readonly)

UringMachine instance associated with scheduler.



87
88
89

# File 'lib/uringmachine/fiber_scheduler.rb', line 87

def machine
  @machine
end

Instance Method Details

#address_resolve(hostname) ⇒ `Array<Addrinfo>`

Resolves an hostname.

Parameters:

hostname (String) —

hostname to resolve

Returns:

(Array<Addrinfo>) —

array of resolved addresses

# File 'lib/uringmachine/fiber_scheduler.rb', line 407

def address_resolve(hostname)
  # p address_resolve: [hostname]
  Resolv.getaddresses(hostname)
end

#block(blocker, timeout = nil) ⇒ `bool`

Blocks the current fiber by yielding to the machine. This hook is called when a synchronization mechanism blocks, e.g. a mutex, a queue, etc.

Parameters:

blocker (any) —

blocker object
timeout (Number, nil) (defaults to: nil) —

optional timeout

Returns:

(bool) —

was the operation successful

# File 'lib/uringmachine/fiber_scheduler.rb', line 170

def block(blocker, timeout = nil)
  # p block: [blocker, timeout]
  if timeout
    @machine.timeout(timeout, Timeout::Error) { @machine.yield }
  else
    @machine.yield
  end
  true
rescue Timeout::Error
  false
end

#blocking_operation_wait(op) ⇒ `void`

This method returns an undefined value.

Runs the given operation in a separate thread, so as not to block other fibers.

Parameters:

op (callable) —

blocking operation

# File 'lib/uringmachine/fiber_scheduler.rb', line 157

def blocking_operation_wait(op)
  # p blocking_operation_wait: [op]
  @thread_pool.process(@machine, op)
end

#fiber(&block) ⇒ `Fiber`

Creates a new fiber with the given block. The created fiber is added to the fiber map, scheduled on the scheduler machine, and started before this method returns (by calling snooze).

Parameters:

block (Proc) —

fiber block

Returns:

(Fiber)

# File 'lib/uringmachine/fiber_scheduler.rb', line 118

def fiber(&block)
  # p fiber: [block]
  fiber = Fiber.new(blocking: false) { @machine.run(fiber, &block) }

  @fiber_map[fiber] = true
  @machine.schedule(fiber, nil)
  @machine.snooze
  fiber
end

#fiber_interrupt(fiber, exception) ⇒ `void`

This method returns an undefined value.

Interrupts the given fiber with an exception.

Parameters:

fiber (Fiber) —

fiber to interrupt
exception (Exception) —

Exception

# File 'lib/uringmachine/fiber_scheduler.rb', line 397

def fiber_interrupt(fiber, exception)
  # p fiber_interrupt: [fiber, exception]
  @machine.schedule(fiber, exception)
  @machine.wakeup
end

#instance_variables_to_inspect ⇒ `Object`

:nodoc:



108
109
110

# File 'lib/uringmachine/fiber_scheduler.rb', line 108

def instance_variables_to_inspect
  [:@machine]
end

#io_close(fd) ⇒ `Integer`

Closes the given fd.

Parameters:

fd (Integer) —

file descriptor

Returns:

(Integer) —

file descriptor

# File 'lib/uringmachine/fiber_scheduler.rb', line 372

def io_close(fd)
  # p io_close: [fd]
  @machine.close_async(fd)
rescue Errno => e
  -e.errno
end

#io_pread(io, buffer, from, length, offset) ⇒ `Integer`

Reads from the given IO at the given file offset

Parameters:

io (IO) —

IO object
buffer (IO::Buffer) —

read buffer
from (Integer) —

read offset
length (Integer) —

read length
offset (Integer) —

buffer offset

Returns:

(Integer) —

bytes read

# File 'lib/uringmachine/fiber_scheduler.rb', line 294

def io_pread(io, buffer, from, length, offset)
  # p io_pread: [io, buffer, from, length, offset]
  length = buffer.size if length == 0

  if (timeout = io.timeout)
    @machine.timeout(timeout, Timeout::Error) do
      @machine.read(io.fileno, buffer, length, offset, from)
    rescue Errno::EINTR
      retry
    end
  else
    @machine.read(io.fileno, buffer, length, offset, from)
  end
rescue Errno::EINTR
  retry
rescue Errno => e
  -e.errno
end

#io_pwrite(io, buffer, from, length, offset) ⇒ `Integer`

Writes to the given IO at the given file offset.

Parameters:

io (IO) —

IO object
buffer (IO::Buffer) —

write buffer
from (Integer) —

file offset
length (Integer) —

write length
offset (Integer) —

buffer offset

Returns:

(Integer) —

bytes written

# File 'lib/uringmachine/fiber_scheduler.rb', line 348

def io_pwrite(io, buffer, from, length, offset)
  # p io_pwrite: [io, buffer, from, length, offset]
  length = buffer.size if length == 0
  buffer = buffer.slice(offset) if offset > 0

  if (timeout = io.timeout)
    @machine.timeout(timeout, Timeout::Error) do
      @machine.write(io.fileno, buffer, length, from)
    rescue Errno::EINTR
      retry
    end
  else
    @machine.write(io.fileno, buffer, length, from)
  end
rescue Errno::EINTR
  retry
rescue Errno => e
  -e.errno
end

#io_read(io, buffer, length, offset) ⇒ `Integer`

Reads from the given IO.

Parameters:

io (IO) —

IO object
buffer (IO::Buffer) —

read buffer
length (Integer) —

read length
offset (Integer) —

buffer offset

Returns:

(Integer) —

bytes read

# File 'lib/uringmachine/fiber_scheduler.rb', line 267

def io_read(io, buffer, length, offset)
  # p io_read: [io, buffer, length, offset]
  length = buffer.size if length == 0

  if (timeout = io.timeout)
    @machine.timeout(timeout, Timeout::Error) do
      @machine.read(io.fileno, buffer, length, offset)
    rescue Errno::EINTR
      retry
    end
  else
    @machine.read(io.fileno, buffer, length, offset)
  end
rescue Errno::EINTR
  retry
rescue Errno => e
  -e.errno
end

#io_select(rios, wios, eios, timeout = nil) ⇒ `Object`

Selects the first ready IOs from the given sets of IOs.

Parameters:

rios (Array<IO>) —

readable IOs
wios (Array<IO>) —

writable IOs
eios (Array<IO>) —

exceptable IOs
timeout (Number, nil) (defaults to: nil) —

optional timeout

# File 'lib/uringmachine/fiber_scheduler.rb', line 242

def io_select(rios, wios, eios, timeout = nil)
  # p io_select: [rios, wios, eios, timeout]
  map_r = map_fds(rios)
  map_w = map_fds(wios)
  map_e = map_fds(eios)

  r, w, e = nil
  if timeout
    @machine.timeout(timeout, Timeout::Error) {
      r, w, e = @machine.select(map_r.keys, map_w.keys, map_e.keys)
    }
  else
    r, w, e = @machine.select(map_r.keys, map_w.keys, map_e.keys)
  end

  [unmap_fds(r, map_r), unmap_fds(w, map_w), unmap_fds(e, map_e)]
end

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

Waits for the given io to become ready.

Parameters:

io (IO) —

IO object
events (Integer) —

readiness bitmask
timeout (Number, nil) (defaults to: nil) —

optional timeout

Returns:

(Integer) —

ready events bitmask

# File 'lib/uringmachine/fiber_scheduler.rb', line 216

def io_wait(io, events, timeout = nil)
  # Useful note from the Carbon Fiber Fiber::Scheduler implementation:
  # Net::HTTP#begin_transport calls `wait_readable(0)` before every
  # keep-alive request to probe for a closed connection.  On a healthy
  # connection this is always "not readable", so returning false
  # directly saves one MSG_PEEK recvfrom per request.  On a genuinely
  # closed connection Net::HTTP will detect EOF on the next real read
  # and reconnect — one extra request's worth of latency, at most.
  return 0 if timeout == 0 && events == ::IO::READABLE && io.is_a?(BasicSocket)

  timeout ||= io.timeout
  if timeout && timeout > 0
    @machine.timeout(timeout, Timeout::Error) {
      @machine.poll(io.fileno, events)
    }
  else
    @machine.poll(io.fileno, events)
  end
end

#io_write(io, buffer, length, offset) ⇒ `Integer`

Writes to the given IO.

Parameters:

io (IO) —

IO object
buffer (IO::Buffer) —

write buffer
length (Integer) —

write length
offset (Integer) —

write offset

Returns:

(Integer) —

bytes written

# File 'lib/uringmachine/fiber_scheduler.rb', line 320

def io_write(io, buffer, length, offset)
  # p io_write: [io, buffer, length, offset]
  length = buffer.size if length == 0
  buffer = buffer.slice(offset) if offset > 0

  if (timeout = io.timeout)
    @machine.timeout(timeout, Timeout::Error) do
      @machine.write(io.fileno, buffer, length)
    rescue Errno::EINTR
      retry
    end
  else
    @machine.write(io.fileno, buffer, length)
  end
rescue Errno::EINTR
  retry
rescue Errno => e
  -e.errno
end

#join(*fibers) ⇒ `void`

This method returns an undefined value.

Waits for the given fibers to terminate. If no fibers are given, waits for all fibers to terminate.

Parameters:

fibers (Array<Fiber>) —

fibers to terminate

# File 'lib/uringmachine/fiber_scheduler.rb', line 141

def join(*fibers)
  if fibers.empty?
    fibers = @fiber_map.keys
    @fiber_map = ObjectSpace::WeakMap.new
  end

  @machine.await(fibers)
end

#kernel_sleep(duration = nil) ⇒ `void`

This method returns an undefined value.

Sleeps for the given duration.

Parameters:

duration (Number, nil) (defaults to: nil) —

sleep duration

# File 'lib/uringmachine/fiber_scheduler.rb', line 199

def kernel_sleep(duration = nil)
  # p kernel_sleep: duration
  duration ? @machine.sleep(duration) : @machine.yield
end

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

Waits for a process to terminate.

Parameters:

pid (Integer) —

process pid (0 for any child process)
flags (Integer) —

waitpid flags

Returns:

(Process::Status) —

terminated process status

# File 'lib/uringmachine/fiber_scheduler.rb', line 386

def process_wait(pid, flags)
  flags = UM::WEXITED if flags == 0
  @machine.waitid_status(UM::P_PID, pid, flags)
end

#scheduler_close ⇒ `void`

This method returns an undefined value.

Waits for all fiber to terminate. Called upon thread termination or when the thread’s fiber scheduler is changed.



132
133
134

# File 'lib/uringmachine/fiber_scheduler.rb', line 132

def scheduler_close
  join()
end

#timeout_after(duration, exception, message, &block) ⇒ `any`

Run the given block with a timeout.

Parameters:

duration (Number) —

timeout duration
exception (Class) —

exception Class
message (String) —

exception message
block (Proc) —

block to run

Returns:

(any) —

block return value

# File 'lib/uringmachine/fiber_scheduler.rb', line 419

def timeout_after(duration, exception, message, &block)
  # p timeout_after:  [duration, exception, message, block]
  @machine.timeout(duration, exception, &block)
end

#unblock(blocker, fiber) ⇒ `void`

This method returns an undefined value.

Unblocks the given fiber by scheduling it. This hook is called when a synchronization mechanism unblocks, e.g. a mutex, a queue, etc.

Parameters:

blocker (any) —

blocker object
fiber (Fiber) —

fiber to resume

# File 'lib/uringmachine/fiber_scheduler.rb', line 189

def unblock(blocker, fiber)
  # p unblock: [blocker, fiber]
  @machine.schedule(fiber, nil)
  @machine.wakeup if Thread.current != @thread
end

#yield ⇒ `Object`

Yields to the next runnable fiber.

# File 'lib/uringmachine/fiber_scheduler.rb', line 205

def yield
  # p yield: []
  @machine.snooze
end

Class: UringMachine::FiberScheduler

Overview

Constant Summary collapse

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(machine = nil, thread_pool = DEFAULT_THREAD_POOL) ⇒ void

Instance Attribute Details

#fiber_map ⇒ Object (readonly)

#machine ⇒ Object (readonly)

Instance Method Details

#address_resolve(hostname) ⇒ Array<Addrinfo>

#block(blocker, timeout = nil) ⇒ bool

#blocking_operation_wait(op) ⇒ void

#fiber(&block) ⇒ Fiber

#fiber_interrupt(fiber, exception) ⇒ void

#instance_variables_to_inspect ⇒ Object

#io_close(fd) ⇒ Integer

#io_pread(io, buffer, from, length, offset) ⇒ Integer

#io_pwrite(io, buffer, from, length, offset) ⇒ Integer

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

#io_select(rios, wios, eios, timeout = nil) ⇒ Object

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

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

#join(*fibers) ⇒ void

#kernel_sleep(duration = nil) ⇒ void

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

#scheduler_close ⇒ void

#timeout_after(duration, exception, message, &block) ⇒ any

#unblock(blocker, fiber) ⇒ void

#yield ⇒ Object

#initialize(machine = nil, thread_pool = DEFAULT_THREAD_POOL) ⇒ `void`

#fiber_map ⇒ `Object` (readonly)

#machine ⇒ `Object` (readonly)

#address_resolve(hostname) ⇒ `Array<Addrinfo>`

#block(blocker, timeout = nil) ⇒ `bool`

#blocking_operation_wait(op) ⇒ `void`

#fiber(&block) ⇒ `Fiber`

#fiber_interrupt(fiber, exception) ⇒ `void`

#instance_variables_to_inspect ⇒ `Object`

#io_close(fd) ⇒ `Integer`

#io_pread(io, buffer, from, length, offset) ⇒ `Integer`

#io_pwrite(io, buffer, from, length, offset) ⇒ `Integer`

#io_read(io, buffer, length, offset) ⇒ `Integer`

#io_select(rios, wios, eios, timeout = nil) ⇒ `Object`

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

#io_write(io, buffer, length, offset) ⇒ `Integer`

#join(*fibers) ⇒ `void`

#kernel_sleep(duration = nil) ⇒ `void`

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

#scheduler_close ⇒ `void`

#timeout_after(duration, exception, message, &block) ⇒ `any`

#unblock(blocker, fiber) ⇒ `void`

#yield ⇒ `Object`