class Sidekiq::Throttled::Strategy::Concurrency
Concurrency
throttling strategy
Constants
- SCRIPT
-
LUA script used to limit fetch concurrency. Logic behind the scene can be described in following pseudo code:
if @limit <= LLEN(@key) return 1 else PUSH(@key, @jid) return 0 end
Public Class Methods
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 31 def initialize(strategy_key, limit:, ttl: 900, key_suffix: nil) @base_key = "#{strategy_key}:concurrency.v2" @limit = limit @ttl = ttl.to_i @key_suffix = key_suffix end
@param [#to_s] strategy_key @param [#to_i, call] limit Amount of allowed concurrent jobs
per processors running for given key.
@param [#to_i] ttl Concurrency
lock TTL in seconds. @param [Proc] key_suffix Dynamic key suffix generator.
Public Instance Methods
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 68 def count(*job_args) Sidekiq.redis { |conn| conn.zcard(key(job_args)) }.to_i end
@return [Integer] Current count of jobs
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 39 def dynamic? @key_suffix || @limit.respond_to?(:call) end
@return [Boolean] Whenever strategy has dynamic config
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 80 def finalize!(jid, *job_args) Sidekiq.redis { |conn| conn.zrem(key(job_args), jid.to_s) } end
Remove jid from the pool of jobs in progress @return [void]
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 74 def reset!(*job_args) Sidekiq.redis { |conn| conn.del(key(job_args)) } end
Resets count of jobs @return [void]
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 56 def retry_in(_jid, *job_args) job_limit = limit(job_args) return 0.0 if !job_limit || count(*job_args) < job_limit oldest_jid_with_score = Sidekiq.redis { |redis| redis.zrange(key(job_args), 0, 0, withscores: true) }.first return 0.0 unless oldest_jid_with_score expiry_time = oldest_jid_with_score.last.to_f expiry_time - Time.now.to_f end
@return [Float] How long, in seconds, before we’ll next be able to take on jobs
Source
# File lib/sidekiq/throttled/strategy/concurrency.rb, line 44 def throttled?(jid, *job_args) job_limit = limit(job_args) return false unless job_limit return true if job_limit <= 0 keys = [key(job_args)] argv = [jid.to_s, job_limit, @ttl, Time.now.to_f] Sidekiq.redis { |redis| 1 == SCRIPT.call(redis, keys: keys, argv: argv) } end
@return [Boolean] whenever job is throttled or not