diff --git a/logstash-core/lib/logstash/agent.rb b/logstash-core/lib/logstash/agent.rb
index 1f307c61838..27a2d60ae2c 100644
--- a/logstash-core/lib/logstash/agent.rb
+++ b/logstash-core/lib/logstash/agent.rb
@@ -76,4 +76,4 @@ def trap_sigint
       end
     end
   end
-end # class LogStash::Agent
+end # class LogStash::Agent
\ No newline at end of file
diff --git a/logstash-core/lib/logstash/config/config_ast.rb b/logstash-core/lib/logstash/config/config_ast.rb
index 0c25e05d6ab..23b9f2403da 100644
--- a/logstash-core/lib/logstash/config/config_ast.rb
+++ b/logstash-core/lib/logstash/config/config_ast.rb
@@ -108,6 +108,7 @@ def compile
         # defines @filter_func and @output_func
 
         definitions << "def #{type}_func(event)"
+        definitions << "  targeted_outputs = []" if type == "output"
         definitions << "  events = [event]" if type == "filter"
         definitions << "  @logger.debug? && @logger.debug(\"#{type} received\", :event => event.to_hash)"
 
@@ -116,6 +117,7 @@ def compile
         end
 
         definitions << "  events" if type == "filter"
+        definitions << "  targeted_outputs" if type == "output"
         definitions << "end"
       end
 
@@ -142,7 +144,7 @@ def compile_initializer
 
 
         code << <<-CODE
-          #{name} = #{plugin.compile_initializer}
+#{name} = #{plugin.compile_initializer}
           @#{plugin.plugin_type}s << #{name}
         CODE
 
@@ -150,7 +152,7 @@ def compile_initializer
         if plugin.plugin_type == "filter"
 
           code << <<-CODE
-            #{name}_flush = lambda do |options, &block|
+#{name}_flush = lambda do |options, &block|
               @logger.debug? && @logger.debug(\"Flushing\", :plugin => #{name})
 
               events = #{name}.flush(options)
@@ -230,18 +232,18 @@ def compile_initializer
 
     def compile
       case plugin_type
-      when "input"
-        return "start_input(#{variable_name})"
-      when "filter"
-        return <<-CODE
+        when "input"
+          return "start_input(#{variable_name})"
+        when "filter"
+          return <<-CODE
           events = #{variable_name}.multi_filter(events)
-        CODE
-      when "output"
-        return "#{variable_name}.handle(event)\n"
-      when "codec"
-        settings = attributes.recursive_select(Attribute).collect(&:compile).reject(&:empty?)
-        attributes_code = "LogStash::Util.hash_merge_many(#{settings.map { |c| "{ #{c} }" }.join(", ")})"
-        return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{attributes_code})"
+          CODE
+        when "output"
+          return "targeted_outputs << #{variable_name}\n"
+        when "codec"
+          settings = attributes.recursive_select(Attribute).collect(&:compile).reject(&:empty?)
+          attributes_code = "LogStash::Util.hash_merge_many(#{settings.map { |c| "{ #{c} }" }.join(", ")})"
+          return "plugin(#{plugin_type.inspect}, #{plugin_name.inspect}, #{attributes_code})"
       end
     end
 
@@ -345,7 +347,7 @@ def validate!
                  :column => input.column_of(interval.first),
                  :byte => interval.first + 1,
                  :after => input[0..interval.first]
-                )
+          )
         )
       end
     end
@@ -402,9 +404,9 @@ def cond_func_#{i}(input_events)
         <<-CODE
           events = cond_func_#{i}(events)
         CODE
-      else
+      else # Output
         <<-CODE
-          #{super}
+#{super}
           end
         CODE
       end
@@ -525,7 +527,7 @@ def _inspect(indent="")
     tv = "...#{tv[-20..-1]}" if tv.size > 20
 
     indent +
-    self.class.to_s.sub(/.*:/,'') +
+      self.class.to_s.sub(/.*:/,'') +
       em.map{|m| "+"+m.to_s.sub(/.*:/,'')}*"" +
       " offset=#{interval.first}" +
       ", #{tv.inspect}" +
@@ -533,13 +535,13 @@ def _inspect(indent="")
       (elements && elements.size > 0 ?
         ":" +
           (elements.select { |e| !e.is_a?(LogStash::Config::AST::Whitespace) && e.elements && e.elements.size > 0 }||[]).map{|e|
-      begin
-        "\n"+e.inspect(indent+"  ")
-      rescue  # Defend against inspect not taking a parameter
-        "\n"+indent+" "+e.inspect
-      end
+            begin
+              "\n"+e.inspect(indent+"  ")
+            rescue  # Defend against inspect not taking a parameter
+              "\n"+indent+" "+e.inspect
+            end
           }.join("") :
         ""
       )
   end
-end
+end
\ No newline at end of file
diff --git a/logstash-core/lib/logstash/outputs/base.rb b/logstash-core/lib/logstash/outputs/base.rb
index 29bf7f7e191..0a9af01084e 100644
--- a/logstash-core/lib/logstash/outputs/base.rb
+++ b/logstash-core/lib/logstash/outputs/base.rb
@@ -4,6 +4,8 @@
 require "logstash/plugin"
 require "logstash/namespace"
 require "logstash/config/mixin"
+require "logstash/util/wrapped_synchronous_queue"
+require "concurrent/atomic/atomic_fixnum"
 
 class LogStash::Outputs::Base < LogStash::Plugin
   include LogStash::Config::Mixin
@@ -23,7 +25,7 @@ class LogStash::Outputs::Base < LogStash::Plugin
   # Note that this setting may not be useful for all outputs.
   config :workers, :validate => :number, :default => 1
 
-  attr_reader :worker_plugins, :worker_queue, :worker_threads
+  attr_reader :worker_plugins, :available_workers, :workers, :single_worker_mutex, :is_multi_worker, :worker_plugins
 
   public
   def workers_not_supported(message=nil)
@@ -40,6 +42,11 @@ def workers_not_supported(message=nil)
   def initialize(params={})
     super
     config_init(params)
+
+    # If we're running with a single thread we must enforce single-threaded concurrency by default
+    # Maybe in a future version we'll assume output plugins are threadsafe
+    @single_worker_mutex = Mutex.new
+    worker_setup
   end
 
   public
@@ -54,36 +61,59 @@ def receive(event)
 
   public
   def worker_setup
+    # TODO: Remove this branch, delete this function
     if @workers == 1
+      @is_multi_worker = false
       @worker_plugins = [self]
-      @worker_threads = []
     else
-      define_singleton_method(:handle, method(:handle_worker))
-      @worker_queue = SizedQueue.new(20)
+      @is_multi_worker = true
+      define_singleton_method(:multi_handle, method(:handle_worker))
+
       @worker_plugins = @workers.times.map { self.class.new(@original_params.merge("workers" => 1)) }
-      @worker_threads = @worker_plugins.map.with_index do |plugin, i|
-        Thread.new(original_params, @worker_queue) do |params, queue|
-          LogStash::Util.set_thread_name(">#{self.class.config_name}.#{i}")
-          LogStash::Util.set_thread_plugin(self)
-          plugin.register
-          while true
-            event = queue.pop
-            plugin.handle(event)
-          end
-        end
+
+      @available_workers = SizedQueue.new(@worker_plugins.length)
+
+      @worker_plugins.each do |wp|
+        wp.register
+        @available_workers << wp
       end
     end
   end
 
   public
+  # Not to be overriden by plugin authors!
   def handle(event)
-    LogStash::Util.set_thread_plugin(self)
-    receive(event)
+    @single_worker_mutex.synchronize { receive(event) }
   end # def handle
 
-  def handle_worker(event)
-    LogStash::Util.set_thread_plugin(self)
-    @worker_queue.push(event)
+  # To be overriden in implementations
+  def multi_receive(events)
+    events.each {|event|
+      receive(event)
+    }
+  end
+
+  # Not to be overriden by plugin authors!
+  def multi_handle(events)
+    @single_worker_mutex.synchronize { multi_receive(events) }
+  end
+
+  def handle_worker(events)
+    worker = @available_workers.pop
+    begin
+      worker.multi_receive(events)
+    ensure
+      @available_workers.push(worker)
+    end
+  end
+
+  def do_close
+    if @worker_plugins
+      @worker_plugins.each do |wp|
+        wp.do_close unless wp === self
+      end
+    end
+    super
   end
 
   private
@@ -91,4 +121,4 @@ def output?(event)
     # TODO: noop for now, remove this once we delete this call from all plugins
     true
   end # def output?
-end # class LogStash::Outputs::Base
+end # class LogStash::Outputs::Base
\ No newline at end of file
diff --git a/logstash-core/lib/logstash/pipeline.rb b/logstash-core/lib/logstash/pipeline.rb
index 48aa1c8a83d..8e091c7075a 100644
--- a/logstash-core/lib/logstash/pipeline.rb
+++ b/logstash-core/lib/logstash/pipeline.rb
@@ -12,20 +12,24 @@
 require "logstash/config/cpu_core_strategy"
 require "logstash/util/defaults_printer"
 require "logstash/shutdown_controller"
+require "logstash/util/wrapped_synchronous_queue"
+require "logstash/pipeline_reporter"
 
 module LogStash; class Pipeline
-  attr_reader :inputs, :filters, :outputs, :input_to_filter, :filter_to_output
+  attr_reader :inputs, :filters, :outputs, :worker_threads, :events_consumed, :events_filtered, :reporter
 
   def initialize(config_str, settings = {})
     @logger = Cabin::Channel.get(LogStash)
+    @reporter = LogStash::PipelineReporter.new(@logger, self)
 
     @inputs = nil
     @filters = nil
     @outputs = nil
 
+    @worker_threads = []
+
     grammar = LogStashConfigParser.new
     @config = grammar.parse(config_str)
-
     if @config.nil?
       raise LogStash::ConfigurationError, grammar.failure_reason
     end
@@ -42,12 +46,21 @@ def initialize(config_str, settings = {})
       raise
     end
 
-    @input_to_filter = SizedQueue.new(20)
-    # if no filters, pipe inputs directly to outputs
-    @filter_to_output = filters? ? SizedQueue.new(20) : @input_to_filter
+    @input_queue = LogStash::Util::WrappedSynchronousQueue.new
+    @events_filtered = Concurrent::AtomicFixnum.new(0)
+    @events_consumed = Concurrent::AtomicFixnum.new(0)
+
+    # We generally only want one thread at a time able to access pop/take/poll operations
+    # from this queue. We also depend on this to be able to block consumers while we snapshot
+    # in-flight buffers
+    @input_queue_pop_mutex = Mutex.new
+
+    @input_threads = []
 
     @settings = {
-      "default-filter-workers" => LogStash::Config::CpuCoreStrategy.fifty_percent
+      "default-pipeline-workers" => LogStash::Config::CpuCoreStrategy.fifty_percent,
+      "batch-size" => 125,
+      "batch-poll-wait" => 50 # in milliseconds
     }
 
     # @ready requires thread safety since it is typically polled from outside the pipeline thread
@@ -64,25 +77,26 @@ def configure(setting, value)
     @settings[setting] = value
   end
 
-  def safe_filter_worker_count
-    default = @settings["default-filter-workers"]
-    thread_count = @settings["filter-workers"] #override from args "-w 8" or config
+  def safe_pipeline_worker_count
+    default = @settings["default-pipeline-workers"]
+    thread_count = @settings["pipeline-workers"] #override from args "-w 8" or config
     safe_filters, unsafe_filters = @filters.partition(&:threadsafe?)
+
     if unsafe_filters.any?
       plugins = unsafe_filters.collect { |f| f.class.config_name }
       case thread_count
-      when nil
-        # user did not specify a worker thread count
-        # warn if the default is multiple
-        @logger.warn("Defaulting filter worker threads to 1 because there are some filters that might not work with multiple worker threads",
-          :count_was => default, :filters => plugins) if default > 1
-        1 # can't allow the default value to propagate if there are unsafe filters
-      when 0, 1
-        1
-      else
-        @logger.warn("Warning: Manual override - there are filters that might not work with multiple worker threads",
-          :worker_threads => thread_count, :filters => plugins)
-        thread_count # allow user to force this even if there are unsafe filters
+        when nil
+          # user did not specify a worker thread count
+          # warn if the default is multiple
+          @logger.warn("Defaulting pipeline worker threads to 1 because there are some filters that might not work with multiple worker threads",
+                       :count_was => default, :filters => plugins) if default > 1
+          1 # can't allow the default value to propagate if there are unsafe filters
+        when 0, 1
+          1
+        else
+          @logger.warn("Warning: Manual override - there are filters that might not work with multiple worker threads",
+                       :worker_threads => thread_count, :filters => plugins)
+          thread_count # allow user to force this even if there are unsafe filters
       end
     else
       thread_count || default
@@ -94,60 +108,162 @@ def filters?
   end
 
   def run
+    LogStash::Util.set_thread_name(">lsipeline")
     @logger.terminal(LogStash::Util::DefaultsPrinter.print(@settings))
 
+    start_workers
+
+    @logger.info("Pipeline started")
+    @logger.terminal("Logstash startup completed")
+
+    @logger.info("Will run till input threads stopped")
+
+    # Block until all inputs have stopped
+    # Generally this happens if SIGINT is sent and `shutdown` is called from an external thread
+    wait_inputs
+    @logger.info("Inputs stopped")
+
+    shutdown_workers
+
+    @logger.info("Pipeline shutdown complete.")
+    @logger.terminal("Logstash shutdown completed")
+
+    # exit code
+    return 0
+  end # def run
+
+  def start_workers
+    @inflight_batches = {}
+
+    @worker_threads.clear # In case we're restarting the pipeline
     begin
       start_inputs
-      start_filters if filters?
-      start_outputs
+      @outputs.each {|o| o.register }
+      @filters.each {|f| f.register}
+
+      pipeline_workers = safe_pipeline_worker_count
+      batch_size = @settings['batch-size']
+      batch_poll_wait = @settings['batch-poll-wait']
+      @logger.info("Starting pipeline",
+                   :id => self.object_id,
+                   :settings => @settings)
+
+      pipeline_workers.times do |t|
+        @worker_threads << Thread.new do
+          LogStash::Util.set_thread_name(">worker#{t}")
+          worker_loop(batch_size, batch_poll_wait)
+        end
+      end
     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
+  end
 
-    @logger.info("Pipeline started")
-    @logger.terminal("Logstash startup completed")
+  # Main body of what a worker threda does
+  # Repeatedly takes batches off the queu, filters, then outputs them
+  def worker_loop(batch_size, batch_poll_wait)
+    running = true
 
-    wait_inputs
+    while running
+      # To understand the purpose behind this synchronize please read the body of take_batch
+      input_batch, shutdown_received = @input_queue_pop_mutex.synchronize { take_batch(batch_size, batch_poll_wait) }
+      running = false if shutdown_received
+      @events_consumed.increment(input_batch.size)
 
-    if filters?
-      shutdown_filters
-      wait_filters
-      flush_filters_to_output!(:final => true)
+      filtered = filter_batch(input_batch)
+      @events_filtered.increment(filtered.size)
+
+      output_batch(filtered)
+
+      inflight_batches_synchronize { set_current_thread_inflight_batch(nil) }
     end
+  end
 
-    shutdown_outputs
-    wait_outputs
+  def take_batch(batch_size, batch_poll_wait)
+    batch = []
+    # Since this is externally synchronized in `worker_look` wec can guarantee that the visibility of an insight batch
+    # guaranteed to be a full batch not a partial batch
+    set_current_thread_inflight_batch(batch)
+
+    shutdown_received = false
+    batch_size.times do |t|
+      event = t==0 ? @input_queue.take : @input_queue.poll(batch_poll_wait)
+      # Exit early so each thread only gets one copy of this
+      # This is necessary to ensure proper shutdown!
+      next if event.nil?
+
+      if event == LogStash::SHUTDOWN
+        shutdown_received = true
+        break
+      end
 
-    @logger.info("Pipeline shutdown complete.")
-    @logger.terminal("Logstash shutdown completed")
+      batch << event
+    end
 
-    # exit code
-    return 0
-  end # def run
+    [batch, shutdown_received]
+  end
 
-  def wait_inputs
-    @input_threads.each(&:join)
+  def filter_batch(batch)
+    batch.reduce([]) do |acc,e|
+      if e.is_a?(LogStash::Event)
+        filtered = filter_func(e)
+        filtered.each {|fe| acc << fe unless fe.cancelled?}
+      end
+      acc
+    end
+  rescue Exception => e
+    # Plugins authors should manage their own exceptions in the plugin code
+    # but if an exception is raised up to the worker thread they are considered
+    # fatal and logstash will not recover from this situation.
+    #
+    # Users need to check their configuration or see if there is a bug in the
+    # plugin.
+    @logger.error("Exception in pipelineworker, the pipeline stopped processing new events, please check your filter configuration and restart Logstash.",
+                  "exception" => e, "backtrace" => e.backtrace)
+    raise
   end
 
-  def shutdown_filters
-    @flusher_thread.kill
-    @input_to_filter.push(LogStash::SHUTDOWN)
+  # Take an array of events and send them to the correct output
+  def output_batch(batch)
+    batch.reduce(Hash.new { |h, k| h[k] = [] }) do |outputs_events, event|
+      # We ask the AST to tell us which outputs to send each event to
+      output_func(event).each do |output|
+        outputs_events[output] << event
+      end
+      outputs_events
+    end.each do |output, events|
+      # Once we have a mapping of outputs => [events] we can execute them
+      output.multi_handle(events)
+    end
   end
 
-  def wait_filters
-    @filter_threads.each(&:join) if @filter_threads
+  def set_current_thread_inflight_batch(batch)
+    @inflight_batches[Thread.current] = batch
   end
 
-  def shutdown_outputs
-    # nothing, filters will do this
-    @filter_to_output.push(LogStash::SHUTDOWN)
+  def inflight_batches_synchronize
+    @input_queue_pop_mutex.synchronize do
+      yield(@inflight_batches)
+    end
   end
 
-  def wait_outputs
-    # Wait for the outputs to stop
-    @output_threads.each(&:join)
+  def dump_inflight(file_path)
+    inflight_batches_synchronize do |batches|
+      File.open(file_path, "w") do |f|
+        batches.values.each do |batch|
+          next unless batch
+          batch.each do |e|
+            f.write(LogStash::Json.dump(e))
+          end
+        end
+      end
+    end
+  end
+
+  def wait_inputs
+    @input_threads.each(&:join)
   end
 
   def start_inputs
@@ -167,45 +283,16 @@ def start_inputs
     end
   end
 
-  def start_filters
-    @filters.each(&:register)
-    # dynamically get thread count based on filter threadsafety
-    # moved this test to here to allow for future config reloading
-    to_start = safe_filter_worker_count
-    @filter_threads = to_start.times.collect do |i|
-      Thread.new do
-        LogStash::Util.set_thread_name("|filterworker.#{i}")
-        filterworker
-      end
-    end
-    actually_started = @filter_threads.select(&:alive?).size
-    msg = "Worker threads expected: #{to_start}, worker threads started: #{actually_started}"
-    if actually_started < to_start
-      @logger.warn(msg)
-    else
-      @logger.info(msg)
-    end
-    @flusher_thread = Thread.new { Stud.interval(5) { @input_to_filter.push(LogStash::FLUSH) } }
-  end
-
-  def start_outputs
-    @outputs.each(&:register)
-    @output_threads = [
-      Thread.new { outputworker }
-    ]
-  end
-
   def start_input(plugin)
     @input_threads << Thread.new { inputworker(plugin) }
   end
 
   def inputworker(plugin)
-    LogStash::Util.set_thread_name("<#{plugin.class.config_name}")
-    LogStash::Util.set_thread_plugin(plugin)
+    LogStash::Util::set_thread_name("<#{plugin.class.config_name}")
     begin
-      plugin.run(@input_to_filter)
+      plugin.run(@input_queue)
     rescue => e
-      # if plugin is stopping, ignore uncatched exceptions and exit worker
+      # if plugin is stop
       if plugin.stop?
         @logger.debug("Input plugin raised exception during shutdown, ignoring it.",
                       :plugin => plugin.class.config_name, :exception => e,
@@ -233,56 +320,6 @@ def inputworker(plugin)
     end
   end # def inputworker
 
-  def filterworker
-    begin
-      while true
-        event = @input_to_filter.pop
-
-        case event
-        when LogStash::Event
-          # filter_func returns all filtered events, including cancelled ones
-          filter_func(event).each { |e| @filter_to_output.push(e) unless e.cancelled? }
-        when LogStash::FlushEvent
-          # handle filter flushing here so that non threadsafe filters (thus only running one filterworker)
-          # don't have to deal with thread safety implementing the flush method
-          flush_filters_to_output!
-        when LogStash::ShutdownEvent
-          # pass it down to any other filterworker and stop this worker
-          @input_to_filter.push(event)
-          break
-        end
-      end
-    rescue Exception => e
-      # Plugins authors should manage their own exceptions in the plugin code
-      # but if an exception is raised up to the worker thread they are considered
-      # fatal and logstash will not recover from this situation.
-      #
-      # Users need to check their configuration or see if there is a bug in the
-      # plugin.
-      @logger.error("Exception in filterworker, the pipeline stopped processing new events, please check your filter configuration and restart Logstash.",
-                    "exception" => e, "backtrace" => e.backtrace)
-      raise
-    ensure
-      @filters.each(&:do_close)
-    end
-  end # def filterworker
-
-  def outputworker
-    LogStash::Util.set_thread_name(">output")
-    @outputs.each(&:worker_setup)
-
-    while true
-      event = @filter_to_output.pop
-      break if event == LogStash::SHUTDOWN
-      output_func(event)
-      LogStash::Util.set_thread_plugin(nil)
-    end
-  ensure
-    @outputs.each do |output|
-      output.worker_plugins.each(&:do_close)
-    end
-  end # def outputworker
-
   # initiate the pipeline shutdown sequence
   # this method is intended to be called from outside the pipeline thread
   # @param before_stop [Proc] code block called before performing stop operation on input plugins
@@ -296,9 +333,30 @@ def shutdown(&before_stop)
 
     before_stop.call if block_given?
 
+    @logger.info "Closing inputs"
     @inputs.each(&:do_stop)
+    @logger.info "Closed inputs"
   end # def shutdown
 
+  # After `shutdown` is called from an external thread this is called from the main thread to
+  # tell the worker threads to stop and then block until they've fully stopped
+  # This also stops all filter and output plugins
+  def shutdown_workers
+    # Each worker thread will receive this exactly once!
+    @worker_threads.each do |t|
+      @logger.debug("Pushing shutdown", :thread => t)
+      @input_queue.push(LogStash::SHUTDOWN)
+    end
+
+    @worker_threads.each do |t|
+      @logger.debug("Shutdown waiting for worker thread #{t}")
+      t.join
+    end
+
+    @filters.each(&:do_close)
+    @outputs.each(&:do_close)
+  end
+
   def plugin(plugin_type, name, *args)
     args << {} if args.empty?
     klass = LogStash::Plugin.lookup(plugin_type, name)
@@ -312,6 +370,7 @@ def filter(event, &block)
     filter_func(event).each { |e| block.call(e) }
   end
 
+
   # perform filters flush and yeild flushed event to the passed block
   # @param options [Hash]
   # @option options [Boolean] :final => true to signal a final shutdown flush
@@ -323,61 +382,17 @@ def flush_filters(options = {}, &block)
     end
   end
 
-  # perform filters flush into the output queue
-  # @param options [Hash]
-  # @option options [Boolean] :final => true to signal a final shutdown flush
-  def flush_filters_to_output!(options = {})
-    flush_filters(options) do |event|
-      unless event.cancelled?
-        @logger.debug? and @logger.debug("Pushing flushed events", :event => event)
-        @filter_to_output.push(event)
-      end
-    end
-  end # flush_filters_to_output!
-
-  def inflight_count
-    data = {}
-    total = 0
-
-    input_to_filter = @input_to_filter.size
-    total += input_to_filter
-    filter_to_output = @filter_to_output.size
-    total += filter_to_output
-
-    data["input_to_filter"] = input_to_filter if input_to_filter > 0
-    data["filter_to_output"] = filter_to_output if filter_to_output > 0
-
-    output_worker_queues = []
-    @outputs.each do |output|
-      next unless output.worker_queue && output.worker_queue.size > 0
-      plugin_info = output.debug_info
-      size = output.worker_queue.size
-      total += size
-      plugin_info << size
-      output_worker_queues << plugin_info
-    end
-    data["output_worker_queues"] = output_worker_queues unless output_worker_queues.empty?
-    data["total"] = total
-    data
-  end
-
-  def stalling_threads
-    plugin_threads
-     .reject {|t| t["blocked_on"] } # known begnin blocking statuses
-     .each {|t| t.delete("backtrace") }
-     .each {|t| t.delete("blocked_on") }
-     .each {|t| t.delete("status") }
-  end
-
-  def plugin_threads
-    input_threads = @input_threads.select {|t| t.alive? }.map {|t| thread_info(t) }
-    filter_threads = @filter_threads.select {|t| t.alive? }.map {|t| thread_info(t) }
-    output_threads = @output_threads.select {|t| t.alive? }.map {|t| thread_info(t) }
-    output_worker_threads = @outputs.flat_map {|output| output.worker_threads }.map {|t| thread_info(t) }
-    input_threads + filter_threads + output_threads + output_worker_threads
+  def plugin_threads_info
+    input_threads = @input_threads.select {|t| t.alive? }
+    worker_threads = @worker_threads.select {|t| t.alive? }
+    (input_threads + worker_threads).map {|t| LogStash::Util.thread_info(t) }
   end
 
-  def thread_info(thread)
-    LogStash::Util.thread_info(thread)
+  def stalling_threads_info
+    plugin_threads_info
+      .reject {|t| t["blocked_on"] } # known benign blocking statuses
+      .each {|t| t.delete("backtrace") }
+      .each {|t| t.delete("blocked_on") }
+      .each {|t| t.delete("status") }
   end
-end; end
+end end
\ No newline at end of file
diff --git a/logstash-core/lib/logstash/pipeline_reporter.rb b/logstash-core/lib/logstash/pipeline_reporter.rb
new file mode 100644
index 00000000000..db36e75916a
--- /dev/null
+++ b/logstash-core/lib/logstash/pipeline_reporter.rb
@@ -0,0 +1,119 @@
+require 'ostruct'
+
+# encoding: utf-8
+module LogStash; class PipelineReporter
+  attr_reader :logger, :pipeline
+
+  # This is an immutable copy of the pipeline state,
+  # It is a proxy to a hash to allow us to add methods dynamically to the hash
+  class Snapshot
+    def initialize(data)
+      @data = data
+    end
+
+    def to_hash
+      @data
+    end
+
+    def to_simple_hash
+      {"inflight_count" => inflight_count, "stalling_thread_info" => format_threads_by_plugin}
+    end
+
+    def method_missing(meth)
+      @data[meth]
+    end
+
+    def format_threads_by_plugin
+      stalled_plugins = {}
+      stalling_threads_info.each do |thr|
+        key = (thr.delete("plugin") || "other")
+        stalled_plugins[key] ||= []
+        stalled_plugins[key] << thr
+      end
+      stalled_plugins
+    end
+  end
+
+  def initialize(logger,pipeline)
+    @logger = logger
+    @pipeline = pipeline
+  end
+
+  def report
+    @logger.warn ["INFLIGHT_EVENTS_REPORT", Time.now.iso8601, report]
+  end
+
+  # The main way of accessing data from the reporter,,
+  # this provides a (more or less) consistent snapshot of what's going on in the
+  # pipeline with some extra decoration
+  def snapshot
+    Snapshot.new(self.to_hash)
+  end
+
+  def to_hash
+    pipeline.inflight_batches_synchronize do |batch_map|
+      worker_states_snap = worker_states(batch_map) # We only want to run this once
+      inflight_count = worker_states_snap.map {|s| s[:inflight_count] }.reduce(:+) || 0
+      {
+        :events_filtered => events_filtered,
+        :events_consumed => events_consumed,
+        :worker_count => pipeline.worker_threads.size,
+        :inflight_count => inflight_count,
+        :worker_states => worker_states_snap,
+        :output_info => output_states,
+        :thread_info => pipeline.plugin_threads_info,
+        :stalling_threads_info => pipeline.stalling_threads_info
+      }
+    end
+  end
+
+  private
+
+  def events_filtered
+    pipeline.events_filtered.value
+  end
+
+  def events_consumed
+    pipeline.events_consumed.value
+  end
+
+  def plugin_threads
+    pipeline.plugin_threads
+  end
+
+  # Not threadsafe! must be called within an `inflight_batches_synchronize` block
+  def worker_states(batch_map)
+      pipeline.worker_threads.map.with_index do |thread,idx|
+        status = thread.status || "dead"
+        inflight_count = batch_map[thread] ? batch_map[thread].size : 0
+        {
+          :status => status,
+          :alive => thread.alive?,
+          :index => idx,
+          :inflight_count => inflight_count
+        }
+    end
+  end
+
+  def output_states
+    pipeline.outputs.map do |output|
+      is_multi_worker = output.workers > 1
+
+      idle, busy = if is_multi_worker
+                     aw_size = output.available_workers.size
+                     [aw_size, output.workers - aw_size]
+                   else
+                     output.single_worker_mutex.locked? ? [0,1] : [1,0]
+                   end
+
+      {
+        :type => output.class.config_name,
+        :config => output.config,
+        :is_multi_worker => is_multi_worker,
+        :workers => output.workers,
+        :busy_workers => busy,
+        :idle_workers => idle
+      }
+    end
+  end
+end end
\ No newline at end of file
diff --git a/logstash-core/lib/logstash/runner.rb b/logstash-core/lib/logstash/runner.rb
index ee9c0dceb61..6038f3c9422 100644
--- a/logstash-core/lib/logstash/runner.rb
+++ b/logstash-core/lib/logstash/runner.rb
@@ -26,9 +26,9 @@ class LogStash::Runner < Clamp::Command
            :default_input => DEFAULT_INPUT, :default_output => DEFAULT_OUTPUT),
     :default => "", :attribute_name => :config_string
 
-  option ["-w", "--filterworkers"], "COUNT",
-    I18n.t("logstash.runner.flag.filterworkers"),
-    :attribute_name => :filter_workers,
+  option ["-w", "--pipelineworkers"], "COUNT",
+    I18n.t("logstash.runner.flag.pipelineworkers"),
+    :attribute_name => :pipeline_workers,
     :default => LogStash::Config::CpuCoreStrategy.fifty_percent, &:to_i
 
   option ["-l", "--log"], "FILE",
@@ -110,7 +110,7 @@ def execute
 
     config_string = format_config(@config_path, @config_string)
 
-    @agent.add_pipeline("base", config_string, :filter_workers => filter_workers)
+    @agent.add_pipeline("base", config_string, :pipeline_workers => pipeline_workers)
 
     if config_test?
       puts "Configuration OK"
diff --git a/logstash-core/lib/logstash/shutdown_controller.rb b/logstash-core/lib/logstash/shutdown_controller.rb
index 6941753bbc8..fc368da51fe 100644
--- a/logstash-core/lib/logstash/shutdown_controller.rb
+++ b/logstash-core/lib/logstash/shutdown_controller.rb
@@ -47,10 +47,10 @@ def start
       cycle_number = 0
       stalled_count = 0
       Stud.interval(@cycle_period) do
-        @reports << Report.from_pipeline(@pipeline)
+        @reports << pipeline_report_snapshot
         @reports.delete_at(0) if @reports.size > @report_every # expire old report
         if cycle_number == (@report_every - 1) # it's report time!
-          logger.warn(@reports.last.to_hash)
+          logger.warn(@reports.last.to_simple_hash)
 
           if shutdown_stalled?
             logger.error("The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information.") if stalled_count == 0
@@ -69,6 +69,10 @@ def start
       end
     end
 
+    def pipeline_report_snapshot
+      @pipeline.reporter.snapshot
+    end
+
     # A pipeline shutdown is stalled if
     # * at least REPORT_EVERY reports have been created
     # * the inflight event count is in monotonically increasing
@@ -78,7 +82,7 @@ def shutdown_stalled?
       return false unless @reports.size == @report_every #
       # is stalled if inflight count is either constant or increasing
       stalled_event_count = @reports.each_cons(2).all? do |prev_report, next_report|
-        prev_report.inflight_count["total"] <= next_report.inflight_count["total"]
+        prev_report.inflight_count <= next_report.inflight_count
       end
       if stalled_event_count
         @reports.each_cons(2).all? do |prev_report, next_report|
@@ -93,35 +97,4 @@ def force_exit
       exit(-1)
     end
   end
-
-  class Report
-
-    attr_reader :inflight_count, :stalling_threads
-
-    def self.from_pipeline(pipeline)
-      new(pipeline.inflight_count, pipeline.stalling_threads)
-    end
-
-    def initialize(inflight_count, stalling_threads)
-      @inflight_count = inflight_count
-      @stalling_threads = format_threads_by_plugin(stalling_threads)
-    end
-
-    def to_hash
-      {
-        "INFLIGHT_EVENT_COUNT" => @inflight_count,
-        "STALLING_THREADS" => @stalling_threads
-      }
-    end
-
-    def format_threads_by_plugin(stalling_threads)
-      stalled_plugins = {}
-      stalling_threads.each do |thr|
-        key = (thr.delete("plugin") || "other")
-        stalled_plugins[key] ||= []
-        stalled_plugins[key] << thr
-      end
-      stalled_plugins
-    end
-  end
 end
diff --git a/logstash-core/lib/logstash/util/worker_threads_default_printer.rb b/logstash-core/lib/logstash/util/worker_threads_default_printer.rb
index 82e88196e96..a4628552f7b 100644
--- a/logstash-core/lib/logstash/util/worker_threads_default_printer.rb
+++ b/logstash-core/lib/logstash/util/worker_threads_default_printer.rb
@@ -6,8 +6,8 @@
 module LogStash module Util class WorkerThreadsDefaultPrinter
 
   def initialize(settings)
-    @setting = settings.fetch('filter-workers', 0)
-    @default = settings.fetch('default-filter-workers', 0)
+    @setting = settings.fetch('pipeline-workers', 0)
+    @default = settings.fetch('default-pipeline-workers', 0)
   end
 
   def visit(collector)
@@ -17,12 +17,12 @@ def visit(collector)
 
   def visit_setting(collector)
     return if @setting == 0
-    collector.push("User set filter workers: #{@setting}")
+    collector.push("User set pipeline workers: #{@setting}")
   end
 
   def visit_default(collector)
     return if @default == 0
-    collector.push "Default filter workers: #{@default}"
+    collector.push "Default pipeline workers: #{@default}"
   end
 
 end end end
diff --git a/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb b/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb
new file mode 100644
index 00000000000..05c5fc30342
--- /dev/null
+++ b/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb
@@ -0,0 +1,25 @@
+module LogStash; module Util
+  class WrappedSynchronousQueue
+    java_import java.util.concurrent.SynchronousQueue
+    java_import java.util.concurrent.TimeUnit
+
+    def initialize()
+      @queue = java.util.concurrent.SynchronousQueue.new()
+    end
+
+    def push(obj)
+      @queue.put(obj)
+    end
+    alias_method(:<<, :push)
+
+    # Blocking
+    def take
+      @queue.take()
+    end
+
+    # Block for X millis
+    def poll(millis)
+      @queue.poll(millis, TimeUnit::MILLISECONDS)
+    end
+  end
+end end
\ No newline at end of file
diff --git a/logstash-core/locales/en.yml b/logstash-core/locales/en.yml
index 1b2ce1dd51c..e60e0b40a84 100644
--- a/logstash-core/locales/en.yml
+++ b/logstash-core/locales/en.yml
@@ -164,8 +164,8 @@ en:
           the empty string for the '-e' flag.
         configtest: |+
           Check configuration for valid syntax and then exit.
-        filterworkers: |+
-          Sets the number of filter workers to run.
+        pipelineworkers: |+
+          Sets the number of pipeline workers to run.
         log: |+
           Write logstash internal logs to the given
           file. Without this flag, logstash will emit
diff --git a/logstash-core/spec/logstash/pipeline_reporter_spec.rb b/logstash-core/spec/logstash/pipeline_reporter_spec.rb
new file mode 100644
index 00000000000..ec2f07d8912
--- /dev/null
+++ b/logstash-core/spec/logstash/pipeline_reporter_spec.rb
@@ -0,0 +1,51 @@
+# encoding: utf-8
+require "spec_helper"
+require "logstash/pipeline"
+require "logstash/pipeline_reporter"
+
+#TODO: Figure out how to add more tests that actually cover inflight events
+#This will require some janky multithreading stuff
+describe LogStash::PipelineReporter do
+  let(:generator_count) { 5 }
+  let(:config) do
+    "input { generator { count => #{generator_count} } } output { null {} } "
+  end
+  let(:pipeline) { LogStash::Pipeline.new(config)}
+  let(:reporter) { pipeline.reporter }
+
+  before do
+    @pre_snapshot = reporter.snapshot
+    pipeline.run
+    @post_snapshot = reporter.snapshot
+  end
+
+  describe "events filtered" do
+    it "should start at zero" do
+      expect(@pre_snapshot.events_filtered).to eql(0)
+    end
+
+    it "should end at the number of generated events" do
+      expect(@post_snapshot.events_filtered).to eql(generator_count)
+    end
+  end
+
+  describe "events consumed" do
+    it "should start at zero" do
+      expect(@pre_snapshot.events_consumed).to eql(0)
+    end
+
+    it "should end at the number of generated events" do
+      expect(@post_snapshot.events_consumed).to eql(generator_count)
+    end
+  end
+
+  describe "inflight count" do
+    it "should be zero before running" do
+      expect(@pre_snapshot.inflight_count).to eql(0)
+    end
+
+    it "should be zero after running" do
+      expect(@post_snapshot.inflight_count).to eql(0)
+    end
+  end
+end
\ No newline at end of file
diff --git a/logstash-core/spec/logstash/pipeline_spec.rb b/logstash-core/spec/logstash/pipeline_spec.rb
index 15b32cb266b..f1fd48f0872 100644
--- a/logstash-core/spec/logstash/pipeline_spec.rb
+++ b/logstash-core/spec/logstash/pipeline_spec.rb
@@ -80,7 +80,7 @@ def close() end
 end
 
 class TestPipeline < LogStash::Pipeline
-  attr_reader :outputs, :filter_threads, :settings, :logger
+  attr_reader :outputs, :settings, :logger
 end
 
 describe LogStash::Pipeline do
@@ -88,7 +88,7 @@ class TestPipeline < LogStash::Pipeline
   let(:safe_thread_count)       { 1 }
   let(:override_thread_count)   { 42 }
 
-  describe "defaulting the filter workers based on thread safety" do
+  describe "defaulting the pipeline workers based on thread safety" do
     before(:each) do
       allow(LogStash::Plugin).to receive(:lookup).with("input", "dummyinput").and_return(DummyInput)
       allow(LogStash::Plugin).to receive(:lookup).with("codec", "plain").and_return(DummyCodec)
@@ -117,13 +117,13 @@ class TestPipeline < LogStash::Pipeline
 
       context "when there is no command line -w N set" do
         it "starts one filter thread" do
-          msg = "Defaulting filter worker threads to 1 because there are some" +
+          msg = "Defaulting pipeline worker threads to 1 because there are some" +
                 " filters that might not work with multiple worker threads"
           pipeline = TestPipeline.new(test_config_with_filters)
           expect(pipeline.logger).to receive(:warn).with(msg,
             {:count_was=>worker_thread_count, :filters=>["dummyfilter"]})
           pipeline.run
-          expect(pipeline.filter_threads.size).to eq(safe_thread_count)
+          expect(pipeline.worker_threads.size).to eq(safe_thread_count)
         end
       end
 
@@ -134,9 +134,9 @@ class TestPipeline < LogStash::Pipeline
           pipeline = TestPipeline.new(test_config_with_filters)
           expect(pipeline.logger).to receive(:warn).with(msg,
             {:worker_threads=> override_thread_count, :filters=>["dummyfilter"]})
-          pipeline.configure("filter-workers", override_thread_count)
+          pipeline.configure("pipeline-workers", override_thread_count)
           pipeline.run
-          expect(pipeline.filter_threads.size).to eq(override_thread_count)
+          expect(pipeline.worker_threads.size).to eq(override_thread_count)
         end
       end
     end
@@ -161,7 +161,7 @@ class TestPipeline < LogStash::Pipeline
       it "starts multiple filter threads" do
         pipeline = TestPipeline.new(test_config_with_filters)
         pipeline.run
-        expect(pipeline.filter_threads.size).to eq(worker_thread_count)
+        expect(pipeline.worker_threads.size).to eq(worker_thread_count)
       end
     end
   end
@@ -215,7 +215,8 @@ class TestPipeline < LogStash::Pipeline
         pipeline.run
 
         expect(pipeline.outputs.size ).to eq(1)
-        expect(pipeline.outputs.first.num_closes).to eq(0)
+        # We even close the parent output worker, even though it doesn't receive messages
+        expect(pipeline.outputs.first.num_closes).to eq(1)
         pipeline.outputs.first.worker_plugins.each do |plugin|
           expect(plugin.num_closes ).to eq(1)
         end
diff --git a/logstash-core/spec/logstash/shutdown_controller_spec.rb b/logstash-core/spec/logstash/shutdown_controller_spec.rb
index 5f755f290a8..96d05f61709 100644
--- a/logstash-core/spec/logstash/shutdown_controller_spec.rb
+++ b/logstash-core/spec/logstash/shutdown_controller_spec.rb
@@ -8,10 +8,16 @@
   let(:check_threshold) { 100 }
   subject { LogStash::ShutdownController.new(pipeline, check_every) }
   let(:pipeline) { double("pipeline") }
+  let(:reporter) { double("reporter") }
+  let(:reporter_snapshot) { double("reporter snapshot") }
   report_count = 0
 
   before :each do
-    allow(LogStash::Report).to receive(:from_pipeline).and_wrap_original do |m, *args|
+    allow(pipeline).to receive(:reporter).and_return(reporter)
+    allow(reporter).to receive(:snapshot).and_return(reporter_snapshot)
+    allow(reporter_snapshot).to receive(:to_simple_hash).and_return({})
+
+    allow(subject).to receive(:pipeline_report_snapshot).and_wrap_original do |m, *args|
       report_count += 1
       m.call(*args)
     end
@@ -22,10 +28,10 @@
   end
 
   context "when pipeline is stalled" do
-    let(:increasing_count) { (1..5000).to_a.map {|i| { "total" => i } } }
+    let(:increasing_count) { (1..5000).to_a }
     before :each do
-      allow(pipeline).to receive(:inflight_count).and_return(*increasing_count)
-      allow(pipeline).to receive(:stalling_threads) { { } }
+      allow(reporter_snapshot).to receive(:inflight_count).and_return(*increasing_count)
+      allow(reporter_snapshot).to receive(:stalling_threads) { { } }
     end
 
     describe ".unsafe_shutdown = true" do
@@ -49,7 +55,7 @@
 
       it "should do exactly \"abort_threshold\"*\"report_every\" stall checks" do
         allow(subject).to receive(:force_exit)
-        expect(LogStash::Report).to receive(:from_pipeline).exactly(abort_threshold*report_every).times.and_call_original
+        expect(subject).to receive(:pipeline_report_snapshot).exactly(abort_threshold*report_every).times.and_call_original
         subject.start
       end
     end
@@ -70,10 +76,10 @@
   end
 
   context "when pipeline is not stalled" do
-    let(:decreasing_count) { (1..5000).to_a.reverse.map {|i| { "total" => i } } }
+    let(:decreasing_count) { (1..5000).to_a.reverse }
     before :each do
-      allow(pipeline).to receive(:inflight_count).and_return(*decreasing_count)
-      allow(pipeline).to receive(:stalling_threads) { { } }
+      allow(reporter_snapshot).to receive(:inflight_count).and_return(*decreasing_count)
+      allow(reporter_snapshot).to receive(:stalling_threads) { { } }
     end
 
     describe ".unsafe_shutdown = true" do
diff --git a/logstash-core/spec/logstash/util/defaults_printer_spec.rb b/logstash-core/spec/logstash/util/defaults_printer_spec.rb
index ed47cf7ca50..a2eda701425 100644
--- a/logstash-core/spec/logstash/util/defaults_printer_spec.rb
+++ b/logstash-core/spec/logstash/util/defaults_printer_spec.rb
@@ -10,7 +10,7 @@
   end
 
   let(:workers)  { 1 }
-  let(:expected) { "Settings: User set filter workers: #{workers}" }
+  let(:expected) { "Settings: User set pipeline workers: #{workers}" }
   let(:settings) { {} }
 
   describe 'class methods API' do
@@ -25,7 +25,7 @@
 
     context 'when the settings hash has content' do
       let(:workers) { 42 }
-      let(:settings) { {'filter-workers' => workers} }
+      let(:settings) { {'pipeline-workers' => workers} }
       it_behaves_like "a defaults printer"
     end
   end
@@ -42,7 +42,7 @@
 
     context 'when the settings hash has content' do
       let(:workers) { 13 }
-      let(:settings) { {'filter-workers' => workers} }
+      let(:settings) { {'pipeline-workers' => workers} }
 
       it_behaves_like "a defaults printer"
     end
diff --git a/logstash-core/spec/logstash/util/worker_threads_default_printer_spec.rb b/logstash-core/spec/logstash/util/worker_threads_default_printer_spec.rb
index c2f5391cf38..410d8c9fbf7 100644
--- a/logstash-core/spec/logstash/util/worker_threads_default_printer_spec.rb
+++ b/logstash-core/spec/logstash/util/worker_threads_default_printer_spec.rb
@@ -19,26 +19,26 @@
     end
 
     context 'when the settings hash has both user and default content' do
-      let(:settings) { {'filter-workers' => 42, 'default-filter-workers' => 5} }
+      let(:settings) { {'pipeline-workers' => 42, 'default-pipeline-workers' => 5} }
 
       it 'adds two strings' do
-        expect(collector).to eq(["User set filter workers: 42", "Default filter workers: 5"])
+        expect(collector).to eq(["User set pipeline workers: 42", "Default pipeline workers: 5"])
       end
     end
 
     context 'when the settings hash has only user content' do
-      let(:settings) { {'filter-workers' => 42} }
+      let(:settings) { {'pipeline-workers' => 42} }
 
-      it 'adds a string with user set filter workers' do
-        expect(collector.first).to eq("User set filter workers: 42")
+      it 'adds a string with user set pipeline workers' do
+        expect(collector.first).to eq("User set pipeline workers: 42")
       end
     end
 
     context 'when the settings hash has only default content' do
-      let(:settings) { {'default-filter-workers' => 5} }
+      let(:settings) { {'default-pipeline-workers' => 5} }
 
-      it 'adds a string with default filter workers' do
-        expect(collector.first).to eq("Default filter workers: 5")
+      it 'adds a string with default pipeline workers' do
+        expect(collector.first).to eq("Default pipeline workers: 5")
       end
     end
   end