Module: OMQ::Reactor

Defined in:
lib/omq/reactor.rb

Overview

Shared IO reactor for the Ruby backend.

When user code runs inside an Async reactor, engine tasks are spawned directly under the caller's Async task. When no reactor is available (e.g. bare Thread.new), a single shared IO thread hosts all engine tasks — mirroring libzmq's IO thread.

Engines obtain the IO thread's root task via Reactor.root_task and use it as their @parent_task. Blocking operations from the main thread are dispatched to the IO thread via Reactor.run.

Constant Summary collapse

THREAD_NAME = 
'omq-io'
NATIVE_FIBER_SCHEDULER = 
Fiber.respond_to?(:scheduler) && Fiber.method(:scheduler).source_location.nil?

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.lingers ⇒ Hash{Numeric => Integer} (readonly)

Returns linger value → active socket count.

Returns:

  • (Hash{Numeric => Integer}) —

    linger value → active socket count



33
34
35
 
# File 'lib/omq/reactor.rb', line 33

def lingers
  @lingers
end

Class Method Details

.native_fiber_scheduler? ⇒ Boolean

Returns:

  • (Boolean) 


36
37
38
 
# File 'lib/omq/reactor.rb', line 36

def native_fiber_scheduler?
  NATIVE_FIBER_SCHEDULER
end

.root_task ⇒ Async::Task

Returns the root Async task inside the shared IO thread. Starts the thread exactly once (double-checked lock).

Returns:

  • (Async::Task) 


46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/omq/reactor.rb', line 46

def root_task
  pid = Process.pid
  return @root_task if @root_task && @pid == pid

  reset_after_fork if @pid && @pid != pid

  @mutex.synchronize do
    return @root_task if @root_task && @pid == pid

    ready        = Thread::Queue.new
    @work_queue  = Async::Queue.new
    @thread      = Thread.new { run_reactor(ready) }
    @thread.name = THREAD_NAME
    @root_task   = ready.pop
    @pid         = pid

    at_exit { stop! }
  end

  @root_task
end

.run(timeout: nil, &block) ⇒ Object

Runs a block inside the Async reactor.

Inside an Async reactor: runs directly. Outside: dispatches to the shared IO thread and blocks the calling thread until the result is available.

Returns:

  • (Object) —

    the block's return value



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/omq/reactor.rb', line 77

def run(timeout: nil, &block)
  task = Async::Task.current?

  if task
    if timeout
      task.with_timeout(timeout, IO::TimeoutError) { yield }
    else
      yield
    end
  elsif !native_fiber_scheduler?
    if timeout
      Timeout.timeout(timeout, IO::TimeoutError) { yield }
    else
      yield
    end
  else
    result = Async::Promise.new
    root_task # ensure started
    @work_queue << [block, result, timeout]
    result.wait
  end
end

.stop! ⇒ void

This method returns an undefined value.

Stops the shared IO thread.



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/omq/reactor.rb', line 128

def stop!
  if @pid && @pid != Process.pid
    reset_after_fork
    return
  end

  return unless @thread&.alive?

  max_linger = @lingers.empty? ? 0 : @lingers.keys.max

  @work_queue << nil if @work_queue
  @thread&.join(max_linger + 1)

  @thread     = nil
  @root_task  = nil
  @work_queue = nil
  @pid        = nil
  @lingers.clear
end

.track_linger(seconds) ⇒ Object

Registers a socket's linger value.

Parameters:

  • seconds (Numeric, nil) —

    linger value



105
106
107
 
# File 'lib/omq/reactor.rb', line 105

def track_linger(seconds)
  @lingers[seconds || 0] += 1
end

.untrack_linger(seconds) ⇒ Object

Unregisters a socket's linger value.

Parameters:

  • seconds (Numeric, nil) —

    linger value



114
115
116
117
118
119
120
121
 
# File 'lib/omq/reactor.rb', line 114

def untrack_linger(seconds)
  key            = seconds || 0
  @lingers[key] -= 1

  if @lingers[key] <= 0
    @lingers.delete(key)
  end
end