Class: Sidekiq::ProcessSet
- Inherits:
-
Object
- Object
- Sidekiq::ProcessSet
- Includes:
- Enumerable
- Defined in:
- lib/sidekiq/api.rb
Overview
Enumerates the set of Sidekiq processes which are actively working right now. Each process sends a heartbeat to Redis every 5 seconds so this set should be relatively accurate, barring network partitions.
Yields a Sidekiq::Process.
Instance Method Summary collapse
-
#cleanup ⇒ Object
Cleans up dead processes recorded in Redis.
- #each ⇒ Object
-
#initialize(clean_plz = true) ⇒ ProcessSet
constructor
A new instance of ProcessSet.
-
#leader ⇒ Object
Returns the identity of the current cluster leader or “” if no leader.
-
#size ⇒ Object
This method is not guaranteed accurate since it does not prune the set based on current heartbeat.
Constructor Details
#initialize(clean_plz = true) ⇒ ProcessSet
Returns a new instance of ProcessSet.
760 761 762 |
# File 'lib/sidekiq/api.rb', line 760 def initialize(clean_plz = true) cleanup if clean_plz end |
Instance Method Details
#cleanup ⇒ Object
Cleans up dead processes recorded in Redis. Returns the number of processes cleaned.
766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 |
# File 'lib/sidekiq/api.rb', line 766 def cleanup count = 0 Sidekiq.redis do |conn| procs = conn.sscan_each("processes").to_a.sort heartbeats = conn.pipelined { procs.each do |key| conn.hget(key, "info") end } # the hash named key has an expiry of 60 seconds. # if it's not found, that means the process has not reported # in to Redis and probably died. to_prune = [] heartbeats.each_with_index do |beat, i| to_prune << procs[i] if beat.nil? end count = conn.srem("processes", to_prune) unless to_prune.empty? end count end |
#each ⇒ Object
788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 |
# File 'lib/sidekiq/api.rb', line 788 def each procs = Sidekiq.redis { |conn| conn.sscan_each("processes").to_a }.sort Sidekiq.redis do |conn| # We're making a tradeoff here between consuming more memory instead of # making more roundtrips to Redis, but if you have hundreds or thousands of workers, # you'll be happier this way result = conn.pipelined { procs.each do |key| conn.hmget(key, "info", "busy", "beat", "quiet") end } result.each do |info, busy, at_s, quiet| # If a process is stopped between when we query Redis for `procs` and # when we query for `result`, we will have an item in `result` that is # composed of `nil` values. next if info.nil? hash = Sidekiq.load_json(info) yield Process.new(hash.merge("busy" => busy.to_i, "beat" => at_s.to_f, "quiet" => quiet)) end end nil end |
#leader ⇒ Object
Returns the identity of the current cluster leader or “” if no leader. This is a Sidekiq Enterprise feature, will always return “” in Sidekiq or Sidekiq Pro.
826 827 828 829 830 831 832 833 |
# File 'lib/sidekiq/api.rb', line 826 def leader @leader ||= begin x = Sidekiq.redis { |c| c.get("dear-leader") } # need a non-falsy value so we can memoize x ||= "" x end end |
#size ⇒ Object
This method is not guaranteed accurate since it does not prune the set based on current heartbeat. #each does that and ensures the set only contains Sidekiq processes which have sent a heartbeat within the last 60 seconds.
819 820 821 |
# File 'lib/sidekiq/api.rb', line 819 def size Sidekiq.redis { |conn| conn.scard("processes") } end |