Class: Sidekiq::ProcessSet

Inherits:
Object
  • Object
show all
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.

Yield Parameters:

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.[](identity) ⇒ Object 



854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
 
# File 'lib/sidekiq/api.rb', line 854

def self.[](identity)
  exists, (info, busy, beat, quiet, rss, rtt_us) = Sidekiq.redis { |conn|
    conn.multi { |transaction|
      transaction.sismember("processes", identity)
      transaction.hmget(identity, "info", "busy", "beat", "quiet", "rss", "rtt_us")
    }
  }

  return nil if exists == 0 || info.nil?

  hash = Sidekiq.load_json(info)
  Process.new(hash.merge("busy" => busy.to_i,
    "beat" => beat.to_f,
    "quiet" => quiet,
    "rss" => rss.to_i,
    "rtt_us" => rtt_us.to_i))
end

Instance Method Details

#each ⇒ Object 



906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
 
# File 'lib/sidekiq/api.rb', line 906

def each
  result = Sidekiq.redis { |conn|
    procs = conn.sscan("processes").to_a.sort

    # 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
    conn.pipelined do |pipeline|
      procs.each do |key|
        pipeline.hmget(key, "info", "busy", "beat", "quiet", "rss", "rtt_us")
      end
    end
  }

  result.each do |info, busy, beat, quiet, rss, rtt_us|
    # 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" => beat.to_f,
      "quiet" => quiet,
      "rss" => rss.to_i,
      "rtt_us" => rtt_us.to_i))
  end
end

#leader ⇒ String

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.

Returns:

  • (String) —

    Identity of cluster leader

  • (String) —

    empty string if no leader



963
964
965
966
967
968
969
970
 
# File 'lib/sidekiq/api.rb', line 963

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 ⇒ Integer

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.

Returns:

  • (Integer) —

    current number of registered Sidekiq processes



940
941
942
 
# File 'lib/sidekiq/api.rb', line 940

def size
  Sidekiq.redis { |conn| conn.scard("processes") }
end

#total_concurrency ⇒ Integer

Total number of threads available to execute jobs. For Sidekiq Enterprise customers this number (in production) must be less than or equal to your licensed concurrency.

Returns:

  • (Integer) —

    the sum of process concurrency



948
949
950
 
# File 'lib/sidekiq/api.rb', line 948

def total_concurrency
  sum { |x| x["concurrency"].to_i }
end

#total_rss_in_kb ⇒ Integer Also known as: total_rss

Returns total amount of RSS memory consumed by Sidekiq processes.

Returns:

  • (Integer) —

    total amount of RSS memory consumed by Sidekiq processes



953
954
955
 
# File 'lib/sidekiq/api.rb', line 953

def total_rss_in_kb
  sum { |x| x["rss"].to_i }
end