Class: Chronos::Internal::WorkerPool

Inherits:
Object
  • Object
show all
Defined in:
lib/chronos/internal/worker_pool.rb

Overview

Fixed-size, lazy worker pool that drains a BoundedQueue.

Examples:

pool.enqueue(event)
pool.flush(2.0)

Constant Summary collapse

POLL_INTERVAL = 
0.05

Instance Method Summary collapse

Constructor Details

#initialize(queue, delivery, worker_count, logger = nil) ⇒ WorkerPool

Returns a new instance of WorkerPool.



19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/chronos/internal/worker_pool.rb', line 19

def initialize(queue, delivery, worker_count, logger = nil)
  @queue = queue
  @delivery = delivery
  @worker_count = worker_count
  @logger = logger || SafeLogger.new(nil)
  @mutex = Mutex.new
  @threads = []
  @active = 0
  @pending = 0
  @closed = false
  @pid = Process.pid
end

Instance Method Details

#close(timeout) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/chronos/internal/worker_pool.rb', line 54

def close(timeout)
  already_closed = @mutex.synchronize do
    was_closed = @closed
    @closed = true
    was_closed
  end
  return true if already_closed

  flushed = flush_without_reopening(timeout)
  @queue.close
  join_workers(timeout)
  flushed
rescue StandardError => error
  @logger.warn("Chronos worker shutdown failed: #{error.class}")
  false
end

#enqueue(event) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
 
# File 'lib/chronos/internal/worker_pool.rb', line 32

def enqueue(event)
  prepare_after_fork
  return false if closed?

  increment_pending
  accepted = @queue.push(event)
  decrement_pending unless accepted
  ensure_started if accepted
  accepted
end

#flush(timeout) ⇒ Object 



43
44
45
46
47
48
49
50
51
52
 
# File 'lib/chronos/internal/worker_pool.rb', line 43

def flush(timeout)
  prepare_after_fork
  ensure_started unless @queue.empty?
  deadline = monotonic_time + timeout.to_f
  loop do
    return true if delivery_complete?
    return false if monotonic_time >= deadline
    sleep(POLL_INTERVAL)
  end
end

#started? ⇒ Boolean

Returns:

  • (Boolean) 


71
72
73
 
# File 'lib/chronos/internal/worker_pool.rb', line 71

def started?
  @mutex.synchronize { !@threads.empty? }
end