module Resque module DynamicQueues def filter_busy_queues qs busy_queues = Resque::Worker.working.map { |worker| worker.job["queue"] }.compact Array(qs.dup).compact - busy_queues end def rotated_queues @n ||= 0 @n += 1 rot_queues = queues # since we rely on the resque-dynamic-queues plugin, this is all the queues, expanded out if rot_queues.size > 0 @n = @n % rot_queues.size rot_queues.rotate(@n) else rot_queues end end def queue_depth queuename busy_queues = Resque::Worker.working.map { |worker| worker.job["queue"] }.compact # find the queuename, count it. busy_queues.select {|q| q == queuename }.size end def get_categorized_queues(queue_list) priority_map = {"Synchronous" => 0, "High" => 1, "Medium" => 2, "Low" => 3} categorized_queues = {} for queue in queue_list.uniq priority = queue.split("_")[1] priority = "Medium" if !["Synchronous", "High", "Medium", "Low"].include?(priority) categorized_queues[priority] ||= [] categorized_queues[priority].push(queue) end return categorized_queues.transform_keys{ |key| priority_map[key.to_s]}.sort end DEFAULT_QUEUE_DEPTH = 0 def should_work_on_queue? queuename return true if @queues.include? '*' # workers with QUEUES=* are special and are not subject to queue depth setting max = DEFAULT_QUEUE_DEPTH unless ENV["RESQUE_QUEUE_DEPTH"].nil? || ENV["RESQUE_QUEUE_DEPTH"] == "" max = ENV["RESQUE_QUEUE_DEPTH"].to_i end return true if max == 0 # 0 means no limiting cur_depth = queue_depth(queuename) log! "queue #{queuename} depth = #{cur_depth} max = #{max}" return true if cur_depth < max false end def get_grouped_queues self.queues.sort.group_by{|u| /(\d{1,20})_.*/.match(u) ? /(\d{1,20})_.*/.match(u).captures.first : nil} end def reserve_with_round_robin grouped_queues = self.get_grouped_queues #Instance queue grouping if !grouped_queues.keys.include?(nil) && grouped_queues.keys.size > 0 @n ||= 0 @n += 1 @n = @n % grouped_queues.keys.size grouped_queues.keys.rotate(@n).each do |key| self.get_categorized_queues(grouped_queues[key]).each do |key, queues| queues.each do |queue| log! "Checking #{queue}" if should_work_on_queue?(queue) && @job_in_progress = Resque::Job.reserve(queue) log! "Found job on #{queue}" return @job_in_progress end end end @n += 1 # Start the next search at the queue after the one from which we pick a job. end nil else return reserve_without_round_robin end rescue Exception => e log "Error reserving job: #{e.inspect}" log e.backtrace.join("\n") raise e end # Returns a list of queues to use when searching for a job. # # A splat ("*") means you want every queue (in alpha order) - this # can be useful for dynamically adding new queues. # # The splat can also be used as a wildcard within a queue name, # e.g. "*high*", and negation can be indicated with a prefix of "!" # # An @key can be used to dynamically look up the queue list for key from redis. # If no key is supplied, it defaults to the worker's hostname, and wildcards # and negations can be used inside this dynamic queue list. Set the queue # list for a key with Resque.set_dynamic_queue(key, ["q1", "q2"] # def queues_with_dynamic queue_names = @queues.dup return queues_without_dynamic if queue_names.grep(/(^!)|(^@)|(\*)/).size == 0 real_queues = Resque.queues matched_queues = [] #Remove Queues under Api Limits Redis.current.zremrangebyscore("APILimits", "0", "(#{Time.now.to_i}") api_limit_instances = Redis.current.zrange("APILimits", 0, -1).map {|key| key.to_i} real_queues = real_queues.select {|key| key if !api_limit_instances.include?((key.match(/^(\d*)_.*/) || [])[1].to_i)} ## 2 #Queue Pausing Resque.redis.zremrangebyscore("PauseQueue", "0", "(#{Time.now.to_i}") paused_instances = Resque.redis.zrange("PauseQueue", 0, -1).map {|key| key.split("__")[0].to_i} real_queues = real_queues.select {|key| key if !paused_instances.include?((key.match(/^(\d*)_.*/) || [])[1].to_i)} while q = queue_names.shift q = q.to_s if q =~ /^(!)?@(.*)/ key = $2.strip key = hostname if key.size == 0 add_queues = Resque.get_dynamic_queue(key) add_queues.map! { |q| q.gsub!(/^!/, '') || q.gsub!(/^/, '!') } if $1 queue_names.concat(add_queues) next end if q =~ /^!/ negated = true q = q[1..-1] end patstr = q.gsub(/\*/, '.*') pattern = /^#{patstr}$/ if negated matched_queues -= matched_queues.grep(pattern) else matches = real_queues.grep(/^#{pattern}$/) matches = [q] if matches.size == 0 && q == patstr matched_queues.concat(matches.sort) end end return matched_queues.uniq end def self.included(receiver) receiver.class_eval do alias queues_without_dynamic queues alias queues queues_with_dynamic alias reserve_without_round_robin reserve alias reserve reserve_with_round_robin end end end end