initial commit of threading worker

README.md

@@ -401,6 +401,7 @@ By default, Backburner comes with the following workers built-in:
 | ------- | -------------------------------                                            |
 | `Backburner::Workers::Simple` | Single threaded, no forking worker. Simplest option. |
 | `Backburner::Workers::Forking` | Basic forking worker that manages crashes and memory bloat. |
+| `Backburner::Workers::Threading` | Basic worker that can start a given number of threads per queue |
 | `Backburner::Workers::ThreadsOnFork` | Forking worker that utilizes threads for concurrent processing. |
 
 You can select the default worker for processing with:
@@ -426,6 +427,11 @@ $ QUEUE=newsletter-sender,push-message THREADS=2 GARBAGE=1000 rake backburner:th
 For more information on the threads_on_fork worker, check out the
 [ThreadsOnFork Worker](https://github.com/nesquena/backburner/wiki/ThreadsOnFork-worker) documentation. Please note that the `ThreadsOnFork` worker does not work on Windows due to its lack of `fork`.
 
+```
+$ QUEUE=newsletter-sender:2,push-message rake backburner:threading:work
+```
+
+Will create two threads listening to newsletter-sender queue and one thread listening to push-message queue.
 
 Additional workers such as individual `threaded` and `forking` strategies will hopefully be contributed in the future.
 If you are interested in helping out, please let us know.

lib/backburner.rb

@@ -13,6 +13,7 @@
 require 'backburner/workers/simple'
 require 'backburner/workers/forking'
 require 'backburner/workers/threads_on_fork'
+require 'backburner/workers/threading'
 require 'backburner/queue'
 
 module Backburner

lib/backburner/tasks.rb

@@ -27,6 +27,16 @@
     end
   end # forking
 
+  namespace :threading do
+    # QUEUE=twitter:10,parse_page rake backburner:threads_on_fork:work
+    # twitter tube will have 10 threads, parse_page only 1 thread
+    desc "Starts backburner worker using threading processing"
+    task :work => :environment do
+      queues = (ENV["QUEUE"] ? ENV["QUEUE"].split(',') : nil) rescue nil
+      Backburner.work queues, :worker => Backburner::Workers::Threading
+    end
+  end # threading
+
   namespace :threads_on_fork do
     # QUEUE=twitter:10:5000:5,parse_page,send_mail,verify_bithday THREADS=2 GARBAGE=1000 rake backburner:threads_on_fork:work
     # twitter tube will have 10 threads, garbage after 5k executions and retry 5 times.
@@ -40,4 +50,4 @@
       Backburner.work queues, :worker => Backburner::Workers::ThreadsOnFork
     end
   end # threads_on_fork
-end
+end

lib/backburner/workers/threading.rb

@@ -0,0 +1,61 @@
+module Backburner
+  module Workers
+    class Threading < Worker
+      # Used to prepare job queues before processing jobs.
+      # Setup beanstalk tube_names and watch all specified tubes for jobs.
+      #
+      # @raise [Beaneater::NotConnected] If beanstalk fails to connect.
+      # @example
+      #   @worker.prepare
+      #
+      def prepare
+        @queue_options = {}
+        self.tube_names.map! { |name|
+          # queue can be in format queue:number_of_threads
+          # number_of_threads is 1 by default
+          data = name.split(":")
+          tube_name = expand_tube_name(data.first)
+          @queue_options[tube_name] = {}
+          @queue_options[tube_name][:number_of_threads] = data[1].nil? || data[1].empty? ? 1 : data[1].to_i
+          tube_name
+        }.uniq!
+        log_info "[Threading worker] Working #{tube_names.size} queues: [ #{tube_names.join(', ')} ]"
+      end
+
+      # Starts processing new jobs indefinitely.
+      # Primary way to consume and process jobs in specified tubes.
+      #
+      # @example
+      #   @worker.start
+      #
+      def start
+        prepare
+
+        threads = []
+        self.tube_names.each do |tube_name|
+          @queue_options[tube_name][:number_of_threads].times do
+            threads << Thread.new do
+              log_info "[Threading worker] Watching queue : #{tube_name}"
+              Thread.current[:tube_name] = tube_name
+              Thread.current[:conn] = new_connection()
+              Thread.current[:conn].tubes.watch!(Thread.current[:tube_name])
+              loop {
+                work_one_job(Thread.current[:conn])
+              }
+            end
+          end
+        end
+
+        threads.each { |thread| thread.join }
+      end
+
+      def on_reconnect(conn)
+        unless Thread.current[:tube_name].nil? then
+          log_info "[Threading worker] Trying to recover connection for queue #{Thread.current[:tube_name]}"
+          Thread.current[:conn] = conn
+          Thread.current[:conn].tubes.watch!(Thread.current[:tube_name])
+        end
+      end
+    end
+  end
+end

test/workers/threading_worker_test.rb

@@ -0,0 +1,18 @@
+require File.expand_path('../../test_helper', __FILE__)
+require File.expand_path('../../fixtures/test_forking_jobs', __FILE__)
+
+describe "Backburner::Workers::Threading module" do
+
+  before do
+    Backburner.default_queues.clear
+    @worker_class = Backburner::Workers::Threading
+  end
+
+  describe "for prepare method" do
+    it "should make tube names array always unique to avoid duplication" do
+      worker = @worker_class.new(["foo", "demo.test.foo"])
+      worker.prepare
+      assert_equal ["demo.test.foo"], worker.tube_names
+    end
+  end # prepare
+end