Module: NNQ::Reactor

Defined in:
lib/nnq/reactor.rb

Overview

Per-process fallback IO thread for non-Async callers.

When user code already runs inside an Async reactor, NNQ tasks attach directly to the caller’s task tree. When the caller is bare (e.g. a plain ‘Thread.new` or the main thread of a script), NNQ::Reactor lazily spawns one shared background thread that hosts an Async reactor and processes work items dispatched via Reactor.run.

This is not an Async scheduler — it is a fallback thread for an Async reactor. NNQ and OMQ each have their own private fallback for bare-thread callers; both can coexist with the user’s own reactor with no extraction or sharing required.

Class Method Summary collapse

Class Method Details

.root_task ⇒ Object 



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/nnq/reactor.rb', line 27

def root_task
  return @root_task if @root_task

  @mutex.synchronize do
    return @root_task if @root_task

    ready       = Thread::Queue.new
    @work_queue = Async::Queue.new
    @thread     = Thread.new { run_reactor(ready) }
    @thread.name = "nnq-io"
    @root_task  = ready.pop
    at_exit { stop! }
  end

  @root_task
end

.run(&block) ⇒ Object 



45
46
47
48
49
50
51
52
53
54
 
# File 'lib/nnq/reactor.rb', line 45

def run(&block)
  if Async::Task.current?
    yield
  else
    result = Async::Promise.new
    root_task # ensure started
    @work_queue.push([block, result])
    result.wait
  end
end

.stop! ⇒ Object 



57
58
59
60
61
62
63
64
 
# File 'lib/nnq/reactor.rb', line 57

def stop!
  return unless @thread&.alive?
  @work_queue&.push(nil)
  @thread&.join(2)
  @thread     = nil
  @root_task  = nil
  @work_queue = nil
end