Use thread pools

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -3,7 +3,9 @@
 require "logstash/inputs/base"
 require "logstash/inputs/threadable"
 require 'redis'
+require 'concurrent'
 require 'concurrent/atomics'
+require 'concurrent/executors'
 
 # This input will read events from a Redis instance; it supports both Redis channels and lists.
 # The list command (BLPOP) used by Logstash is supported in Redis v1.3.1+, and
@@ -230,77 +232,81 @@ def list_pattern_stop
 
     @redis.quit rescue nil
     @redis = nil
-    @worker_threads.each do |key, thread|
-      if thread.alive?
-        thread.join(5)
-        @logger.warn("Thread #{key} joined")
-      end
-    end
+    @threadpool.shutdown
+    @threadpool.wait_for_termination
   end
 
+  # private
   def launch_worker(output_queue, key)
-     Thread.new do
-      redis = new_redis_instance
-      begin
-        (0..MAX_ITEMS_PER_WORKER).each do
-          if stop?
-            @logger.warn("Breaking from thread #{key} as it was requested to stop")
-            break
-          end
-          value = redis.lpop(key)
-          break if value.nil?
-          queue_event(value, output_queue)
-          @items_processed.increment
+    @logger.debug("Launched worker for #{key}")
+    redis = new_redis_instance
+    begin
+      (0..MAX_ITEMS_PER_WORKER).each do
+        if stop?
+          @logger.debug("Breaking from thread #{key} as it was requested to stop")
+          break
         end
-      ensure
-        redis.quit rescue nil
+        value = redis.lpop(key)
+        break if value.nil?
+        queue_event(value, output_queue)
+        @items_processed.increment
       end
+    ensure
+      redis.quit rescue nil
     end
+    @logger.debug("Exit worker for #{key}")
   end
 
-  def clear_finished_workers
-    finished_threads = []
-    @worker_threads.each do |key, thread|
-      next if thread.alive?
-      finished_threads << key
-    end
-    finished_threads.each { |key| @worker_threads.delete(key) }
+  # private
+  def threadpool_capacity?
+    @threadpool.remaining_capacity > 0
   end
 
+  # private
   def ensure_workers(output_queue)
-    free_slots = @worker_thread_count - @worker_threads.length
-    return if free_slots == 0
+    return unless threadpool_capacity?
     keys = @redis.keys(@key)
     keys.shuffle
     keys.each do |key|
-      next if @worker_threads.has_key?(key)
-      @worker_threads[key] = launch_worker(output_queue, key)
-      free_slots -= 1
-      break if free_slots == 0
+      break unless threadpool_capacity?
+      next if @current_workers.include?(key)
+      @current_workers.add(key)
+      @threadpool.post do
+        begin
+          launch_worker(output_queue, key)
+        ensure
+          @current_workers.delete(key)
+        end
+      end
     end
   end
 
+  # private
   def init_list_pattern_runner(output_queue)
-    @worker_threads ||= {}
-    @redis ||= connect
+    @threadpool_queue_max ||= 2 * @worker_thread_count
+    @threadpool ||= Concurrent::ThreadPoolExecutor.new(
+        min_threads: @worker_thread_count,
+        max_threads: @worker_thread_count,
+        max_queue: @threadpool_queue_max
+    )
     @items_processed ||= Concurrent::AtomicFixnum.new
-    clear_finished_workers
-    ensure_workers(output_queue)
+    @current_workers ||= Concurrent::Set.new
+    @redis ||= connect
   end
 
   # private
   def list_pattern_runner(output_queue)
-    items = 0
     while !stop?
       begin
-        init_list_pattern_runner(output_queue)
+        init_list_pattern_runner(output_queue) if @redis.nil?
+        ensure_workers(output_queue)
         sleep(0.1)
       rescue ::Redis::BaseError => e
         @logger.warn("Redis connection problem", :exception => e)
         @redis = nil
       end
     end
-    @logger.warn("Total items processed: #{@items_processed}")
+    @logger.debug("Total items processed: #{@items_processed.value}")
   end
 
   def list_batch_listener(redis, output_queue)