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
 
# 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_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).



186
187
188
189
190
191
192
 
# File 'lib/patient_http/processor.rb', line 186

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) 


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

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

#draining? ⇒ Boolean

Check if processor is draining.

Returns:

  • (Boolean) 


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

def draining?
  @lifecycle.draining?
end

#enqueue(task) ⇒ void

This method returns an undefined value.

Enqueue a request task for processing.

Parameters:

Raises:



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

def enqueue(task)
  at_capacity = false

  @tasks_lock.synchronize do
    raise NotRunningError.new("Cannot enqueue request: processor is #{state}") unless running?

    # Check capacity - raise error if at max connections
    total = @queue.size + @pending_tasks.size + @inflight_requests.size
    if total >= @config.max_connections
      at_capacity = true
    else
      task.enqueued!
      @queue.push(task)
    end
  end

  if at_capacity
    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) 


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

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) 


286
287
288
 
# File 'lib/patient_http/processor.rb', line 286

def inflight_count
  @inflight_requests.size
end

#inflight_request_ids ⇒ Array<String>

Get the IDs of in-flight requests.

Returns:

  • (Array<String>) 


304
305
306
307
308
 
# File 'lib/patient_http/processor.rb', line 304

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:



314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/patient_http/processor.rb', line 314

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.



365
366
367
368
369
370
371
372
 
# File 'lib/patient_http/processor.rb', line 365

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

#running? ⇒ Boolean

Check if processor is running.

Returns:

  • (Boolean) 


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

def running?
  @lifecycle.running?
end

#start ⇒ void

This method returns an undefined value.

Start the processor.



46
47
48
49
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
 
# File 'lib/patient_http/processor.rb', line 46

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.
      orphaned_tasks = @tasks_lock.synchronize do
        if @reactor_generation == generation
          drain_tracked_tasks_locked
        else
          []
        end
      end
      reenqueue_tasks(orphaned_tasks)
    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) 


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

def starting?
  @lifecycle.starting?
end

#state ⇒ Symbol

Get the current processor state.

Returns:

  • (Symbol) —

    the current state



225
226
227
 
# File 'lib/patient_http/processor.rb', line 225

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)



111
112
113
114
115
116
117
118
119
120
121
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
 
# File 'lib/patient_http/processor.rb', line 111

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) 


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

def stopped?
  @lifecycle.stopped?
end

#stopping? ⇒ Boolean

Check if processor is stopping.

Returns:

  • (Boolean) 


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

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) 


295
296
297
298
299
 
# File 'lib/patient_http/processor.rb', line 295

def total_count
  @tasks_lock.synchronize do
    @queue.size + @pending_tasks.size + @inflight_requests.size
  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



346
347
348
 
# File 'lib/patient_http/processor.rb', line 346

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



355
356
357
358
359
 
# File 'lib/patient_http/processor.rb', line 355

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



335
336
337
338
 
# File 'lib/patient_http/processor.rb', line 335

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