Class: Phronomy::EventLoop

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/engine/event_loop.rb

Overview

Runtime-owned FIFO event loop for FSMSession instances.

EventLoop owns the framework's sole control-plane OS thread. All session lifecycle progression happens by short event dispatches on this thread.

Constant Summary collapse

SYSTEM_CHANNEL_ID = 
"__event_loop__"
QUEUE_BACKLOG_WARNING_THRESHOLD = 
1_000
QUEUE_BACKLOG_WARNING_INTERVAL_SECONDS = 
60.0

Instance Method Summary collapse

Constructor Details

#initialize(runtime:) ⇒ EventLoop

Returns a new instance of EventLoop.



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/phronomy/engine/event_loop.rb', line 21

def initialize(runtime:)
  @runtime = runtime
  @queue = Phronomy::Concurrency::AsyncQueue.new
  @queue_metrics_mutex = Mutex.new
  @queue_depth = 0
  @max_queue_depth = 0
  @last_queue_backlog_warning_at = nil

  @fsms = {}
  @waiting = {}
  @admitted_session_ids = Set.new

  @lifecycle_mutex = Mutex.new
  @idle_cond = ConditionVariable.new
  @shutdown_mutex = Mutex.new
  @state = :running
  @outstanding_sessions = 0
  @shutdown_status = nil

  @lag_mutex = Mutex.new
  @last_lag_ns = 0
  @max_lag_ns = 0
  @dispatch_count = 0
  @total_lag_ns = 0

  @thread = Thread.new { run_loop }
  @thread.name = "phronomy-event-loop"
end

Instance Method Details

#admitted_session?(session_id) ⇒ Boolean

Returns:

  • (Boolean) 


163
164
165
 
# File 'lib/phronomy/engine/event_loop.rb', line 163

def admitted_session?(session_id)
  @lifecycle_mutex.synchronize { @admitted_session_ids.include?(session_id) }
end

#average_lag_seconds ⇒ Object 



58
59
60
61
62
63
 
# File 'lib/phronomy/engine/event_loop.rb', line 58

def average_lag_seconds
  @lag_mutex.synchronize do
    return 0.0 if @dispatch_count.zero?
    @total_lag_ns.to_f / @dispatch_count / 1_000_000_000.0
  end
end

#begin_draining ⇒ Object 



175
176
177
178
179
180
 
# File 'lib/phronomy/engine/event_loop.rb', line 175

def begin_draining
  @lifecycle_mutex.synchronize do
    @state = :draining if @state == :running
  end
  self
end

#current? ⇒ Boolean

Returns:

  • (Boolean) 


167
168
169
 
# File 'lib/phronomy/engine/event_loop.rb', line 167

def current?
  Thread.current.equal?(@thread)
end

#idle? ⇒ Boolean

Returns:

  • (Boolean) 


182
183
184
 
# File 'lib/phronomy/engine/event_loop.rb', line 182

def idle?
  @lifecycle_mutex.synchronize { @outstanding_sessions.zero? }
end

#last_lag_seconds ⇒ Object 



50
51
52
 
# File 'lib/phronomy/engine/event_loop.rb', line 50

def last_lag_seconds
  @lag_mutex.synchronize { @last_lag_ns } / 1_000_000_000.0
end

#max_lag_seconds ⇒ Object 



54
55
56
 
# File 'lib/phronomy/engine/event_loop.rb', line 54

def max_lag_seconds
  @lag_mutex.synchronize { @max_lag_ns } / 1_000_000_000.0
end

#max_queue_depth ⇒ Object 



69
70
71
 
# File 'lib/phronomy/engine/event_loop.rb', line 69

def max_queue_depth
  @queue_metrics_mutex.synchronize { @max_queue_depth }
end

#post(event) ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/phronomy/engine/event_loop.rb', line 111

def post(event)
  queued_depth = nil
  accepted = @lifecycle_mutex.synchronize do
    next false unless accepting_events?

    terminal_session_id = nil
    if terminal_management_event?(event)
      terminal_session_id = event.payload.fetch(:session_id)
      @admitted_session_ids.delete(terminal_session_id)
    end

    begin
      queued_depth = enqueue([event, monotonic_nanoseconds])
    rescue
      @admitted_session_ids.add(terminal_session_id) if terminal_session_id
      raise
    end
    true
  end
  return false unless accepted

  check_queue_backlog(queued_depth, event)
  true
end

#post_to_session(event) ⇒ Object 



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/phronomy/engine/event_loop.rb', line 136

def post_to_session(event)
  if event.target_id == SYSTEM_CHANNEL_ID
    raise ArgumentError, "post_to_session cannot target the system channel"
  end

  queued_depth = nil
  accepted = @lifecycle_mutex.synchronize do
    next false unless accepting_events?
    next false unless @admitted_session_ids.include?(event.target_id)

    queued_depth = enqueue([event, monotonic_nanoseconds])
    true
  end
  return false unless accepted

  check_queue_backlog(queued_depth, event)
  true
end

#queue_depth ⇒ Object 



65
66
67
 
# File 'lib/phronomy/engine/event_loop.rb', line 65

def queue_depth
  @queue_metrics_mutex.synchronize { @queue_depth }
end

#register(fsm_session, completion: nil) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/phronomy/engine/event_loop.rb', line 73

def register(fsm_session, completion: nil)
  if current? && !completion.is_a?(Phronomy::Task)
    raise Phronomy::Error,
      "Cannot call a synchronous invocation API from an EventLoop action. " \
      "Schedule work asynchronously instead."
  end

  completion_handle = completion || Phronomy::Concurrency::AsyncQueue.new
  event = Phronomy::Event.new(
    type: :start,
    target_id: SYSTEM_CHANNEL_ID,
    payload: {session: fsm_session, completion: completion_handle}
  )
  queued_depth = nil

  @lifecycle_mutex.synchronize do
    ensure_accepting_registrations!
    if @admitted_session_ids.include?(fsm_session.id)
      raise Phronomy::Error,
        "FSMSession #{fsm_session.id.inspect} is already registered"
    end

    @admitted_session_ids.add(fsm_session.id)
    @outstanding_sessions += 1
    begin
      queued_depth = enqueue([event, monotonic_nanoseconds])
    rescue
      @admitted_session_ids.delete(fsm_session.id)
      @outstanding_sessions -= 1
      @idle_cond.broadcast if @outstanding_sessions.zero?
      raise
    end
  end

  check_queue_backlog(queued_depth, event)
  completion_handle
end

#shutdown(deadline:, cancel_grace: deadline) ⇒ Object

Legacy entry point kept for any callers that pass deadline:/cancel_grace:.



224
225
226
 
# File 'lib/phronomy/engine/event_loop.rb', line 224

def shutdown(deadline:, cancel_grace: deadline)
  stop_and_join(deadline: deadline)
end

#state ⇒ Object 



171
172
173
 
# File 'lib/phronomy/engine/event_loop.rb', line 171

def state
  @lifecycle_mutex.synchronize { @state }
end

#stop_and_join(deadline:) ⇒ Object

Sends STOP to the queue and joins the EventLoop thread. Assumes sessions have already been drained before this call.



199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
 
# File 'lib/phronomy/engine/event_loop.rb', line 199

def stop_and_join(deadline:)
  @shutdown_mutex.synchronize do
    return @shutdown_status if @shutdown_status

    if state == :failed
      join_until(deadline)
      @shutdown_status = :failed
      return @shutdown_status
    end

    begin_stopping_if_idle
    join_until(deadline)

    @shutdown_status = if thread_alive?
      @lifecycle_mutex.synchronize { @state = :failed }
      :cancel_timeout
    elsif state == :failed
      :failed
    else
      finalize_terminated(:terminated)
    end
  end
end

#thread_alive? ⇒ Boolean Also known as: task_alive?

Returns:

  • (Boolean) 


228
229
230
 
# File 'lib/phronomy/engine/event_loop.rb', line 228

def thread_alive?
  @thread&.alive? || false
end

#wait_until_idle(deadline) ⇒ Object 



186
187
188
189
190
191
192
193
194
195
 
# File 'lib/phronomy/engine/event_loop.rb', line 186

def wait_until_idle(deadline)
  @lifecycle_mutex.synchronize do
    until @outstanding_sessions.zero?
      remaining = deadline - monotonic_now
      return false if remaining <= 0
      @idle_cond.wait(@lifecycle_mutex, remaining)
    end
    true
  end
end

#wake ⇒ Object

Interrupts the queue wait so EventLoop can recompute the next timer deadline.



156
157
158
159
160
161
 
# File 'lib/phronomy/engine/event_loop.rb', line 156

def wake
  @queue.push(WAKE)
  true
rescue ClosedQueueError
  false
end