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 



880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
 
# File 'lib/sidekiq/api.rb', line 880

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 



932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
 
# File 'lib/sidekiq/api.rb', line 932

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



989
990
991
992
993
994
995
996
 
# File 'lib/sidekiq/api.rb', line 989

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



966
967
968
 
# File 'lib/sidekiq/api.rb', line 966

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



974
975
976
 
# File 'lib/sidekiq/api.rb', line 974

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



979
980
981
 
# File 'lib/sidekiq/api.rb', line 979

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