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
COMPLETION_RETRY_DELAY =

Base delay between attempts when delivering a completed result fails. The delay grows linearly with each attempt.

0.5
COMPLETION_SHUTDOWN_TIMEOUT =

Seconds allowed for the completion executor to drain during shutdown. The reactor's teardown and stop() share this budget so the reactor can never spend longer draining than stop() is willing to wait for it.

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from TimeHelper

#monotonic_time, #wall_clock_time

Constructor Details

#initialize(config, name: "default") ⇒ void

Initialize the processor.

Parameters:

  • config (Configuration) —

    the configuration object

  • name (String, Symbol) (defaults to: "default") —

    optional name to distinguish this processor when a process runs more than one



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/patient_http/processor.rb', line 38

def initialize(config, name: "default")
  @config = config
  @name = name.to_s
  @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 = []
  @completion_executor = nil
end

Instance Attribute Details

#config ⇒ Configuration (readonly)

Returns the configuration object for the processor.

Returns:



22
23
24
 
# File 'lib/patient_http/processor.rb', line 22

def config
  @config
end

#name ⇒ String (readonly)

Returns the processor's name; used in thread names so multiple named processors in one process are distinguishable.

Returns:

  • (String) —

    the processor's name; used in thread names so multiple named processors in one process are distinguishable



26
27
28
 
# File 'lib/patient_http/processor.rb', line 26

def name
  @name
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.



30
31
32
 
# File 'lib/patient_http/processor.rb', line 30

def testing_callback
  @testing_callback
end

Instance Method Details

#capacity_available? ⇒ Boolean

Check if the processor can accept at least one more request. Advisory only; see #remaining_capacity.

Returns:

  • (Boolean) 


361
362
363
 
# File 'lib/patient_http/processor.rb', line 361

def capacity_available?
  remaining_capacity > 0
end

#drain ⇒ void

This method returns an undefined value.

Drain the processor (stop accepting new requests).



249
250
251
252
253
254
255
 
# File 'lib/patient_http/processor.rb', line 249

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) 


319
320
321
 
# File 'lib/patient_http/processor.rb', line 319

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

#draining? ⇒ Boolean

Check if processor is draining.

Returns:

  • (Boolean) 


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

def draining?
  @lifecycle.draining?
end

#enqueue(task) ⇒ void

This method returns an undefined value.

Enqueue a request task for processing.

Parameters:

Raises:



263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
 
# File 'lib/patient_http/processor.rb', line 263

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, and no results still being delivered by the completion executor).

Returns:

  • (Boolean) 


334
335
336
337
338
339
340
341
 
# File 'lib/patient_http/processor.rb', line 334

def idle?
  executor = @completion_executor
  tracking_empty = @tasks_lock.synchronize do
    @queue.empty? && @pending_tasks.empty? && @inflight_requests.empty?
  end

  tracking_empty && (executor.nil? || executor.idle?)
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) 


371
372
373
 
# File 'lib/patient_http/processor.rb', line 371

def inflight_count
  @inflight_requests.size
end

#inflight_request_ids ⇒ Array<String>

Get the IDs of in-flight requests.

Returns:

  • (Array<String>) 


389
390
391
392
393
 
# File 'lib/patient_http/processor.rb', line 389

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:



410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
 
# File 'lib/patient_http/processor.rb', line 410

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

#remaining_capacity ⇒ Integer

Check how many more requests the processor can accept before reaching max capacity. This is an advisory value: the authoritative check happens inside #enqueue, so a concurrent enqueue can still hit MaxCapacityError. It performs no observer notifications and no durable registration, so it is cheap to call before paying enqueue costs.

Returns:

  • (Integer) —

    remaining capacity (never negative)



350
351
352
353
354
355
 
# File 'lib/patient_http/processor.rb', line 350

def remaining_capacity
  @tasks_lock.synchronize do
    remaining = @config.max_connections - (@queue.size + @pending_tasks.size + @inflight_requests.size)
    (remaining > 0) ? remaining : 0
  end
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.



461
462
463
464
465
466
467
468
 
# File 'lib/patient_http/processor.rb', line 461

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

#running? ⇒ Boolean

Check if processor is running.

Returns:

  • (Boolean) 


298
299
300
 
# File 'lib/patient_http/processor.rb', line 298

def running?
  @lifecycle.running?
end

#start ⇒ void

This method returns an undefined value.

Start the processor.



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
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
 
# File 'lib/patient_http/processor.rb', line 67

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

    # The completion executor delivers finished results on its own worker
    # threads so the reactor thread never blocks on response decoding,
    # serialization, or callback delivery. A new executor is created for
    # each run, like the reactor thread.
    executor = CompletionExecutor.new(
      threads: @config.completion_threads,
      logger: @config.logger,
      thread_name_prefix: thread_name("patient-http-completion"),
      on_finished: -> { signal_idle }
    )
    @tasks_lock.synchronize { @completion_executor = executor }

    @reactor_thread = Thread.new do
      Thread.current.name = thread_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 = @tasks_lock.synchronize { @reactor_generation == generation }
      if current_generation
        begin
          # Drain the completion executor before stealing tracked tasks so
          # results already handed off are delivered rather than retried.
          # Tasks whose completion job never ran stay in in-flight tracking
          # and are re-enqueued below. The drain is bounded so it cannot
          # outlast stop()'s own shutdown budget, and it runs in its own
          # block so the re-enqueue still happens if a stop() that gave up
          # waiting kills this thread mid-drain.
          executor.shutdown(timeout: COMPLETION_SHUTDOWN_TIMEOUT)
        ensure
          orphaned_tasks = @tasks_lock.synchronize do
            if @reactor_generation == generation
              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
        end
      end
    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) 


291
292
293
 
# File 'lib/patient_http/processor.rb', line 291

def starting?
  @lifecycle.starting?
end

#state ⇒ Symbol

Get the current processor state.

Returns:

  • (Symbol) —

    the current state



284
285
286
 
# File 'lib/patient_http/processor.rb', line 284

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)



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
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
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
 
# File 'lib/patient_http/processor.rb', line 163

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, including
    # results still being delivered by the completion executor.
    # 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? && completion_executor_settled?
          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)
      # The join stays short: the reactor's teardown drains the completion
      # executor, which can join this very thread when stop was called from
      # a completion callback. Killing the reactor breaks that standoff and
      # its teardown still re-enqueues from an ensure block.
      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

    # Shut down the completion executor. The reactor's own teardown
    # normally drains it already; this pass reaps any worker that is still
    # stuck past the deadline. Remaining queued jobs belong to tasks that
    # were re-enqueued above, so they no-op when their claim fails.
    @completion_executor&.shutdown(timeout: COMPLETION_SHUTDOWN_TIMEOUT)

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


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

def stopped?
  @lifecycle.stopped?
end

#stopping? ⇒ Boolean

Check if processor is stopping.

Returns:

  • (Boolean) 


326
327
328
 
# File 'lib/patient_http/processor.rb', line 326

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) 


380
381
382
383
384
 
# File 'lib/patient_http/processor.rb', line 380

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


400
401
402
403
404
 
# File 'lib/patient_http/processor.rb', line 400

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



442
443
444
 
# File 'lib/patient_http/processor.rb', line 442

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



451
452
453
454
455
 
# File 'lib/patient_http/processor.rb', line 451

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



431
432
433
434
 
# File 'lib/patient_http/processor.rb', line 431

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