removed started? and use AtomicBoolean for ready?

cleanup @filter_to_output initialization

remove @run_mutex, use ready? to synchonize shutdown agains startup sequence

remove ShutdownSignal handling & cleanup

lib/logstash/pipeline.rb

@@ -1,6 +1,7 @@
 # encoding: utf-8
-require "thread" #
+require "thread"
 require "stud/interval"
+require "concurrent"
 require "logstash/namespace"
 require "logstash/errors"
 require "logstash/event"
@@ -34,29 +35,19 @@ class LogStash::Pipeline
     end
 
     @input_to_filter = SizedQueue.new(20)
-
-    # If no filters, pipe inputs directly to outputs
-    if !filters?
-      @filter_to_output = @input_to_filter
-    else
-      @filter_to_output = SizedQueue.new(20)
-    end
+    # if no filters, pipe inputs directly to outputs
+    @filter_to_output = filters? ? SizedQueue.new(20) : @input_to_filter
     @settings = {
       "filter-workers" => 1,
     }
 
-    @run_mutex = Mutex.new
-    @ready = false
-    @started = false
+    # @ready requires thread safety since it is typically polled from outside the pipeline thread
+    @ready = Concurrent::AtomicBoolean.new(false)
     @input_threads = []
   end # def initialize
 
   def ready?
-    @run_mutex.synchronize{@ready}
-  end
-
-  def started?
-    @run_mutex.synchronize{@started}
+    @ready.value
   end
 
   def configure(setting, value)
@@ -75,14 +66,15 @@ class LogStash::Pipeline
   end
 
   def run
-    @run_mutex.synchronize{@started = true}
-
-    # synchronize @input_threads between run and shutdown
-    @run_mutex.synchronize{start_inputs}
-    start_filters if filters?
-    start_outputs
-
-    @run_mutex.synchronize{@ready = true}
+    begin
+      start_inputs
+      start_filters if filters?
+      start_outputs
+    ensure
+      # it is important to garantee @ready to be true after the startup sequence has been completed
+      # to potentially unblock the shutdown method which may be waiting on @ready to proceed
+      @ready.make_true
+    end
 
     @logger.info("Pipeline started")
     @logger.terminal("Logstash startup completed")
@@ -246,25 +238,20 @@ class LogStash::Pipeline
     end
   end # def outputworker
 
-  # Shutdown this pipeline.
-  #
-  # This method is intended to be called from another thread
+  # initiate the pipeline shutdown sequence
+  # this method is intended to be called from outside the pipeline thread
   def shutdown
+    # shutdown can only start once the pipeline has completed its startup.
+    # avoid potential race conditoon between the startup sequence and this
+    # shutdown method which can be called from another thread at any time
+    sleep(0.1) while !ready?
+
+    # TODO: should we also check against calling shutdown multiple times concurently?
+
     InflightEventsReporter.logger = @logger
     InflightEventsReporter.start(@input_to_filter, @filter_to_output, @outputs)
 
-    # sometimes an input is stuck in a blocking I/O so we need to tell it to teardown directly
-    @inputs.each do |input|
-      begin
-        # input teardown must be synchronized since is can be called concurrently by
-        # the input worker thread and from the pipeline thread shutdown method.
-        # this means that input teardown methods must support multiple calls.
-        @run_mutex.synchronize{input.stop}
-      rescue LogStash::ShutdownSignal
-        # teardown could receive the ShutdownSignal, retry it
-        retry
-      end
-    end
+    @inputs.each(&:stop)
   end # def shutdown
 
   def plugin(plugin_type, name, *args)