Class: PatientHttp::Processor

Inherits:
Object
  • Object
show all
Includes:
RedirectHelper, TimeHelper
Defined in:
lib/patient_http/processor.rb

Overview

Core processor that handles async HTTP requests in a dedicated thread

Constant Summary collapse

DEQUEUE_TIMEOUT =

Timing constants for the reactor loop

1.0

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from TimeHelper

#monotonic_time, #wall_clock_time

Constructor Details

#initialize(config) ⇒ void

Initialize the processor.

Parameters:



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/patient_http/processor.rb', line 23

def initialize(config)
  @config = config
  @lifecycle = LifecycleManager.new
  @queue = Thread::Queue.new
  @reactor_thread = nil
  # Serializes start/stop so a start cannot interleave with a stop that is
  # still reaping its reactor thread (and vice versa).
  @lifecycle_mutex = Mutex.new
  # Incremented once per reactor run; lets a reactor's teardown detect
  # whether it is still the current run before mutating shared state.
  @reactor_generation = 0
  @inflight_requests = Concurrent::Hash.new
  @pending_tasks = Concurrent::Hash.new
  # Tasks pushed onto @queue but not yet popped by the reactor. Kept in a
  # hash because Thread::Queue cannot be enumerated; used to report all
  # tracked task ids (e.g. for heartbeat updates on queued tasks).
  @queued_tasks = Concurrent::Hash.new
  @tasks_lock = Mutex.new
  @idle_condition = ConditionVariable.new
  @testing_callback = nil
  @http_client = Client.new(self)
  @observers = []
end

Instance Attribute Details

#config ⇒ Configuration (readonly)

Returns the configuration object for the processor.

Returns:



13
14
15
 
# File 'lib/patient_http/processor.rb', line 13

def config
  @config
end

#testing_callback ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Callback to invoke after each request. Only available in testing mode.



17
18
19
 
# File 'lib/patient_http/processor.rb', line 17

def testing_callback
  @testing_callback
end

Instance Method Details

#drain ⇒ void

This method returns an undefined value.

Drain the processor (stop accepting new requests).



197
198
199
200
201
202
203
 
# File 'lib/patient_http/processor.rb', line 197

def drain
  @tasks_lock.synchronize do
    return unless @lifecycle.drain!
  end

  @config.logger&.info("[PatientHttp] Processor draining (no longer accepting new requests)")
end

#drained? ⇒ Boolean

Check if processor is drained (draining and idle).

Returns:

  • (Boolean) 


267
268
269
 
# File 'lib/patient_http/processor.rb', line 267

def drained?
  @lifecycle.draining? && idle?
end

#draining? ⇒ Boolean

Check if processor is draining.

Returns:

  • (Boolean) 


260
261
262
 
# File 'lib/patient_http/processor.rb', line 260

def draining?
  @lifecycle.draining?
end

#enqueue(task) ⇒ void

This method returns an undefined value.

Enqueue a request task for processing.

Parameters:

Raises:



211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/patient_http/processor.rb', line 211

def enqueue(task)
  raise NotRunningError.new("Cannot enqueue request: processor is #{state}") unless running?

  accepted = announce_and_enqueue(task) do
    # The pre-check above is advisory; re-check the running state under
    # the lock since observers were notified outside of it.
    raise NotRunningError.new("Cannot enqueue request: processor is #{state}") unless running?

    # Check capacity - the task is only accepted below max connections.
    @queue.size + @pending_tasks.size + @inflight_requests.size < @config.max_connections
  end

  unless accepted
    notify_observers { |observer| observer.capacity_exceeded }
    raise MaxCapacityError.new("Cannot enqueue request: already at max capacity (#{@config.max_connections} connections)")
  end
end

#idle? ⇒ Boolean

Check if processor is idle (no queued or in-flight requests).

Returns:

  • (Boolean) 


281
282
283
284
285
 
# File 'lib/patient_http/processor.rb', line 281

def idle?
  @tasks_lock.synchronize do
    @queue.empty? && @pending_tasks.empty? && @inflight_requests.empty?
  end
end

#inflight_count ⇒ Integer

Get the number of in-flight requests (actively executing HTTP calls).

This does not include queued or pending tasks. For the total pipeline count used by the capacity check, see #total_count.

Returns:

  • (Integer) 


293
294
295
 
# File 'lib/patient_http/processor.rb', line 293

def inflight_count
  @inflight_requests.size
end

#inflight_request_ids ⇒ Array<String>

Get the IDs of in-flight requests.

Returns:

  • (Array<String>) 


311
312
313
314
315
 
# File 'lib/patient_http/processor.rb', line 311

def inflight_request_ids
  @tasks_lock.synchronize do
    @inflight_requests.keys
  end
end

#observe(observer) ⇒ void

This method returns an undefined value.

Add an observer for processor events.

Parameters:



332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
 
# File 'lib/patient_http/processor.rb', line 332

def observe(observer)
  notify_start = false

  @tasks_lock.synchronize do
    raise ArgumentError.new("Observer already added") if @observers.include?(observer)

    @observers << observer
    # Only self-notify when already running. An observer added while the
    # processor is still starting is picked up by start's atomic observer
    # snapshot, so notifying here too would deliver start twice.
    notify_start = running?
  end

  notify_observer(observer) { |o| o.start } if notify_start
end

#run ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Run the processor in a block. This is intended for use in tests to ensure the processor is started and stopped properly.



383
384
385
386
387
388
389
390
 
# File 'lib/patient_http/processor.rb', line 383

def run
  start
  wait_for_running
  yield
ensure
  stop(timeout: 0)
  wait_for_idle
end

#running? ⇒ Boolean

Check if processor is running.

Returns:

  • (Boolean) 


246
247
248
 
# File 'lib/patient_http/processor.rb', line 246

def running?
  @lifecycle.running?
end

#start ⇒ void

This method returns an undefined value.

Start the processor.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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
110
111
112
113
114
115
116
 
# File 'lib/patient_http/processor.rb', line 50

def start
  observers_to_notify = nil

  # Hold the lifecycle mutex across the whole start so a concurrent stop
  # cannot interleave with (and reap) the reactor thread we are creating.
  @lifecycle_mutex.synchronize do
    # Claim this reactor run's generation atomically with the state
    # transition. The reactor thread captures it below and its teardown
    # only mutates shared state while it is still the current generation.
    generation = @tasks_lock.synchronize do
      return unless @lifecycle.start!
      @reactor_generation += 1
    end

    @reactor_thread = Thread.new do
      Thread.current.name = "patient-http-processor"
      run_reactor
    rescue => e
      @config.logger&.error("[PatientHttp] Processor error: #{e.message}\n#{e.backtrace.join("\n")}")

      raise if PatientHttp.testing?
    ensure
      # Mark the processor stopped when the reactor exits and re-enqueue any
      # tasks still being tracked, so a reactor that exits without a stop()
      # call (e.g. an unhandled error) does not lose in-flight/pending
      # requests or leak stale tracking entries into a later run.
      #
      # Only act while this is still the current generation: a newer start
      # (after a stop) owns the processor state and a stale reactor from a
      # prior run must not clobber it. Snapshot and clear happen under the
      # lock; re-enqueueing runs outside it. This is idempotent with stop()'s
      # reenqueue_pending_requests: whichever runs second snapshots an empty
      # set.
      current_generation = false
      orphaned_tasks = @tasks_lock.synchronize do
        if @reactor_generation == generation
          current_generation = true
          drain_tracked_tasks_locked
        else
          []
        end
      end
      reenqueue_tasks(orphaned_tasks)
      # Hand back tasks still sitting in the queue as well; a reactor that
      # exits without a stop() call is the last owner of those tasks.
      # stop() performs the same drain after reaping the reactor, and
      # whichever drain runs second finds nothing left.
      reenqueue_remaining_queue_items if current_generation
    end

    # The transition can fail if the reactor thread already failed and
    # marked the processor stopped. Capture the observer snapshot under the
    # same lock as the transition so an observer registered concurrently via
    # #observe is notified of start by exactly one path (here or in #observe).
    started, observers = @tasks_lock.synchronize do
      [@lifecycle.running!, @observers.dup]
    end
    observers_to_notify = observers if started

    # Block until the reactor is ready
    @lifecycle.wait_for_reactor(timeout: 5)
  end

  # Notify observers outside the lifecycle mutex so an observer callback
  # that re-enters the processor cannot deadlock.
  observers_to_notify&.each { |observer| notify_observer(observer) { |o| o.start } }
end

#starting? ⇒ Boolean

Check if processor is starting.

Returns:

  • (Boolean) 


239
240
241
 
# File 'lib/patient_http/processor.rb', line 239

def starting?
  @lifecycle.starting?
end

#state ⇒ Symbol

Get the current processor state.

Returns:

  • (Symbol) —

    the current state



232
233
234
 
# File 'lib/patient_http/processor.rb', line 232

def state
  @lifecycle.state
end

#stop(timeout: nil) ⇒ void

This method returns an undefined value.

Stop the processor.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    how long to wait for in-flight requests (seconds)



122
123
124
125
126
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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/patient_http/processor.rb', line 122

def stop(timeout: nil)
  timeout ||= @config.shutdown_timeout
  should_notify_stop = false

  # Hold the lifecycle mutex across the whole stop so a concurrent start
  # cannot begin (and reassign @reactor_thread) while we are tearing down.
  @lifecycle_mutex.synchronize do
    # Atomically transition to stopping and capture the reactor thread for
    # this run. Joining/killing the captured reference rather than the ivar
    # means we can never tear down a reactor from a different run.
    reactor = @tasks_lock.synchronize do
      return unless @lifecycle.stop!
      @reactor_thread
    end

    # Interrupt the reactor's queue wait by pushing a sentinel value
    @queue.push(nil)

    # Wait for in-flight and pending requests to complete.
    # Queue items are not checked here — they will be re-enqueued by
    # reenqueue_remaining_queue_items after the reactor thread exits.
    if timeout > 0
      deadline = monotonic_time + timeout
      @tasks_lock.synchronize do
        loop do
          break if @pending_tasks.empty? && @inflight_requests.empty?
          remaining = deadline - monotonic_time
          break if remaining <= 0
          @idle_condition.wait(@tasks_lock, remaining)
        end
      end
    end

    reenqueue_pending_requests

    # Reap the reactor thread — unless stop was called from the reactor
    # thread itself (e.g. from a task callback or observer), where joining
    # the current thread would raise ThreadError. In that case the reactor
    # exits on its own once the callback returns (its loop sees the stopped
    # state) and its ensure block performs the same cleanup.
    if reactor && !reactor.equal?(Thread.current)
      reactor.join(1) if reactor.alive?
      if reactor.alive?
        reactor.kill
        # Wait for the killed thread's ensure blocks so a stale lifecycle
        # transition cannot fire during a subsequent start.
        reactor.join(1)
      end
    end
    @tasks_lock.synchronize do
      @reactor_thread = nil if @reactor_thread.equal?(reactor)
    end

    # Run a second pass now that the reactor has exited to catch any task
    # that slipped into pending/in-flight tracking after the first snapshot
    # (a task can be popped from the queue but not yet tracked when the
    # snapshot is taken).
    reenqueue_pending_requests

    # Drain any items left in the queue after the reactor has exited.
    # This must happen after the reactor thread is done to avoid consuming
    # the nil sentinel that wakes the reactor.
    reenqueue_remaining_queue_items

    should_notify_stop = true
  end

  # Notify observers outside the lifecycle mutex so an observer callback
  # that re-enters the processor cannot deadlock.
  notify_observers { |observer| observer.stop } if should_notify_stop
end

#stopped? ⇒ Boolean

Check if processor is stopped.

Returns:

  • (Boolean) 


253
254
255
 
# File 'lib/patient_http/processor.rb', line 253

def stopped?
  @lifecycle.stopped?
end

#stopping? ⇒ Boolean

Check if processor is stopping.

Returns:

  • (Boolean) 


274
275
276
 
# File 'lib/patient_http/processor.rb', line 274

def stopping?
  @lifecycle.stopping?
end

#total_count ⇒ Integer

Get the total number of tasks in the pipeline (queued + pending + in-flight).

This is the count used by #enqueue for capacity enforcement.

Returns:

  • (Integer) 


302
303
304
305
306
 
# File 'lib/patient_http/processor.rb', line 302

def total_count
  @tasks_lock.synchronize do
    @queue.size + @pending_tasks.size + @inflight_requests.size
  end
end

#tracked_request_ids ⇒ Array<String>

Get the IDs of all tasks in the pipeline (queued, pending, and in-flight). Use this to keep durable tracking (e.g. heartbeats) alive for tasks the processor has accepted but not yet started.

Returns:

  • (Array<String>) 


322
323
324
325
326
 
# File 'lib/patient_http/processor.rb', line 322

def tracked_request_ids
  @tasks_lock.synchronize do
    (@queued_tasks.keys + @pending_tasks.keys + @inflight_requests.keys).uniq
  end
end

#wait_for_idle(timeout: 1) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Wait for the queue to be empty and all in-flight requests to complete. This is mainly for use in tests.

Parameters:

  • timeout (Numeric) (defaults to: 1) —

    maximum time to wait in seconds (default: 5)

Returns:

  • (Boolean) —

    true if processing completed, false if timeout reached



364
365
366
 
# File 'lib/patient_http/processor.rb', line 364

def wait_for_idle(timeout: 1)
  @lifecycle.wait_for_condition(timeout: timeout) { idle? }
end

#wait_for_processing(timeout: 1) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Wait for at least one request to start processing. This is mainly for use in tests.

Parameters:

  • timeout (Numeric) (defaults to: 1) —

    maximum time to wait in seconds (default: 5)

Returns:

  • (Boolean) —

    true if a request started processing, false if timeout reached



373
374
375
376
377
 
# File 'lib/patient_http/processor.rb', line 373

def wait_for_processing(timeout: 1)
  @lifecycle.wait_for_condition(timeout: timeout) do
    !@inflight_requests.empty? || !@pending_tasks.empty?
  end
end

#wait_for_running(timeout: 5) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Wait for the processor to start.

Parameters:

  • timeout (Numeric) (defaults to: 5) —

    maximum time to wait in seconds (default: 5)

Returns:

  • (Boolean) —

    true if started, false if timeout reached



353
354
355
356
 
# File 'lib/patient_http/processor.rb', line 353

def wait_for_running(timeout: 5)
  start
  @lifecycle.wait_for_running(timeout: timeout)
end