Class: HastCI::TaskBuffer

Inherits:
Object
  • Object
show all
Defined in:
lib/hastci/task_buffer.rb

Instance Method Summary collapse

Constructor Details

#initialize(min_size:, max_size:, fetcher:, error_collector:, poll_interval: DEFAULT_POLL_INTERVAL, on_cancelled: nil, sleeper: HastCI::DEFAULT_SLEEPER) ⇒ TaskBuffer

Returns a new instance of TaskBuffer.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/hastci/task_buffer.rb', line 10

def initialize(min_size:, max_size:, fetcher:, error_collector:, poll_interval: DEFAULT_POLL_INTERVAL,
  on_cancelled: nil, sleeper: HastCI::DEFAULT_SLEEPER)
  @min_size = min_size
  @max_size = max_size
  @fetcher = fetcher
  @error_collector = error_collector
  @poll_interval = poll_interval
  @on_cancelled = on_cancelled
  @sleeper = sleeper

  @queue = SizedQueue.new(max_size)
  @thread = nil
  @running = false
  @mutex = Mutex.new
  @prefetch_condition = ConditionVariable.new
  @drained = false
  @cancelled = false
end

Instance Method Details

#cancelled? ⇒ Boolean

Returns:

  • (Boolean) 


85
86
87
 
# File 'lib/hastci/task_buffer.rb', line 85

def cancelled?
  @mutex.synchronize { @cancelled }
end

#drained? ⇒ Boolean

Returns:

  • (Boolean) 


81
82
83
 
# File 'lib/hastci/task_buffer.rb', line 81

def drained?
  @drained && @queue.empty?
end

#next_task ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/hastci/task_buffer.rb', line 56

def next_task
  loop do
    error = @error_collector.first_error
    raise error if error

    return nil if queue_closed_and_empty?

    begin
      task = @queue.pop(true)
    rescue ThreadError
      @sleeper.call(@poll_interval)
      next
    end

    return nil if task.nil?

    signal_prefetch_if_needed
    return task
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


52
53
54
 
# File 'lib/hastci/task_buffer.rb', line 52

def running?
  @mutex.synchronize { @running }
end

#size ⇒ Object 



77
78
79
 
# File 'lib/hastci/task_buffer.rb', line 77

def size
  @queue.size
end

#start ⇒ Object 



29
30
31
32
33
34
35
36
37
38
 
# File 'lib/hastci/task_buffer.rb', line 29

def start
  @mutex.synchronize do
    return if @running

    @running = true
    @drained = false
    @cancelled = false
    @thread = Thread.new { prefetch_loop }
  end
end

#stop ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/hastci/task_buffer.rb', line 40

def stop
  @mutex.synchronize do
    return unless @running

    @running = false
    @prefetch_condition.signal
  end

  @thread.join(SHUTDOWN_TIMEOUT)
  @queue.close
end