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.

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.



17
18
19
20
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
49
 
# File 'lib/phronomy/engine/event_loop.rb', line 17

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 and @waiting are dispatcher-thread-owned.
  @fsms = {}
  @waiting = {}

  # Admission is shared by caller threads and the dispatcher. A session ID
  # enters this set before its :start event is queued and leaves when its
  # terminal management event is queued.
  @admitted_session_ids = Set.new

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

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

  @task = @runtime.__spawn_event_loop_service { run_loop }
end

Instance Method Details

#admitted_session?(session_id) ⇒ Boolean

Returns:

  • (Boolean) 


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

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

#average_lag_seconds ⇒ Object 



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

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 



195
196
197
198
199
200
 
# File 'lib/phronomy/engine/event_loop.rb', line 195

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

#current? ⇒ Boolean

Returns:

  • (Boolean) 


187
188
189
 
# File 'lib/phronomy/engine/event_loop.rb', line 187

def current?
  Phronomy::Task.current.equal?(@task)
end

#idle? ⇒ Boolean

Returns:

  • (Boolean) 


202
203
204
205
206
 
# File 'lib/phronomy/engine/event_loop.rb', line 202

def idle?
  @lifecycle_mutex.synchronize do
    @outstanding_sessions.zero?
  end
end

#last_lag_seconds ⇒ Object 



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

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

#max_lag_seconds ⇒ Object 



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

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

#max_queue_depth ⇒ Object 



71
72
73
 
# File 'lib/phronomy/engine/event_loop.rb', line 71

def max_queue_depth
  @queue_metrics_mutex.synchronize { @max_queue_depth }
end

#post(event) ⇒ Object

Internal post operation. Management terminal events close admission for their session before the terminal event is enqueued.



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/phronomy/engine/event_loop.rb', line 127

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

Posts an event only when the target session has been admitted and has not queued a terminal management event. The event remains FIFO with all other EventLoop work.

A true result reports admission, not transition success.



159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
 
# File 'lib/phronomy/engine/event_loop.rb', line 159

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 



67
68
69
 
# File 'lib/phronomy/engine/event_loop.rb', line 67

def queue_depth
  @queue_metrics_mutex.synchronize { @queue_depth }
end

#register(fsm_session, completion: nil) ⇒ Object 



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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/phronomy/engine/event_loop.rb', line 75

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_queue =
    completion || Phronomy::Concurrency::AsyncQueue.new
  scheduler = Phronomy::Runtime::Scheduler.current
  if scheduler &&
      completion_queue.respond_to?(:expect_cross_thread_push)
    completion_queue.expect_cross_thread_push(scheduler)
  end

  event = Phronomy::Event.new(
    type: :start,
    target_id: SYSTEM_CHANNEL_ID,
    payload: {
      session: fsm_session,
      completion: completion_queue
    }
  )
  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_queue
end

#shutdown(deadline:, cancel_grace:) ⇒ Object 



220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/phronomy/engine/event_loop.rb', line 220

def shutdown(deadline:, cancel_grace:)
  @shutdown_mutex.synchronize do
    return @shutdown_status if @shutdown_status

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

    begin_draining
    if wait_until_idle(deadline)
      begin_stopping_if_idle
      join_until(deadline)
    end

    @shutdown_status =
      if task_alive?
        cancel_and_cleanup(cancel_grace)
      elsif state == :failed
        :failed
      else
        finalize_terminated(:terminated)
      end
  end
end

#state ⇒ Object 



191
192
193
 
# File 'lib/phronomy/engine/event_loop.rb', line 191

def state
  @lifecycle_mutex.synchronize { @state }
end

#task_alive? ⇒ Boolean

Returns:

  • (Boolean) 


247
248
249
 
# File 'lib/phronomy/engine/event_loop.rb', line 247

def task_alive?
  @task&.alive? || false
end

#wait_until_idle(deadline) ⇒ Object 



208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/phronomy/engine/event_loop.rb', line 208

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