Class: CI::Queue::Redis::Worker

Inherits:
Base
  • Object
show all
Defined in:
lib/ci/queue/redis/worker.rb

Direct Known Subclasses

Grind, TestTimeRecord

Constant Summary

Constants inherited from Base

Base::CONNECTION_ERRORS, Base::TEN_MINUTES

Constants included from Common

Common::CONNECTION_ERRORS

Instance Attribute Summary collapse

Attributes included from Common

#config

Instance Method Summary collapse

Methods inherited from Base

#boot_heartbeat_process!, #created_at=, #custom_config, #custom_middlewares, #ensure_heartbeat_thread_alive!, #exhausted?, #expired?, #increment_test_failed, #max_test_failed?, #progress, #queue_initialized?, #queue_initializing?, #reconnect_attempts, #remaining, #running, #size, #stop_heartbeat!, #test_failed, #to_a, #wait_for_master, #with_heartbeat, #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.



16
17
18
19
20
 
# File 'lib/ci/queue/redis/worker.rb', line 16

def initialize(redis, config)
  @reserved_test = nil
  @shutdown_required = false
  super(redis, config)
end

Instance Attribute Details

#total ⇒ Object (readonly)

Returns the value of attribute total.



14
15
16
 
# File 'lib/ci/queue/redis/worker.rb', line 14

def total
  @total
end

Instance Method Details

#acknowledge(test) ⇒ Object 



100
101
102
103
104
105
106
107
108
 
# File 'lib/ci/queue/redis/worker.rb', line 100

def acknowledge(test)
  test_key = test.id
  raise_on_mismatching_test(test_key)
  eval_script(
    :acknowledge,
    keys: [key('running'), key('processed'), key('owners')],
    argv: [test_key],
  ) == 1
end

#build ⇒ Object 



92
93
94
 
# File 'lib/ci/queue/redis/worker.rb', line 92

def build
  @build ||= CI::Queue::Redis::BuildRecord.new(self, redis, config)
end

#distributed? ⇒ Boolean

Returns:

  • (Boolean) 


22
23
24
 
# File 'lib/ci/queue/redis/worker.rb', line 22

def distributed?
  true
end

#master? ⇒ Boolean

Returns:

  • (Boolean) 


45
46
47
 
# File 'lib/ci/queue/redis/worker.rb', line 45

def master?
  @master
end

#poll ⇒ Object 



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/ci/queue/redis/worker.rb', line 49

def poll
  wait_for_master
  until shutdown_required? || config.circuit_breakers.any?(&:open?) || exhausted? || max_test_failed?
    if test = reserve
      yield index.fetch(test)
    else
      sleep 0.05
    end
  end
  redis.pipelined do |pipeline|
    pipeline.expire(key('worker', worker_id, 'queue'), config.redis_ttl)
    pipeline.expire(key('processed'), config.redis_ttl)
  end
rescue *CONNECTION_ERRORS
end

#populate(tests, random: Random.new) ⇒ Object 



26
27
28
29
30
31
 
# File 'lib/ci/queue/redis/worker.rb', line 26

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

Returns:

  • (Boolean) 


33
34
35
 
# File 'lib/ci/queue/redis/worker.rb', line 33

def populated?
  !!defined?(@index)
end

#release! ⇒ Object 



132
133
134
135
136
137
138
139
 
# File 'lib/ci/queue/redis/worker.rb', line 132

def release!
  eval_script(
    :release,
    keys: [key('running'), key('worker', worker_id, 'queue'), key('owners')],
    argv: [],
  )
  nil
end

#report_worker_error(error) ⇒ Object 



96
97
98
 
# File 'lib/ci/queue/redis/worker.rb', line 96

def report_worker_error(error)
  build.report_worker_error(error)
end

#requeue(test, offset: Redis.requeue_offset) ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/ci/queue/redis/worker.rb', line 110

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'),
      key('worker', worker_id, 'queue'),
      key('owners'),
    ],
    argv: [config.max_requeues, global_max_requeues, test_key, offset],
  ) == 1

  @reserved_test = test_key unless requeued
  requeued
end

#retry_queue ⇒ Object 



79
80
81
82
83
84
85
86
 
# File 'lib/ci/queue/redis/worker.rb', line 79

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

Returns:

  • (Boolean) 


66
67
68
69
70
 
# File 'lib/ci/queue/redis/worker.rb', line 66

def retrying?
  redis.exists?(key('worker', worker_id, 'queue'))
rescue *CONNECTION_ERRORS
  false
end

#shutdown! ⇒ Object 



37
38
39
 
# File 'lib/ci/queue/redis/worker.rb', line 37

def shutdown!
  @shutdown_required = true
end

#shutdown_required? ⇒ Boolean

Returns:

  • (Boolean) 


41
42
43
 
# File 'lib/ci/queue/redis/worker.rb', line 41

def shutdown_required?
  @shutdown_required
end

#supervisor ⇒ Object 



88
89
90
 
# File 'lib/ci/queue/redis/worker.rb', line 88

def supervisor
  Supervisor.new(redis_url, config)
end