Class: Sidekiq::ProcessSet
- Inherits:
-
Object
- Object
- Sidekiq::ProcessSet
- Includes:
- Enumerable, RedisScanner
- Defined in:
- lib/sidekiq/api.rb
Overview
Enumerates the set of Sidekiq processes which are actively working right now. Each process send 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.
Methods included from RedisScanner
Constructor Details
#initialize(clean_plz = true) ⇒ ProcessSet
Returns a new instance of ProcessSet.
736 737 738 |
# File 'lib/sidekiq/api.rb', line 736 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.
742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 |
# File 'lib/sidekiq/api.rb', line 742 def cleanup count = 0 Sidekiq.redis do |conn| procs = sscan(conn, 'processes').sort heartbeats = conn.pipelined do procs.each do |key| conn.hget(key, 'info') end 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
764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 |
# File 'lib/sidekiq/api.rb', line 764 def each procs = Sidekiq.redis { |conn| sscan(conn, 'processes') }.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 do procs.each do |key| conn.hmget(key, 'info', 'busy', 'beat', 'quiet') end 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.
802 803 804 805 806 807 808 809 |
# File 'lib/sidekiq/api.rb', line 802 def leader @leader ||= begin x = Sidekiq.redis {|c| c.get("dear-leader") } # need a non-falsy value so we can memoize x = "" unless 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.
795 796 797 |
# File 'lib/sidekiq/api.rb', line 795 def size Sidekiq.redis { |conn| conn.scard('processes') } end |