Class: PatientHttp::CompletionExecutor Private

Inherits:
Object
  • Object
show all
Defined in:
lib/patient_http/completion_executor.rb

Overview

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

Fixed pool of worker threads that deliver completed request results.

The processor's reactor thread hands each finished HTTP exchange to this pool so response decoding, serialization, and callback delivery never block the event loop. Jobs are arbitrary callables consumed from a single queue.

Instance Method Summary collapse

Constructor Details

#initialize(threads:, logger: nil, thread_name_prefix: "patient-http-completion", on_finished: nil) ⇒ CompletionExecutor

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.

Initialize the executor and start its worker threads.

Parameters:

  • threads (Integer) —

    number of worker threads

  • logger (Logger, nil) (defaults to: nil) —

    logger for unexpected job errors

  • thread_name_prefix (String) (defaults to: "patient-http-completion") —

    prefix for worker thread names

  • on_finished (#call, nil) (defaults to: nil) —

    invoked after each job completes, outside any executor lock, so the owner can re-check idle conditions



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/patient_http/completion_executor.rb', line 20

def initialize(threads:, logger: nil, thread_name_prefix: "patient-http-completion", on_finished: nil)
  @queue = Thread::Queue.new
  @logger = logger
  @on_finished = on_finished
  @mutex = Mutex.new
  # Jobs enqueued but not yet fully executed. Tracked separately from the
  # queue size so a job that has been popped but is still running keeps
  # the executor non-idle.
  @outstanding = 0
  @threads = Array.new(threads) do |index|
    Thread.new do
      Thread.current.name = "#{thread_name_prefix}-#{index + 1}"
      run_worker
    end
  end
end

Instance Method Details

#enqueue(job) ⇒ void

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.

This method returns an undefined value.

Enqueue a job for execution.

Parameters:

  • job (#call) —

    the job to run

Raises:

  • (ClosedQueueError) —

    if the executor has been shut down



42
43
44
45
46
47
48
49
50
51
 
# File 'lib/patient_http/completion_executor.rb', line 42

def enqueue(job)
  @mutex.synchronize { @outstanding += 1 }
  begin
    @queue.push(job)
  rescue ClosedQueueError
    @mutex.synchronize { @outstanding -= 1 }
    raise
  end
  nil
end

#idle? ⇒ 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.

Check whether the executor has no queued or running jobs.

Returns:

  • (Boolean) 


56
57
58
 
# File 'lib/patient_http/completion_executor.rb', line 56

def idle?
  @mutex.synchronize { @outstanding == 0 }
end

#shutdown(timeout: 5) ⇒ void

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.

This method returns an undefined value.

Shut down the executor: close the queue so workers drain remaining jobs and exit, then join them within the timeout. Workers still alive after the deadline are killed; their tasks remain durably tracked and are recovered by the owner's re-enqueue logic.

Safe to call more than once and from a worker thread itself (the current thread is never joined or killed).

Parameters:

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

    seconds to wait for workers to drain



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/patient_http/completion_executor.rb', line 78

def shutdown(timeout: 5)
  @queue.close

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
  @threads.each do |thread|
    next if thread.equal?(Thread.current)

    remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
    thread.join(remaining.positive? ? remaining : 0)
    if thread.alive?
      thread.kill
      thread.join(1)
    end
  end

  discard_undrained_jobs
  nil
end

#worker_thread?(thread = Thread.current) ⇒ 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.

Check whether the given thread is one of this executor's workers.

Parameters:

  • thread (Thread) (defaults to: Thread.current) —

    the thread to check

Returns:

  • (Boolean) 


64
65
66
 
# File 'lib/patient_http/completion_executor.rb', line 64

def worker_thread?(thread = Thread.current)
  @threads.include?(thread)
end