Class: CI::Queue::Redis::Base

Inherits:
Object
  • Object
show all
Includes:
Common
Defined in:
lib/ci/queue/redis/base.rb

Direct Known Subclasses

Supervisor, Worker

Constant Summary collapse

TEN_MINUTES = 
60 * 10
CONNECTION_ERRORS = 
[
  ::Redis::BaseConnectionError,
  ::SocketError, # https://github.com/redis/redis-rb/pull/631
].freeze

Instance Attribute Summary

Attributes included from Common

#config

Instance Method Summary collapse

Methods included from Common

#distributed?, #flaky?, #release!, #report_failure!, #report_success!, #rescue_connection_errors, #retrying?

Constructor Details

#initialize(redis_url, config) ⇒ Base

Returns a new instance of Base.



14
15
16
17
18
19
20
21
22
23
 
# File 'lib/ci/queue/redis/base.rb', line 14

def initialize(redis_url, config)
  @redis_url = redis_url
  @redis = ::Redis.new(
    url: redis_url,
    # Booting a CI worker is costly, so in case of a Redis blip,
    # it makes sense to retry for a while before giving up.
    reconnect_attempts: [0, 0, 0.1, 0.5, 1, 3, 5],
  )
  @config = config
end

Instance Method Details

#created_at=(timestamp) ⇒ Object 



38
39
40
 
# File 'lib/ci/queue/redis/base.rb', line 38

def created_at=(timestamp)
  redis.setnx(key('created-at'), timestamp)
end

#exhausted? ⇒ Boolean

Returns:

  • (Boolean) 


25
26
27
 
# File 'lib/ci/queue/redis/base.rb', line 25

def exhausted?
  queue_initialized? && size == 0
end

#expired? ⇒ Boolean

Returns:

  • (Boolean) 


29
30
31
32
33
34
35
36
 
# File 'lib/ci/queue/redis/base.rb', line 29

def expired?
  if (created_at = redis.get(key('created-at')))
    (created_at.to_f + config.redis_ttl + TEN_MINUTES) < Time.now.to_f
  else
    # if there is no created at set anymore we assume queue is expired
    true
  end
end

#increment_test_failed ⇒ Object 



87
88
89
 
# File 'lib/ci/queue/redis/base.rb', line 87

def increment_test_failed
  redis.incr(key('test_failed_count'))
end

#max_test_failed? ⇒ Boolean

Returns:

  • (Boolean) 


95
96
97
98
99
 
# File 'lib/ci/queue/redis/base.rb', line 95

def max_test_failed?
  return false if config.max_test_failed.nil?

  test_failed >= config.max_test_failed
end

#progress ⇒ Object 



56
57
58
 
# File 'lib/ci/queue/redis/base.rb', line 56

def progress
  total - size
end

#queue_initialized? ⇒ Boolean

Returns:

  • (Boolean) 


76
77
78
79
80
81
 
# File 'lib/ci/queue/redis/base.rb', line 76

def queue_initialized?
  @queue_initialized ||= begin
    status = master_status
    status == 'ready' || status == 'finished'
  end
end

#queue_initializing? ⇒ Boolean

Returns:

  • (Boolean) 


83
84
85
 
# File 'lib/ci/queue/redis/base.rb', line 83

def queue_initializing?
  master_status == 'setup'
end

#size ⇒ Object 



42
43
44
45
46
47
 
# File 'lib/ci/queue/redis/base.rb', line 42

def size
  redis.multi do |transaction|
    transaction.llen(key('queue'))
    transaction.zcard(key('running'))
  end.inject(:+)
end

#test_failed ⇒ Object 



91
92
93
 
# File 'lib/ci/queue/redis/base.rb', line 91

def test_failed
  redis.get(key('test_failed_count')).to_i
end

#to_a ⇒ Object 



49
50
51
52
53
54
 
# File 'lib/ci/queue/redis/base.rb', line 49

def to_a
  redis.multi do |transaction|
    transaction.lrange(key('queue'), 0, -1)
    transaction.zrange(key('running'), 0, -1)
  end.flatten.reverse.map { |k| index.fetch(k) }
end

#wait_for_master(timeout: 30) ⇒ Object

Raises:



60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/ci/queue/redis/base.rb', line 60

def wait_for_master(timeout: 30)
  return true if master?
  (timeout * 10 + 1).to_i.times do
    if queue_initialized?
      return true
    else
      sleep 0.1
    end
  end
  raise LostMaster, "The master worker is still `#{master_status}` after #{timeout} seconds waiting."
end

#workers_count ⇒ Object 



72
73
74
 
# File 'lib/ci/queue/redis/base.rb', line 72

def workers_count
  redis.scard(key('workers'))
end