Track threads and allocated values

Track threads and allocated values.
Dispose value when a thread dies.
Dispose allocated values on close()

Author: brenuart

src/main/java/net/logstash/logback/composite/AbstractCompositeJsonFormatter.java

@@ -25,8 +25,8 @@
 import net.logstash.logback.decorate.JsonGeneratorDecorator;
 import net.logstash.logback.decorate.NullJsonFactoryDecorator;
 import net.logstash.logback.decorate.NullJsonGeneratorDecorator;
-import net.logstash.logback.util.ObjectPool;
 import net.logstash.logback.util.ProxyOutputStream;
+import net.logstash.logback.util.ThreadLocalHolder;
 
 import ch.qos.logback.access.spi.IAccessEvent;
 import ch.qos.logback.classic.spi.ILoggingEvent;
@@ -93,7 +93,7 @@ public abstract class AbstractCompositeJsonFormatter<Event extends DeferredProce
 
     private volatile boolean started;
 
-    private ObjectPool<JsonFormatter> pool;
+    private ThreadLocalHolder<JsonFormatter> threadLocalJsonFormatter;
 
 
     public AbstractCompositeJsonFormatter(ContextAware declaredOrigin) {
@@ -119,14 +119,14 @@ public void start() {
         jsonProviders.setJsonFactory(jsonFactory);
         jsonProviders.start();
 
-        pool = new ObjectPool<>(this::createJsonFormatter);
+        threadLocalJsonFormatter = new ThreadLocalHolder<>(this::createJsonFormatter);
         started = true;
     }
 
     @Override
     public void stop() {
         if (isStarted()) {
-            pool.clear();
+            threadLocalJsonFormatter.close();
             jsonProviders.stop();
             jsonFactory = null;
             started = false;
@@ -152,7 +152,7 @@ public void writeEvent(Event event, OutputStream outputStream) throws IOExceptio
             throw new IllegalStateException("Formatter is not started");
         }
 
-        try (JsonFormatter formatter = this.pool.acquire()) {
+        try (JsonFormatter formatter = this.threadLocalJsonFormatter.acquire()) {
             formatter.writeEvent(outputStream, event);
         }
     }
@@ -174,7 +174,7 @@ private JsonFormatter createJsonFormatter() {
 
     }
 
-    private class JsonFormatter implements ObjectPool.Lifecycle, Closeable {
+    private class JsonFormatter implements ThreadLocalHolder.Lifecycle, Closeable {
         private final JsonGenerator generator;
         private final DisconnectedOutputStream stream;
         private boolean recyclable = true;
@@ -206,11 +206,18 @@ public boolean recycle() {
         @Override
         public void dispose() {
             CloseUtil.closeQuietly(this.generator);
+            
+            // Note:
+            //   The stream is disconnected at this point.
+            //   Closing the JsonGenerator may throw additional exception if it is flagged as not recyclable,
+            //   meaning it already threw a exception earlier during the writeEvent() method. The generator
+            //   is disposed here and won't be reused anymore - we can safely ignore these new exceptions
+            //   here.
         }
 
         @Override
         public void close() throws IOException {
-            AbstractCompositeJsonFormatter.this.pool.release(this);
+            AbstractCompositeJsonFormatter.this.threadLocalJsonFormatter.release();
         }
     }
 

src/main/java/net/logstash/logback/composite/JsonProviders.java

@@ -41,8 +41,8 @@
  */
 public class JsonProviders<Event extends DeferredProcessingAware> implements JsonFactoryAware {
 
-    private final List<JsonProvider<Event>> jsonProviders = new ArrayList<>();
-
+    private final ArrayList<JsonProvider<Event>> jsonProviders = new ArrayList<>();
+    
     public void start() {
         for (JsonProvider<Event> jsonProvider : jsonProviders) {
             jsonProvider.start();

src/main/java/net/logstash/logback/encoder/CompositeJsonEncoder.java

@@ -25,7 +25,7 @@
 import net.logstash.logback.decorate.JsonFactoryDecorator;
 import net.logstash.logback.decorate.JsonGeneratorDecorator;
 import net.logstash.logback.util.ReusableByteBuffer;
-import net.logstash.logback.util.ReusableByteBufferPool;
+import net.logstash.logback.util.ThreadLocalReusableByteBuffer;
 
 import ch.qos.logback.core.encoder.Encoder;
 import ch.qos.logback.core.encoder.EncoderBase;
@@ -49,9 +49,9 @@ public abstract class CompositeJsonEncoder<Event extends DeferredProcessingAware
     private int minBufferSize = 1024;
 
     /**
-     * Pool of reusable byte buffers used when calling {@link #encode(DeferredProcessingAware)}
+     * Per-thread {@link ReusableByteBuffer} instance used when calling {@link #encode(DeferredProcessingAware)}
      */
-    private ReusableByteBufferPool bufferPool;
+    private ThreadLocalReusableByteBuffer threadLocalBuffer;
 
     private Encoder<Event> prefix;
     private Encoder<Event> suffix;
@@ -86,7 +86,7 @@ public byte[] encode(Event event) {
             throw new IllegalStateException("Encoder is not started");
         }
 
-        ReusableByteBuffer buffer = bufferPool.acquire();
+        ReusableByteBuffer buffer = threadLocalBuffer.acquire();
 
         try {
             encode(buffer, event);
@@ -97,7 +97,7 @@ public byte[] encode(Event event) {
             return EMPTY_BYTES;
 
         } finally {
-            bufferPool.release(buffer);
+            threadLocalBuffer.release();
         }
     }
 
@@ -134,7 +134,7 @@ public void start() {
         startWrapped(prefix);
         startWrapped(suffix);
 
-        this.bufferPool = ReusableByteBufferPool.create(minBufferSize);
+        this.threadLocalBuffer = new ThreadLocalReusableByteBuffer(minBufferSize);
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })
@@ -178,7 +178,7 @@ public void stop() {
             stopWrapped(prefix);
             stopWrapped(suffix);
 
-            bufferPool = null;
+            threadLocalBuffer = null;
         }
     }
 

src/main/java/net/logstash/logback/layout/CompositeJsonLayout.java

@@ -28,7 +28,7 @@
 import net.logstash.logback.encoder.CompositeJsonEncoder;
 import net.logstash.logback.encoder.SeparatorParser;
 import net.logstash.logback.util.ReusableByteBuffer;
-import net.logstash.logback.util.ReusableByteBufferPool;
+import net.logstash.logback.util.ThreadLocalReusableByteBuffer;
 
 import ch.qos.logback.core.Layout;
 import ch.qos.logback.core.LayoutBase;
@@ -64,9 +64,9 @@ public abstract class CompositeJsonLayout<Event extends DeferredProcessingAware>
     private int minBufferSize = 1024;
 
     /**
-     * Pool of reusable byte buffers
+     * Per-thread {@link ReusableByteBuffer}
      */
-    private ReusableByteBufferPool bufferPool;
+    private ThreadLocalReusableByteBuffer threadLocalBuffer;
 
     private final AbstractCompositeJsonFormatter<Event> formatter;
 
@@ -83,7 +83,7 @@ public String doLayout(Event event) {
             throw new IllegalStateException("Layout is not started");
         }
 
-        ReusableByteBuffer buffer = bufferPool.acquire();
+        ReusableByteBuffer buffer = threadLocalBuffer.acquire();
         try {
             writeEvent(buffer, event);
             return new String(buffer.toByteArray());
@@ -93,7 +93,7 @@ public String doLayout(Event event) {
             return null;
 
         } finally {
-            bufferPool.release(buffer);
+            threadLocalBuffer.release();
         }
     }
 
@@ -136,7 +136,7 @@ public void start() {
         startWrapped(prefix);
         startWrapped(suffix);
 
-        this.bufferPool = ReusableByteBufferPool.create(minBufferSize);
+        this.threadLocalBuffer = new ThreadLocalReusableByteBuffer(minBufferSize);
     }
 
     private void startWrapped(Layout<Event> wrapped) {
@@ -170,7 +170,7 @@ public void stop() {
         stopWrapped(prefix);
         stopWrapped(suffix);
 
-        this.bufferPool = null;
+        this.threadLocalBuffer = null;
     }
 
     private void stopWrapped(Layout<Event> wrapped) {