RIP Celluloid, Hello concurrent-ruby/Thread pool

.gitignore

@@ -3,7 +3,7 @@
 .bundle
 .config
 .yardoc
-Gemfile.lock
+# Gemfile.lock
 InstalledFiles
 _yardoc
 coverage

Gemfile.lock

@@ -0,0 +1,76 @@
+PATH
+  remote: .
+  specs:
+    shoryuken (2.1.1)
+      aws-sdk-core (~> 2)
+      concurrent-ruby
+
+GEM
+  remote: https://rubygems.org/
+  specs:
+    aws-sdk-core (2.6.32)
+      aws-sigv4 (~> 1.0)
+      jmespath (~> 1.0)
+    aws-sigv4 (1.0.0)
+    byebug (9.0.5)
+    codeclimate-test-reporter (1.0.3)
+      simplecov
+    coderay (1.1.1)
+    concurrent-ruby (1.0.2)
+    diff-lcs (1.2.5)
+    docile (1.1.5)
+    dotenv (2.1.1)
+    jmespath (1.3.1)
+    json (2.0.2)
+    method_source (0.8.2)
+    mini_portile2 (2.1.0)
+    multi_xml (0.5.5)
+    nokogiri (1.6.8)
+      mini_portile2 (~> 2.1.0)
+      pkg-config (~> 1.1.7)
+    pkg-config (1.1.7)
+    pry (0.10.4)
+      coderay (~> 1.1.0)
+      method_source (~> 0.8.1)
+      slop (~> 3.4)
+    pry-byebug (3.4.0)
+      byebug (~> 9.0)
+      pry (~> 0.10)
+    rake (11.2.2)
+    rspec (3.5.0)
+      rspec-core (~> 3.5.0)
+      rspec-expectations (~> 3.5.0)
+      rspec-mocks (~> 3.5.0)
+    rspec-core (3.5.2)
+      rspec-support (~> 3.5.0)
+    rspec-expectations (3.5.0)
+      diff-lcs (>= 1.2.0, < 2.0)
+      rspec-support (~> 3.5.0)
+    rspec-mocks (3.5.0)
+      diff-lcs (>= 1.2.0, < 2.0)
+      rspec-support (~> 3.5.0)
+    rspec-support (3.5.0)
+    simplecov (0.12.0)
+      docile (~> 1.1.0)
+      json (>= 1.8, < 3)
+      simplecov-html (~> 0.10.0)
+    simplecov-html (0.10.0)
+    slop (3.6.0)
+
+PLATFORMS
+  ruby
+
+DEPENDENCIES
+  bundler (~> 1.6)
+  codeclimate-test-reporter
+  dotenv
+  multi_xml
+  nokogiri
+  pry-byebug
+  rake
+  rspec
+  shoryuken!
+  simplecov
+
+BUNDLED WITH
+   1.13.6

Procfile

@@ -0,0 +1,2 @@
+worker: bundle exec bin/shoryuken -r ./putsreq_worker.rb -q default
+

Rakefile

@@ -10,7 +10,6 @@ end
 desc 'Open Shoryuken pry console'
 task :console do
   require 'pry'
-  require 'celluloid/current'
   require 'shoryuken'
 
   config_file = File.join File.expand_path('..', __FILE__), 'shoryuken.yml'
@@ -28,3 +27,15 @@ task :console do
   ARGV.clear
   Pry.start
 end
+
+desc 'Enqueue 1k messages'
+task :enqueue_1k do
+  require 'shoryuken'
+
+  require File.join(File.expand_path('..', __FILE__), 'putsreq_worker')
+
+  1000.times do |i|
+    PutsReqWorker.perform_async(i.to_s)
+    print '.'
+  end
+end

examples/default_worker.rb

@@ -4,6 +4,6 @@ class DefaultWorker
   shoryuken_options queue: 'default', auto_delete: true
 
   def perform(sqs_msg, body)
-    Shoryuken.logger.info("Received message: '#{body}'")
+    Shoryuken.logger.debug("Received message: '#{body}'")
   end
 end

lib/shoryuken.rb

@@ -1,6 +1,8 @@
 require 'yaml'
+require 'json'
 require 'aws-sdk-core'
 require 'time'
+require 'concurrent'
 
 require 'shoryuken/version'
 require 'shoryuken/core_ext'
@@ -22,6 +24,10 @@
 require 'shoryuken/sns_arn'
 require 'shoryuken/topic'
 require 'shoryuken/polling'
+require 'shoryuken/manager'
+require 'shoryuken/launcher'
+require 'shoryuken/processor'
+require 'shoryuken/fetcher'
 
 module Shoryuken
   DEFAULTS = {

lib/shoryuken/cli.rb

@@ -14,8 +14,6 @@ class CLI
     include Util
     include Singleton
 
-    attr_accessor :launcher
-
     def run(args)
       self_read, self_write = IO.pipe
 
@@ -41,9 +39,8 @@ def run(args)
 
       loader.load
 
-      load_celluloid
+      initialize_concurrent_logger
 
-      require 'shoryuken/launcher'
       @launcher = Shoryuken::Launcher.new
 
       if callback = Shoryuken.start_callback
@@ -54,43 +51,33 @@ def run(args)
       fire_event(:startup)
 
       begin
-        launcher.run
+        @launcher.run
 
         while (readable_io = IO.select([self_read]))
           signal = readable_io.first[0].gets.strip
           handle_signal(signal)
         end
       rescue Interrupt
-        launcher.stop(shutdown: true)
+        @launcher.stop(shutdown: true)
         exit 0
       end
     end
 
     private
 
-    def load_celluloid
-      require 'celluloid/current'
-      Celluloid.logger = (Shoryuken.options[:verbose] ? Shoryuken.logger : nil)
-
-      require 'shoryuken/manager'
-    end
+    def initialize_concurrent_logger
+      return unless Shoryuken.logger
 
-    def celluloid_loaded?
-      defined?(::Celluloid)
+      Concurrent.global_logger = lambda do |level, progname, msg = nil, &block|
+        Shoryuken.logger.log(level, msg, progname, &block)
+      end
     end
 
     def daemonize(options)
       return unless options[:daemon]
 
       fail ArgumentError, "You really should set a logfile if you're going to daemonize" unless options[:logfile]
 
-      if celluloid_loaded?
-        # Celluloid can't be loaded until after we've daemonized
-        # because it spins up threads and creates locks which get
-        # into a very bad state if forked.
-        raise "Celluloid cannot be required until here, or it will break Shoryuken's daemonization"
-      end
-
       files_to_reopen = []
       ObjectSpace.each_object(File) do |file|
         files_to_reopen << file unless file.closed?
@@ -187,7 +174,7 @@ def handle_signal(sig)
       when 'USR1'
         logger.info { 'Received USR1, will soft shutdown down' }
 
-        launcher.stop
+        @launcher.stop
         fire_event(:quiet, true)
         exit 0
       when 'TTIN'
@@ -200,11 +187,6 @@ def handle_signal(sig)
           end
         end
 
-        ready  = launcher.manager.instance_variable_get(:@ready).size
-        busy   = launcher.manager.instance_variable_get(:@busy).size
-        queues = launcher.manager.instance_variable_get(:@queues)
-
-        logger.info { "Ready: #{ready}, Busy: #{busy}, Active Queues: #{unparse_queues(queues)}" }
       else
         logger.info { "Received #{sig}, will shutdown down" }
 

lib/shoryuken/fetcher.rb

@@ -5,23 +5,21 @@ class Fetcher
     FETCH_LIMIT = 10
 
     def fetch(queue, available_processors)
-      watchdog('Fetcher#fetch died') do
-        started_at = Time.now
-
-        logger.debug { "Looking for new messages in '#{queue}'" }
-
-        begin
-          limit = available_processors > FETCH_LIMIT ? FETCH_LIMIT : available_processors
-
-          sqs_msgs = Array(receive_messages(queue, limit))
-          logger.info { "Found #{sqs_msgs.size} messages for '#{queue.name}'" }
-          logger.debug { "Fetcher for '#{queue}' completed in #{elapsed(started_at)} ms" }
-          sqs_msgs
-        rescue => ex
-          logger.error { "Error fetching message: #{ex}" }
-          logger.error { ex.backtrace.first }
-          []
-        end
+      started_at = Time.now
+
+      logger.debug { "Looking for new messages in '#{queue}'" }
+
+      begin
+        limit = available_processors > FETCH_LIMIT ? FETCH_LIMIT : available_processors
+
+        sqs_msgs = Array(receive_messages(queue, limit))
+        logger.info { "Found #{sqs_msgs.size} messages for '#{queue.name}'" }
+        logger.debug { "Fetcher for '#{queue}' completed in #{elapsed(started_at)} ms" }
+        sqs_msgs
+      rescue => ex
+        logger.error { "Error fetching message: #{ex}" }
+        logger.error { ex.backtrace.first }
+        []
       end
     end
 

lib/shoryuken/launcher.rb

@@ -1,42 +1,29 @@
 module Shoryuken
   class Launcher
-    include Celluloid
     include Util
 
-    trap_exit :actor_died
-
-    attr_accessor :manager
-
     def initialize
-      @condvar = Celluloid::Condition.new
-      @manager = Shoryuken::Manager.new_link(@condvar)
+      count = Shoryuken.options.fetch(:concurrency, 25)
 
-      @done = false
+      raise(ArgumentError, "Concurrency value #{count} is invalid, it needs to be a positive number") unless count > 0
 
-      manager.fetcher = Shoryuken::Fetcher.new
-      manager.polling_strategy = Shoryuken.options[:polling_strategy].new(Shoryuken.queues)
+      @managers = Array.new(count) do
+        Shoryuken::Manager.new(1,
+                               Shoryuken::Fetcher.new,
+                               Shoryuken.options[:polling_strategy].new(Shoryuken.queues))
+      end
     end
 
     def stop(options = {})
-      watchdog('Launcher#stop') do
-        @done = true
-
-        manager.async.stop(shutdown: !!options[:shutdown], timeout: Shoryuken.options[:timeout])
-        @condvar.wait
-        manager.terminate
-      end
+      @managers.map do |manager|
+        Thread.new { manager.stop(shutdown: !!options[:shutdown], timeout: Shoryuken.options[:timeout]) }
+      end.each(&:join)
     end
 
     def run
-      watchdog('Launcher#run') do
-        manager.async.start
-      end
-    end
-
-    def actor_died(actor, reason)
-      return if @done
-      logger.warn { "Shoryuken died due to the following error, cannot recover, process exiting: #{reason}" }
-      exit 1
+      @managers.map do |manager|
+        Thread.new { manager.start }
+      end.each(&:join)
     end
   end
 end