Class: CI::Queue::Redis::Worker
- Inherits:
-
Base
- Object
- Base
- CI::Queue::Redis::Worker
show all
- Defined in:
- lib/ci/queue/redis/worker.rb
Constant Summary
Constants inherited
from Base
Base::CONNECTION_ERRORS
Constants included
from Common
Common::CONNECTION_ERRORS
Instance Attribute Summary collapse
Attributes included from Common
#config
Instance Method Summary
collapse
Methods inherited from Base
#exhausted?, #progress, #queue_initialized?, #size, #to_a, #wait_for_master, #workers_count
Methods included from Common
#flaky?, #report_failure!, #report_success!, #rescue_connection_errors
Constructor Details
#initialize(redis, config) ⇒ Worker
Returns a new instance of Worker.
17
18
19
20
21
|
# File 'lib/ci/queue/redis/worker.rb', line 17
def initialize(redis, config)
@reserved_test = nil
@shutdown_required = false
super(redis, config)
end
|
Instance Attribute Details
#total ⇒ Object
Returns the value of attribute total.
15
16
17
|
# File 'lib/ci/queue/redis/worker.rb', line 15
def total
@total
end
|
Instance Method Details
#acknowledge(test) ⇒ Object
81
82
83
84
85
86
87
88
89
|
# File 'lib/ci/queue/redis/worker.rb', line 81
def acknowledge(test)
test_key = test.id
raise_on_mismatching_test(test_key)
eval_script(
:acknowledge,
keys: [key('running'), key('processed')],
argv: [test_key],
) == 1
end
|
#build ⇒ Object
77
78
79
|
# File 'lib/ci/queue/redis/worker.rb', line 77
def build
@build ||= CI::Queue::Redis::BuildRecord.new(self, redis, config)
end
|
#master? ⇒ Boolean
42
43
44
|
# File 'lib/ci/queue/redis/worker.rb', line 42
def master?
@master
end
|
#poll ⇒ Object
46
47
48
49
50
51
52
53
54
55
56
|
# File 'lib/ci/queue/redis/worker.rb', line 46
def poll
wait_for_master
until shutdown_required? || config.circuit_breakers.any?(&:open?) || exhausted?
if test = reserve
yield index.fetch(test), @last_warning
else
sleep 0.05
end
end
rescue *CONNECTION_ERRORS
end
|
#populate(tests, random: Random.new) ⇒ Object
23
24
25
26
27
28
|
# File 'lib/ci/queue/redis/worker.rb', line 23
def populate(tests, random: Random.new)
@index = tests.map { |t| [t.id, t] }.to_h
tests = Queue.shuffle(tests, random)
push(tests.map(&:id))
self
end
|
#populated? ⇒ Boolean
30
31
32
|
# File 'lib/ci/queue/redis/worker.rb', line 30
def populated?
!!defined?(@index)
end
|
#requeue(test, offset: Redis.requeue_offset) ⇒ Object
91
92
93
94
95
96
97
98
99
100
101
102
103
104
|
# File 'lib/ci/queue/redis/worker.rb', line 91
def requeue(test, offset: Redis.requeue_offset)
test_key = test.id
raise_on_mismatching_test(test_key)
global_max_requeues = config.global_max_requeues(total)
requeued = config.max_requeues > 0 && global_max_requeues > 0 && eval_script(
:requeue,
keys: [key('processed'), key('requeues-count'), key('queue'), key('running')],
argv: [config.max_requeues, global_max_requeues, test_key, offset],
) == 1
@reserved_test = test_key unless requeued
requeued
end
|
#retry_queue ⇒ Object
64
65
66
67
68
69
70
71
|
# File 'lib/ci/queue/redis/worker.rb', line 64
def retry_queue
failures = build.failed_tests.to_set
log = redis.lrange(key('worker', worker_id, 'queue'), 0, -1)
log.select! { |id| failures.include?(id) }
log.uniq!
log.reverse!
Retry.new(log, config, redis: redis)
end
|
#retrying? ⇒ Boolean
58
59
60
61
62
|
# File 'lib/ci/queue/redis/worker.rb', line 58
def retrying?
redis.exists(key('worker', worker_id, 'queue'))
rescue *CONNECTION_ERRORS
false
end
|
#shutdown! ⇒ Object
34
35
36
|
# File 'lib/ci/queue/redis/worker.rb', line 34
def shutdown!
@shutdown_required = true
end
|
#shutdown_required? ⇒ Boolean
38
39
40
|
# File 'lib/ci/queue/redis/worker.rb', line 38
def shutdown_required?
@shutdown_required
end
|
#supervisor ⇒ Object
73
74
75
|
# File 'lib/ci/queue/redis/worker.rb', line 73
def supervisor
Supervisor.new(redis_url, config)
end
|