A manager every 10 concurrency

ruby-shoryuken · Dec 24, 2016 · 540b45a · 540b45a
1 parent 59bc99b
commit 540b45a
Show file tree

Hide file tree

Showing 9 changed files with 23 additions and 9 deletions.
diff --git a/Thread-pool 0.05.png b/Thread-pool 0.05.png
diff --git a/Thread-pool 0.10.png b/Thread-pool 0.10.png
diff --git a/Thread-pool 0.15.png b/Thread-pool 0.15.png
diff --git a/Thread-pool 0.25.png b/Thread-pool 0.25.png
diff --git a/Thread-pool 1.png b/Thread-pool 1.png
diff --git a/Thread-pool done & dispatch.png b/Thread-pool done & dispatch.png
diff --git a/Thread-pool wait 10 seconds.png b/Thread-pool wait 10 seconds.png
diff --git a/lib/shoryuken/launcher.rb b/lib/shoryuken/launcher.rb
@@ -3,17 +3,32 @@ class Launcher
     include Util
 
     def initialize
-      @manager = Shoryuken::Manager.new(Shoryuken::Fetcher.new,
-                                        Shoryuken.options[:polling_strategy].new(Shoryuken.queues))
+      count = Shoryuken.options.fetch(:concurrency, 25)
+
+      raise(ArgumentError, "Concurrency value #{count} is invalid, it needs to be a positive number") unless count > 0
+
+      manager_count = count / 10
+      manager_count = 1 if manager_count < 1
+
+      concurrency = count / manager_count
+
+      @managers = Array.new(manager_count) do
+        Shoryuken::Manager.new(concurrency,
+                               Shoryuken::Fetcher.new,
+                               Shoryuken.options[:polling_strategy].new(Shoryuken.queues))
+      end
     end
 
     def stop(options = {})
-      @manager.stop(shutdown: !!options[:shutdown],
-                    timeout: Shoryuken.options[:timeout])
+      @managers.map do |manager|
+        Thread.new { manager.stop(shutdown: !!options[:shutdown], timeout: Shoryuken.options[:timeout]) }
+      end.each(&:join)
     end
 
     def run
-      @manager.start
+      @managers.map do |manager|
+        Thread.new { manager.start }
+      end.each(&:join)
     end
   end
 end
diff --git a/lib/shoryuken/manager.rb b/lib/shoryuken/manager.rb
@@ -4,9 +4,8 @@ class Manager
 
     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
+    def initialize(count, fetcher, polling_strategy)
+      @count = count
 
       @queues = Shoryuken.queues.dup.uniq
 
@@ -16,7 +15,7 @@ def initialize(fetcher, polling_strategy)
       @fetcher = fetcher
       @polling_strategy = polling_strategy
 
-      @heartbeat = Concurrent::TimerTask.new(run_now: true, execution_interval: 0.15, timeout_interval: 60) { dispatch }
+      @heartbeat = Concurrent::TimerTask.new(run_now: true, execution_interval: 0.1, timeout_interval: 60) { dispatch }
 
       @pool = Concurrent::FixedThreadPool.new(@count, max_queue: @count)
     end