Throttling jobs via arbitrary resource - Issue #27

.gitignore

@@ -1,12 +1,15 @@
 *.gem
+*~
+.DS_STORE
 .bundle
 pkg/*
-.DS_STORE
 
 spec/redis.config.yml
 bundle
 spec/tmp
 coverage
 bin
 
+phantomjs/
+
 .rspec-local

.gitmodules

@@ -1,3 +1,3 @@
 [submodule "lib/qless/qless-core"]
 	path = lib/qless/qless-core
-	url = https://github.com/seomoz/qless-core.git
+	url = https://github.com/backupify/qless-core.git

Gemfile

@@ -8,3 +8,11 @@ group :extras do
 end
 
 gem 'thin' # needed by qless-web binary
+
+group :development do
+  gem 'byebug', :platforms => [:ruby_20, :ruby_21]
+  gem 'pry'
+  gem 'pry-byebug', :platforms => [:ruby_20, :ruby_21]
+  gem 'pry-stack_explorer'
+  gem 'cane', :platforms => [:ruby_20, :ruby_21]
+end

Gemfile.lock

@@ -11,6 +11,13 @@ GEM
     ast (1.1.0)
     atomic (1.1.14)
     avl_tree (1.1.3)
+    binding_of_caller (0.7.2)
+      debug_inspector (>= 0.0.1)
+    byebug (2.7.0)
+      columnize (~> 0.3)
+      debugger-linecache (~> 1.2)
+    cane (2.6.2)
+      parallel
     capybara (1.1.4)
       mime-types (>= 1.16)
       nokogiri (>= 1.3.3)
@@ -20,8 +27,10 @@ GEM
       xpath (~> 0.1.4)
     childprocess (0.3.9)
       ffi (~> 1.0, >= 1.0.11)
+    coderay (1.1.0)
     columnize (0.3.6)
     daemons (1.1.9)
+    debug_inspector (0.0.2)
     debugger (1.6.2)
       columnize (>= 0.3.1)
       debugger-linecache (~> 1.2.0)
@@ -40,6 +49,7 @@ GEM
     http_parser.rb (0.5.3)
     launchy (2.1.2)
       addressable (~> 2.3)
+    method_source (0.8.2)
     metriks (0.9.9.5)
       atomic (~> 1.0)
       avl_tree (~> 1.1.2)
@@ -50,6 +60,7 @@ GEM
     multipart-post (1.2.0)
     nokogiri (1.6.0)
       mini_portile (~> 0.5.0)
+    parallel (1.4.1)
     parser (2.0.0.pre8)
       ast (~> 1.1)
       slop (~> 3.4, >= 3.4.5)
@@ -60,6 +71,16 @@ GEM
       http_parser.rb (~> 0.5.3)
       multi_json (~> 1.0)
     powerpack (0.0.8)
+    pry (0.10.1)
+      coderay (~> 1.1.0)
+      method_source (~> 0.8.1)
+      slop (~> 3.4)
+    pry-byebug (1.3.3)
+      byebug (~> 2.7)
+      pry (~> 0.10)
+    pry-stack_explorer (0.4.9.2)
+      binding_of_caller (>= 0.7)
+      pry (>= 0.9.11)
     rack (1.5.2)
     rack-protection (1.5.0)
       rack
@@ -119,12 +140,17 @@ PLATFORMS
   ruby
 
 DEPENDENCIES
+  byebug
+  cane
   capybara (~> 1.1.2)
   debugger
   faye-websocket (~> 0.4.0)
   launchy (~> 2.1.0)
   metriks (~> 0.9)
   poltergeist (~> 1.0.0)
+  pry
+  pry-byebug
+  pry-stack_explorer
   qless!
   rake (~> 10.0)
   rspec (~> 2.12)

README.md

@@ -1,4 +1,4 @@
-qless
+qless [![Build Status](https://travis-ci.org/seomoz/qless.svg?branch=master)](https://travis-ci.org/seomoz/qless)
 =====
 
 Qless is a powerful `Redis`-based job queueing system inspired by
@@ -162,31 +162,41 @@ it is empty, before trying to pop job off the second queue. The
 round-robin reserver will pop a job off the first queue, then the second
 queue, and so on. You could also easily implement your own.
 
-To start a worker, load the qless rake tasks in your Rakefile, and
-define a `qless:setup` task:
+To start a worker, write a bit of Ruby code that instantiates a
+worker and runs it. You could write a rake task to do this, for
+example:
 
 ``` ruby
-require 'qless/tasks'
 namespace :qless do
-  task :setup do
-    require 'my_app/environment' # to ensure all job classes are loaded
-
-    # Set options via environment variables
-    # The only required option is QUEUES; the
-    # rest have reasonable defaults.
-    ENV['REDIS_URL'] ||= 'redis://some-host:7000/3'
-    ENV['QUEUES'] ||= 'fizz,buzz'
-    ENV['JOB_RESERVER'] ||= 'Ordered'
-    ENV['INTERVAL'] ||= '10' # 10 seconds
-    ENV['VERBOSE'] ||= 'true'
-  end
-end
-```
+  desc "Run a Qless worker"
+  task :work do
+    # Load your application code. All job classes must be loaded.
+    require 'my_app/environment'
 
-Then run the `qless:work` rake task:
+    # Require the parts of qless you need
+    require 'qless'
+    require 'qless/job_reservers/ordered'
+    require 'qless/worker'
 
-```
-rake qless:work
+    # Create a client
+    client = Qless::Client.new(:host => 'foo.bar.com', :port => 1234)
+
+    # Get the queues you use
+    queues = %w[ queue_1 queue_2 ].map do |name|
+      client.queues[name]
+    end
+
+    # Create a job reserver; different reservers use different
+    # strategies for which order jobs are popped off of queues
+    reserver = Qless::JobReservers::Ordered.new(queues)
+
+    # Create a forking worker that uses the given reserver to pop jobs.
+    worker = Qless::Workers::ForkingWorker.new(reserver)
+
+    # Start the worker!
+    worker.run
+  end
+end
 ```
 
 The following signals are supported in the parent process:

Rakefile

@@ -21,7 +21,7 @@ task :check_coverage do
   end
 end
 
-task default: [:spec, :check_coverage]
+task default: [:spec, :check_coverage, :cane]
 
 namespace :core do
   qless_core_dir = "./lib/qless/qless-core"
@@ -82,15 +82,24 @@ namespace :core do
   task verify: %w[ verify:clean verify:current ]
 end
 
+desc "Starts a qless console"
+task :console do
+  ENV['PUBLIC_SEQUEL_API'] = 'true'
+  ENV['NO_NEW_RELIC'] = 'true'
+  exec "bundle exec pry -r./conf/console"
+end
+
 require 'qless/tasks'
 
 namespace :qless do
-  task :setup do
+  desc "Runs a test worker so you can send signals to it for testing"
+  task :run_test_worker do
     require 'qless'
+    require 'qless/job_reservers/ordered'
+    require 'qless/worker'
     queue = Qless::Client.new.queues["example"]
     queue.client.redis.flushdb
 
-    ENV['QUEUES'] = queue.name
     ENV['VVERBOSE'] = '1'
 
     class ExampleJob
@@ -105,6 +114,55 @@ namespace :qless do
     20.times do |i|
       queue.put(ExampleJob, sleep: i)
     end
+
+    reserver = Qless::JobReservers::Ordered.new([queue])
+    Qless::Workers::ForkingWorker.new(reserver, log_level: Logger::INFO).run
+  end
+end
+
+
+namespace :cane do
+  begin
+    require 'cane/rake_task'
+
+    libs = [
+      { name: 'qless', dir: '.', root: '.' },
+    ]
+
+    libs.each do |lib|
+      desc "Runs cane code quality checks for #{lib[:name]}"
+      Cane::RakeTask.new(lib[:name]) do |cane|
+        cane.no_doc   = true
+
+        cane.abc_glob = "#{lib[:dir]}/{lib,spec}/**/*.rb"
+        cane.abc_max = 15
+        cane.abc_exclude = %w[
+          Middleware::(anon)#expect_job_to_timeout
+          Qless::Job#initialize
+          Qless::Middleware::RequeueExceptions#handle_exception
+          Qless::Middleware::Timeout#initialize
+          Qless::WorkerHelpers#run_jobs
+          Qless::Workers::BaseWorker#initialize
+          Qless::Workers::BaseWorker#register_signal_handlers
+          Qless::Workers::ForkingWorker#register_signal_handlers
+          Qless::Workers::SerialWorker#run
+        ]
+
+        cane.style_glob = "#{lib[:dir]}/lib/**/*.rb"
+        cane.style_measure = 100
+        cane.style_exclude = %w[
+        ]
+      end
+    end
+
+    desc "Runs cane code quality checks for all projects"
+    task all: libs.map { |l| l[:name] }
+
+  rescue LoadError
+    task :all do
+      puts "cane is not supported in ruby #{RUBY_VERSION}"
+    end
   end
 end
 
+task cane: "cane:all"

conf/console.rb

@@ -0,0 +1,20 @@
+$LOAD_PATH.unshift File.expand_path('../../lib', __FILE__)
+
+require 'irb/completion'
+
+QLESS_CONSOLE = true
+
+require 'qless'
+
+module StdoutLogger
+  def logger
+    @logger ||= Logger.new($stdout)
+  end
+end
+
+# Load everything!
+Dir["./lib/**/*.rb"].sort.each do |f|
+  require f.gsub("./lib/", "")
+end
+
+require 'pp'

exe/install_phantomjs

@@ -10,7 +10,7 @@ then
   brew install phantomjs
 elif [[ "$os_name" == 'Linux' ]]
 then
-  version=phantomjs-1.7.0-linux-i686
+  version=phantomjs-1.7.0-linux-x86_64
   wget http://phantomjs.googlecode.com/files/$version.tar.bz2
   tar xjf $version.tar.bz2
   mv $version phantomjs

exe/qless-growl

@@ -7,7 +7,7 @@ require 'ruby-growl'
 require 'micro-optparse'
 
 @options = Parser.new do |p|
-  p.banner = 'This agent lets you get campfire notifications for the progress of tracked jobs'
+  p.banner = 'This agent lets you get growl notifications for the progress of tracked jobs'
   p.option :growl    , 'host for the growl daemon', :default => 'localhost'
   p.option :app      , 'application name for notifications', :default => 'qless'
   p.option :host     , 'host:port for your qless redis instance', :default => 'localhost:6379'

lib/qless.rb

@@ -23,6 +23,7 @@ module Qless
 require 'qless/version'
 require 'qless/config'
 require 'qless/queue'
+require 'qless/throttle'
 require 'qless/job'
 require 'qless/lua_script'
 require 'qless/failure_formatter'
@@ -135,6 +136,24 @@ def [](name)
     end
   end
 
+  # A class for interacting with throttles. Not meant to be instantiated directly,
+  # it's accessed through Client#throttles
+  class ClientThrottles
+    def initialize(client)
+      @client = client
+    end
+
+    def [](name)
+      Throttle.new(name, @client)
+    end
+
+    def counts
+      @client.queues.counts.map do |queue|
+        Queue.new(queue['name'], @client).throttle
+      end
+    end
+  end
+
   # A class for interacting with events. Not meant to be instantiated directly,
   # it's accessed through Client#events
   class ClientEvents
@@ -169,20 +188,24 @@ def stop
   # The client for interacting with Qless
   class Client
     # Lua script
-    attr_reader :_qless, :config, :redis, :jobs, :queues, :workers
+    attr_reader :_qless, :config, :redis, :jobs, :queues, :throttles, :workers
     attr_accessor :worker_name
 
     def initialize(options = {})
+      default_options = {:ensure_minimum_version => true}
+      options = default_options.merge(options)
+
       # This is the redis instance we're connected to. Use connect so REDIS_URL
       # will be honored
       @redis   = options[:redis] || Redis.connect(options)
       @options = options
-      assert_minimum_redis_version('2.5.5')
+      assert_minimum_redis_version('2.5.5') if @options.delete(:ensure_minimum_version)
       @config = Config.new(self)
       @_qless = Qless::LuaScript.new('qless', @redis)
 
       @jobs    = ClientJobs.new(self)
       @queues  = ClientQueues.new(self)
+      @throttles  = ClientThrottles.new(self)
       @workers = ClientWorkers.new(self)
       @worker_name = [Socket.gethostname, Process.pid.to_s].join('-')
     end