Use concurrency :shared and sync codecs to optimize performance

This provides a nice boost:

Before:

```
time bin/logstash -e "input { generator { count => 3000000} } filter {  } output { file { path => '/tmp/newfileout'} }"
Settings: Default pipeline workers: 8
Pipeline main started
Pipeline main has been shutdown
stopping pipeline {:id=>"main"}
      139.95 real       223.61 user        28.93 sys
```

After

```
rm /tmp/newfileout; time bin/logstash -e "input { generator { count => 3000000} } filter {  } output { file { codec => json_lines path => '/tmp/newfileout'} }" ; ls -lh /tmp/newfileout
Settings: Default pipeline workers: 8
Pipeline main started
Pipeline main has been shutdown
stopping pipeline {:id=>"main"}
       56.12 real       192.99 user        17.38 sys
```

Author: andrewvc

lib/logstash/outputs/file.rb

@@ -3,6 +3,8 @@
 require "logstash/outputs/base"
 require "logstash/errors"
 require "zlib"
+java_import java.util.concurrent.locks.ReentrantReadWriteLock
+java_import java.lang.StringBuilder
 
 # This output writes events to files on disk. You can use fields
 # from the event as parts of the filename and/or path.
@@ -17,6 +19,8 @@
 #  }
 # }
 class LogStash::Outputs::File < LogStash::Outputs::Base
+  concurrency :shared
+  
   FIELD_REF = /%\{[^}]+\}/
 
   config_name "file"
@@ -76,9 +80,10 @@ class LogStash::Outputs::File < LogStash::Outputs::Base
   def register
     require "fileutils" # For mkdir_p
 
-    workers_not_supported
-
     @files = {}
+    @files_lock = java.util.concurrent.locks.ReentrantReadWriteLock.new
+    @files_read_lock = @files_lock.readLock
+    @files_write_lock = @files_lock.writeLock
 
     @path = File.expand_path(path)
 
@@ -91,6 +96,7 @@ def register
     end
     @failure_path = File.join(@file_root, @filename_failure)
 
+
     now = Time.now
     @last_flush_cycle = now
     @last_stale_cleanup_cycle = now
@@ -101,8 +107,6 @@ def register
      @codec = LogStash::Plugin.lookup("codec", "line").new
      @codec.format = @message_format
     end
-
-    @codec.on_event(&method(:write_event))
   end # def register
 
   private
@@ -125,20 +129,42 @@ def root_directory
   end
 
   public
-  def receive(event)
-    @codec.encode(event)
+  def multi_receive_encoded(events_and_encoded)
+    encoded_by_path = Hash.new {|h,k| h[k] = []}
+    
+    events_and_encoded.each do |event,encoded|
+      file_output_path = write_event(event)
+      encoded_by_path[file_output_path] << encoded
+    end
+
+    encoded_by_path.each do |path,chunks|
+      fd = open(path)
+      chunks.each(&fd.method(:write))
+      flush(fd)
+    end
+    
     close_stale_files
   end # def receive
 
+  def with_lock(lock)
+    lock.lock
+    yield
+  ensure
+    lock.unlock
+  end
+
   public
   def close
     @logger.debug("Close: closing files")
-    @files.each do |path, fd|
-      begin
-        fd.close
-        @logger.debug("Closed file #{path}", :fd => fd)
-      rescue Exception => e
-        @logger.error("Exception while flushing and closing files.", :exception => e)
+    
+    with_lock(@files_read_lock) do
+      @files.each do |path, fd|
+        begin
+          fd.close
+          @logger.debug("Closed file #{path}", :fd => fd)
+        rescue Exception => e
+          @logger.error("Exception while flushing and closing files.", :exception => e)
+        end
       end
     end
   end
@@ -150,7 +176,7 @@ def inside_file_root?(log_path)
   end
 
   private
-  def write_event(event, data)
+  def write_event(event)
     file_output_path = generate_filepath(event)
     if path_with_field_ref? && !inside_file_root?(file_output_path)
       @logger.warn("File: the event tried to write outside the files root, writing the event to the failure file",  :event => event, :filename => @failure_path)
@@ -159,10 +185,9 @@ def write_event(event, data)
       file_output_path = @failure_path
     end
     @logger.debug("File, writing event to file.", :filename => file_output_path)
-    fd = open(file_output_path)
-    # TODO(sissel): Check if we should rotate the file.
-    fd.write(data)
-    flush(fd)
+
+    
+    file_output_path    
   end
 
   private
@@ -195,10 +220,14 @@ def flush(fd)
   def flush_pending_files
     return unless Time.now - @last_flush_cycle >= flush_interval
     @logger.debug("Starting flush cycle")
-    @files.each do |path, fd|
-      @logger.debug("Flushing file", :path => path, :fd => fd)
-      fd.flush
+
+    with_lock(@files_read_lock) do
+      @files.each do |path, fd|
+        @logger.debug("Flushing file", :path => path, :fd => fd)
+        fd.flush
+      end
     end
+    
     @last_flush_cycle = Time.now
   end
 
@@ -207,22 +236,25 @@ def flush_pending_files
   def close_stale_files
     now = Time.now
     return unless now - @last_stale_cleanup_cycle >= @stale_cleanup_interval
-    @logger.info("Starting stale files cleanup cycle", :files => @files)
-    inactive_files = @files.select { |path, fd| not fd.active }
-    @logger.debug("%d stale files found" % inactive_files.count, :inactive_files => inactive_files)
-    inactive_files.each do |path, fd|
-      @logger.info("Closing file %s" % path)
-      fd.close
-      @files.delete(path)
+
+    with_lock(@files_write_lock) do
+      @logger.info("Starting stale files cleanup cycle", :files => @files)
+      inactive_files = @files.select { |path, fd| not fd.active }
+      @logger.debug("%d stale files found" % inactive_files.count, :inactive_files => inactive_files)
+      inactive_files.each do |path, fd|
+        @logger.info("Closing file %s" % path)
+        fd.close
+        @files.delete(path)
+      end
+      # mark all files as inactive, a call to write will mark them as active again
+      @files.each { |path, fd| fd.active = false }
+      @last_stale_cleanup_cycle = now
     end
-    # mark all files as inactive, a call to write will mark them as active again
-    @files.each { |path, fd| fd.active = false }
-    @last_stale_cleanup_cycle = now
   end
 
   private
   def cached?(path)
-     @files.include?(path) && !@files[path].nil?
+    with_lock(@files_read_lock) { @files.include?(path) && !@files[path].nil? }
   end
 
   private
@@ -233,13 +265,15 @@ def deleted?(path)
   private
   def open(path)
     if !deleted?(path) && cached?(path)
-      return @files[path]
+      with_lock(@files_read_lock) { return @files[path] }
     elsif deleted?(path)
-      if @create_if_deleted
-        @logger.debug("Required path was deleted, creating the file again", :path => path)
-        @files.delete(path)
-      else
-        return @files[path] if cached?(path)
+      with_lock(@files_write_lock) do
+        if @create_if_deleted
+          @logger.debug("Required path was deleted, creating the file again", :path => path)
+          @files.delete(path)
+        else
+          return @files[path] if cached?(path)
+        end
       end
     end
     @logger.info("Opening file", :path => path)
@@ -267,18 +301,21 @@ def open(path)
     if gzip
       fd = Zlib::GzipWriter.new(fd)
     end
-    @files[path] = IOWriter.new(fd)
+    with_lock(@files_write_lock) { @files[path] = IOWriter.new(fd) }
   end
 end # class LogStash::Outputs::File
 
 # wrapper class
 class IOWriter
   def initialize(io)
     @io = io
+    @write_mutex = Mutex.new
   end
-  def write(*args)
-    @io.write(*args)
-    @active = true
+  def write(string)
+    @write_mutex.synchronize do
+      @io.write(string)
+      @active = true
+    end
   end
   def flush
     @io.flush