module Sidekiq::LimitFetch::Global class Semaphore include Sidekiq::LimitFetch::Redis PREFIX = 'limit_fetch' attr_reader :local_busy def initialize(name) @name = name @lock = Mutex.new @local_busy = 0 end def limit value = redis {|it| it.get "#{PREFIX}:limit:#@name" } value.to_i if value end def limit=(value) @limit_changed = true if value redis {|it| it.set "#{PREFIX}:limit:#@name", value } else redis {|it| it.del "#{PREFIX}:limit:#@name" } end end def limit_changed? @limit_changed end def process_limit value = redis {|it| it.get "#{PREFIX}:process_limit:#@name" } value.to_i if value end def process_limit=(value) if value redis {|it| it.set "#{PREFIX}:process_limit:#@name", value } else redis {|it| it.del "#{PREFIX}:process_limit:#@name" } end end def acquire Selector.acquire([@name], determine_namespace).size > 0 end def release redis {|it| it.lrem "#{PREFIX}:probed:#@name", 1, Selector.uuid } end def busy redis {|it| it.llen "#{PREFIX}:busy:#@name" } end def busy_processes redis {|it| it.lrange "#{PREFIX}:busy:#@name", 0, -1 } end def increase_busy increase_local_busy redis {|it| it.rpush "#{PREFIX}:busy:#@name", Selector.uuid } end def decrease_busy decrease_local_busy redis {|it| it.lrem "#{PREFIX}:busy:#@name", 1, Selector.uuid } end def probed redis {|it| it.llen "#{PREFIX}:probed:#@name" } end def probed_processes redis {|it| it.lrange "#{PREFIX}:probed:#@name", 0, -1 } end def pause redis {|it| it.set "#{PREFIX}:pause:#@name", true } end def pause_for_ms ms redis {|it| it.psetex "#{PREFIX}:pause:#@name", ms, true } end def unpause redis {|it| it.del "#{PREFIX}:pause:#@name" } end def paused? redis {|it| it.get "#{PREFIX}:pause:#@name" } end def block redis {|it| it.set "#{PREFIX}:block:#@name", true } end def block_except(*queues) raise ArgumentError if queues.empty? redis {|it| it.set "#{PREFIX}:block:#@name", queues.join(',') } end def unblock redis {|it| it.del "#{PREFIX}:block:#@name" } end def blocking? redis {|it| it.get "#{PREFIX}:block:#@name" } end def increase_local_busy @lock.synchronize { @local_busy += 1 } end def decrease_local_busy @lock.synchronize { @local_busy -= 1 } end def local_busy? @local_busy > 0 end def explain <<-END.gsub(/^ {8}/, '') Current sidekiq process: #{Selector.uuid} All processes: #{Monitor.all_processes.join "\n"} Stale processes: #{Monitor.old_processes.join "\n"} Locked queue processes: #{probed_processes.sort.join "\n"} Busy queue processes: #{busy_processes.sort.join "\n"} Limit: #{limit.inspect} Process limit: #{process_limit.inspect} Blocking: #{blocking?} END end def remove_locks_except!(processes) locked_processes = probed_processes.uniq (locked_processes - processes).each do |dead_process| remove_lock! dead_process end end def remove_lock!(process) redis do |it| it.lrem "#{PREFIX}:probed:#@name", 0, process it.lrem "#{PREFIX}:busy:#@name", 0, process end end end end