WIP: native threads

lib/shoryuken/manager.rb

@@ -2,35 +2,22 @@ module Shoryuken
   class Manager
     include Util
 
-    BATCH_LIMIT = 10
-
     def initialize(fetcher, polling_strategy)
       @count = Shoryuken.options[:concurrency] || 25
       raise(ArgumentError, "Concurrency value #{@count} is invalid, it needs to be a positive number") unless @count > 0
 
-      @queues = Shoryuken.queues.dup.uniq
-
-      @done = Concurrent::AtomicBoolean.new(false)
-
       @fetcher = fetcher
       @polling_strategy = polling_strategy
 
-      @heartbeat = Concurrent::TimerTask.new(run_now: true, execution_interval: 1, timeout_interval: 60) { dispatch }
-
-      @ready = Concurrent::AtomicFixnum.new(@count)
-
-      @pool = Concurrent::FixedThreadPool.new(@count)
+      create_processors
     end
 
     def start
       logger.info { 'Starting' }
-
-      @heartbeat.execute
+      @processors.each(&:start)
     end
 
     def stop(options = {})
-      @done.make_true
-
       if (callback = Shoryuken.stop_callback)
         logger.info { 'Calling Shoryuken.on_stop block' }
         callback.call
@@ -40,94 +27,41 @@ def stop(options = {})
 
       logger.info { "Shutting down workers" }
 
-      @heartbeat.kill
-
       if options[:shutdown]
         hard_shutdown_in(options[:timeout])
       else
         soft_shutdown
       end
     end
 
-    def processor_done(queue)
-      logger.debug { "Process done for '#{queue}'" }
-
-      @ready.increment
-    end
-
-    private
-
-    def dispatch
-      return if @done.true?
-
-      logger.debug { "Ready: #{@ready.value}, Busy: #{busy}, Active Queues: #{@polling_strategy.active_queues}" }
-
-      if @ready.value == 0
-        return logger.debug { 'Pausing fetcher, because all processors are busy' }
-      end
-
-      unless queue = @polling_strategy.next_queue
-        return logger.debug { 'Pausing fetcher, because all queues are paused' }
+    def create_processors
+      @processors = (1..@count).map do
+        # fix this initialisation. we should creating new copies instead of calling .dup
+        Processor.new(@fetcher.dup, @polling_strategy.dup)
       end
-
-      batched_queue?(queue) ? dispatch_batch(queue) : dispatch_single_messages(queue)
     end
 
-    def busy
-      @count - @ready.value
-    end
-
-    def assign(queue, sqs_msg)
-      logger.debug { "Assigning #{sqs_msg.message_id}" }
-
-      @ready.decrement
-
-      @pool.post { Processor.new(self).process(queue, sqs_msg) }
-    end
-
-    def dispatch_batch(queue)
-      batch = @fetcher.fetch(queue, BATCH_LIMIT)
-      @polling_strategy.messages_found(queue.name, batch.size)
-      assign(queue.name, patch_batch!(batch))
-    end
-
-    def dispatch_single_messages(queue)
-      messages = @fetcher.fetch(queue, @ready.value)
-      @polling_strategy.messages_found(queue.name, messages.size)
-      messages.each { |message| assign(queue.name, message) }
-    end
-
-    def batched_queue?(queue)
-      Shoryuken.worker_registry.batch_receive_messages?(queue.name)
-    end
+    private
 
     def soft_shutdown
-      @pool.shutdown
-      @pool.wait_for_termination
+      @processors.each(&:terminate)
     end
 
     def hard_shutdown_in(delay)
-      if busy > 0
-        logger.info { "Pausing up to #{delay} seconds to allow workers to finish..." }
-      end
+      logger.info { "Pausing up to #{delay} seconds to allow workers to finish..." }
 
-      @pool.shutdown
+      soft_shutdown
 
-      unless @pool.wait_for_termination(delay)
-        logger.info { "Hard shutting down #{busy} busy workers" }
-
-        @pool.kill
+      wait_until = Time.now + delay
+      while Time.now < wait_until && has_alive_processors?
+        sleep(0.1)
       end
-    end
 
-    def patch_batch!(sqs_msgs)
-      sqs_msgs.instance_eval do
-        def message_id
-          "batch-with-#{size}-messages"
-        end
-      end
+      @processors.each(&:kill)
+    end
 
-      sqs_msgs
+    def has_alive_processors?
+      @processors.any?(&:alive?)
     end
   end
 end

lib/shoryuken/processor.rb

@@ -2,8 +2,77 @@ module Shoryuken
   class Processor
     include Util
 
-    def initialize(manager)
-      @manager = manager
+    BATCH_LIMIT = 10
+
+    def initialize(fetcher, polling_strategy)
+      @fetcher = fetcher
+      @polling_strategy = polling_strategy
+      # what is the worst thing that could happen because it's not atomic?
+      # is it possible that it evaluates to true even if we never set it to true?
+      @done = false
+    end
+
+    def start
+      # does updating and evaluating @thread has to be wrapped in mutex?
+      # can it throw exception?
+      # what should we do if it dies?
+      @thread ||= spawn_thread(&method(:loop))
+    end
+
+    def terminate(wait=false)
+      @done = true
+      # does updating and evaluating @thread has to be wrapped in mutex?
+      return if !@thread
+      @thread.value if wait
+    end
+
+    def kill
+      @done = true
+      # does updating and evaluating @thread has to be wrapped in mutex?
+      return if !@thread
+      @thread.raise ::Shoryuken::Shutdown
+    end
+
+    def alive?
+      # is this ok?
+      [email protected]
+    end
+
+    private
+
+    def loop
+      while !@done
+        do_work
+      end
+    rescue ::Shoryuken::Shutdown
+      # killed, bail!
+    rescue => e
+      # are we safe to retry here? What are kind of errors we do not want to recover from?
+      # OOM?
+      logger.debug { "Processor caught exception: #{e}" }
+      retry
+    end
+
+    def do_work
+      queue = @polling_strategy.next_queue
+      return logger.info { 'Doing nothing, all queues are paused' } if queue.nil?
+      batched_queue?(queue) ? dispatch_batch(queue) : dispatch_single_message(queue)
+    end
+
+    def dispatch_batch(queue)
+      batch = @fetcher.fetch(queue, BATCH_LIMIT)
+      @polling_strategy.messages_found(queue.name, batch.size)
+      process(queue.name, patch_batch!(batch))
+    end
+
+    def dispatch_single_message(queue)
+      messages = @fetcher.fetch(queue, 1)
+      @polling_strategy.messages_found(queue.name, messages.size)
+      process(queue.name, messages.first) if messages.length > 0
+    end
+
+    def batched_queue?(queue)
+      Shoryuken.worker_registry.batch_receive_messages?(queue.name)
     end
 
     def process(queue, sqs_msg)
@@ -13,11 +82,13 @@ def process(queue, sqs_msg)
       worker.class.server_middleware.invoke(worker, queue, sqs_msg, body) do
         worker.perform(sqs_msg, body)
       end
-
-      @manager.processor_done(queue)
     end
-
-    private
+
+    def spawn_thread(&block)
+      Thread.new do 
+        block.call
+      end
+    end
 
     def get_body(worker_class, sqs_msg)
       if sqs_msg.is_a? Array