Remove sidekiq/api dependency from scheduled, run cleanup on scheduler start, #5513

Author: mperham

lib/sidekiq/api.rb

@@ -11,6 +11,17 @@
   require "sidekiq/metrics/query"
 end
 
+#
+# Sidekiq's Data API provides a Ruby object model on top
+# of Sidekiq's runtime data in Redis. This API should never
+# be used within application code for business logic.
+#
+# The Sidekiq server process never uses this API: all data
+# manipulation is done directly for performance reasons to
+# ensure we are using Redis as efficiently as possible at
+# every callsite.
+#
+
 module Sidekiq
   # Retrieve runtime statistics from Redis regarding
   # this Sidekiq cluster.
@@ -893,10 +904,12 @@ def initialize(clean_plz = true)
     # :nodoc:
     # @api private
     def cleanup
+      # dont run cleanup more than once per minute
       return 0 unless Sidekiq.redis { |conn| conn.set("process_cleanup", "1", nx: true, ex: 60) }
+
       count = 0
       Sidekiq.redis do |conn|
-        procs = conn.sscan_each("processes").to_a.sort
+        procs = conn.sscan_each("processes").to_a
         heartbeats = conn.pipelined { |pipeline|
           procs.each do |key|
             pipeline.hget(key, "info")

lib/sidekiq/scheduled.rb

@@ -190,22 +190,47 @@ def process_count
         pcount
       end
 
-      def initial_wait
-        # periodically clean out the `processes` set in Redis which can collect
-        # references to dead processes over time. The process count affects how
-        # often we scan for scheduled jobs.
-        ps = Sidekiq::ProcessSet.new(false)
-        ps.cleanup if rand(1000) % 10 == 0 # only cleanup 10% of the time
+      # A copy of Sidekiq::ProcessSet#cleanup because server
+      # should never depend on sidekiq/api.
+      def cleanup
+        # dont run cleanup more than once per minute
+        return 0 unless Sidekiq.redis { |conn| conn.set("process_cleanup", "1", nx: true, ex: 60) }
 
-        # Have all processes sleep between 5-15 seconds.  10 seconds
-        # to give time for the heartbeat to register (if the poll interval is going to be calculated by the number
+        count = 0
+        Sidekiq.redis do |conn|
+          procs = conn.sscan_each("processes").to_a
+          heartbeats = conn.pipelined { |pipeline|
+            procs.each do |key|
+              pipeline.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 = procs.select.with_index { |proc, i|
+            heartbeats[i].nil?
+          }
+          count = conn.srem("processes", to_prune) unless to_prune.empty?
+        end
+        count
+      end
+
+      def initial_wait
+        # Have all processes sleep between 5-15 seconds. 10 seconds to give time for
+        # the heartbeat to register (if the poll interval is going to be calculated by the number
         # of workers), and 5 random seconds to ensure they don't all hit Redis at the same time.
         total = 0
         total += INITIAL_WAIT unless @config[:poll_interval_average]
         total += (5 * rand)
 
         @sleeper.pop(total)
       rescue Timeout::Error
+      ensure
+        # periodically clean out the `processes` set in Redis which can collect
+        # references to dead processes over time. The process count affects how
+        # often we scan for scheduled jobs.
+        cleanup
       end
     end
   end