discourse/lib/scheduler/manager.rb

# Initially we used sidetiq, this was a problem:
#
# 1. No mechnism to add "randomisation" into job execution
# 2. No stats about previous runs or failures
# 3. Dependency on ice_cube gem causes runaway CPU

module Scheduler
  class Manager
    extend Sidekiq::ExceptionHandler
    attr_accessor :random_ratio, :redis


    class Runner
      def initialize(manager)
        @queue = Queue.new
        @manager = manager
        @thread = Thread.new do
          while true
            klass = @queue.deq
            failed = false
            start = Time.now.to_f
            info = @manager.schedule_info(klass)
            begin
              info.prev_result = "RUNNING"
              info.write!
              klass.new.perform
            rescue => e
              Scheduler::Manager.handle_exception(e)
              failed = true
            end
            duration = ((Time.now.to_f - start) * 1000).to_i
            info.prev_duration = duration
            info.prev_result = failed ? "FAILED" : "OK"
            info.write!
          end
        end
      end

      def stop!
        @thread.kill
      end

      def enq(klass)
        @queue << klass
      end

      def wait_till_done
        while !@queue.empty? && !(@queue.num_waiting > 0)
          sleep 0.001
        end
      end
    end

    def self.without_runner(redis=nil)
      self.new(redis, true)
    end

    def initialize(redis = nil, skip_runner = false)
      @redis = $redis || redis
      @random_ratio = 0.1
      unless skip_runner
        @runner = Runner.new(self)
        self.class.current = self
      end
      @manager_id = SecureRandom.hex
    end

    def self.current
      @current
    end

    def self.current=(manager)
      @current = manager
    end

    def schedule_info(klass)
      ScheduleInfo.new(klass, self)
    end

    def next_run(klass)
      schedule_info(klass).next_run
    end

    def ensure_schedule!(klass)
      lock do
        schedule_info(klass).schedule!
      end

    end

    def remove(klass)
      lock do
        schedule_info(klass).del!
      end
    end

    def tick
      lock do
        (key, due), _ = redis.zrange Manager.queue_key, 0, 0, withscores: true
        return unless key
        if due.to_i <= Time.now.to_i
          klass = begin
                    key.constantize
                  rescue NameError
                    nil
                  end
          return unless klass
          info = schedule_info(klass)
          info.prev_run = Time.now.to_i
          info.prev_result = "QUEUED"
          info.prev_duration = -1
          info.next_run = nil
          info.schedule!
          @runner.enq(klass)
        end
      end
    end

    def blocking_tick
      tick
      @runner.wait_till_done
    end

    def stop!
      @runner.stop!
      self.class.current = nil
    end


    def lock
      got_lock = false
      lock_key = Manager.lock_key

      while(!got_lock)
        begin
          if redis.setnx lock_key, Time.now.to_i + 60
            redis.expire lock_key, 60
            got_lock = true
          else
            begin
              redis.watch lock_key
              time = redis.get Manager.lock_key
              if time && time.to_i < Time.now.to_i
                got_lock = redis.multi do
                  redis.set Manager.lock_key, Time.now.to_i + 60
                end
              end
            ensure
              redis.unwatch
            end
          end

        end
      end
      yield
    ensure
      redis.del Manager.lock_key
    end

    def self.discover_schedules
      schedules = []
      ObjectSpace.each_object(Scheduler::Schedule) do |schedule|
        schedules << schedule if schedule.scheduled?
      end
      schedules
    end

    def self.lock_key
      "_scheduler_lock_"
    end

    def self.queue_key
      "_scheduler_queue_"
    end

    def self.schedule_key(klass)
      "_scheduler_#{klass}"
    end
  end
end