487197 - Deflater/Inflater memory leak with WebSocket permessage-defl…

…ate extension + CompressExtension implementations are now part of the Jetty LifeCycle + Deflater and Inflater implementations are only instantiated when needed. + CompressExtension.doStop() LifeCycle will call .end() on instantiated Deflater and Inflater implementations
jetty · Feb 4, 2016 · cfe823a · cfe823a
1 parent 980ab31
commit cfe823a
Show file tree

Hide file tree

Showing 3 changed files with 42 additions and 9 deletions.
diff --git a/...common/src/main/java/org/eclipse/jetty/websocket/common/extensions/AbstractExtension.java b/...common/src/main/java/org/eclipse/jetty/websocket/common/extensions/AbstractExtension.java
@@ -23,7 +23,9 @@
 import org.eclipse.jetty.io.ByteBufferPool;
 import org.eclipse.jetty.util.annotation.ManagedAttribute;
 import org.eclipse.jetty.util.annotation.ManagedObject;
+import org.eclipse.jetty.util.component.AbstractLifeCycle;
 import org.eclipse.jetty.util.component.ContainerLifeCycle;
+import org.eclipse.jetty.util.component.Dumpable;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.api.BatchMode;
@@ -38,7 +40,7 @@
 import org.eclipse.jetty.websocket.common.scopes.WebSocketContainerScope;
 
 @ManagedObject("Abstract Extension")
-public abstract class AbstractExtension extends ContainerLifeCycle implements Extension
+public abstract class AbstractExtension extends AbstractLifeCycle implements Dumpable, Extension
 {
     private final Logger log;
     private WebSocketPolicy policy;
@@ -52,11 +54,15 @@ public AbstractExtension()
     {
         log = Log.getLogger(this.getClass());
     }
-
+    
     @Override
+    public String dump()
+    {
+        return ContainerLifeCycle.dump(this);
+    }
+
     public void dump(Appendable out, String indent) throws IOException
     {
-        super.dump(out, indent);
         // incoming
         dumpWithHeading(out, indent, "incoming", this.nextIncoming);
         dumpWithHeading(out, indent, "outgoing", this.nextOutgoing);

diff --git a/...et-common/src/main/java/org/eclipse/jetty/websocket/common/extensions/ExtensionStack.java b/...et-common/src/main/java/org/eclipse/jetty/websocket/common/extensions/ExtensionStack.java
@@ -29,6 +29,7 @@
 import org.eclipse.jetty.util.annotation.ManagedAttribute;
 import org.eclipse.jetty.util.annotation.ManagedObject;
 import org.eclipse.jetty.util.component.ContainerLifeCycle;
+import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.api.BatchMode;
@@ -89,6 +90,11 @@ protected void doStart() throws Exception
                 Extension ext = exts.next();
                 ext.setNextOutgoingFrames(nextOutgoing);
                 nextOutgoing = ext;
+
+                if (ext instanceof LifeCycle)
+                {
+                    addBean(ext,true);
+                }
             }
 
             // Connect incomings

diff --git a/...c/main/java/org/eclipse/jetty/websocket/common/extensions/compress/CompressExtension.java b/...c/main/java/org/eclipse/jetty/websocket/common/extensions/compress/CompressExtension.java
@@ -74,28 +74,34 @@ public abstract class CompressExtension extends AbstractExtension
 
     private final Queue<FrameEntry> entries = new ConcurrentArrayQueue<>();
     private final IteratingCallback flusher = new Flusher();
-    private final Deflater deflater;
-    private final Inflater inflater;
+    private Deflater deflaterImpl;
+    private Inflater inflaterImpl;
     protected AtomicInteger decompressCount = new AtomicInteger(0);
     private int tailDrop = TAIL_DROP_NEVER;
     private int rsvUse = RSV_USE_ALWAYS;
 
     protected CompressExtension()
     {
-        deflater = new Deflater(Deflater.DEFAULT_COMPRESSION,NOWRAP);
-        inflater = new Inflater(NOWRAP);
         tailDrop = getTailDropMode();
         rsvUse = getRsvUseMode();
     }
 
     public Deflater getDeflater()
     {
-        return deflater;
+        if (deflaterImpl == null)
+        {
+            deflaterImpl = new Deflater(Deflater.DEFAULT_COMPRESSION,NOWRAP);
+        }
+        return deflaterImpl;
     }
 
     public Inflater getInflater()
     {
-        return inflater;
+        if (inflaterImpl == null)
+        {
+            inflaterImpl = new Inflater(NOWRAP);
+        }
+        return inflaterImpl;
     }
 
     /**
@@ -155,6 +161,8 @@ protected void decompress(ByteAccumulator accumulator, ByteBuffer buf) throws Da
         }
         byte[] output = new byte[DECOMPRESS_BUF_SIZE];
 
+        Inflater inflater = getInflater();
+
         while(buf.hasRemaining() && inflater.needsInput())
         {
             if (!supplyInput(inflater,buf))
@@ -346,6 +354,17 @@ public static boolean endsWithTail(ByteBuffer buf)
         }
         return true;
     }
+
+    @Override
+    protected void doStop() throws Exception
+    {
+        LOG.info("doStop()");
+        if(deflaterImpl != null)
+            deflaterImpl.end();
+        if(inflaterImpl != null)
+            inflaterImpl.end();
+        super.doStop();
+    }
 
     @Override
     public String toString()
@@ -429,6 +448,8 @@ private void compress(FrameEntry entry, boolean first)
                 LOG.debug("Compressing {}: {} bytes in {} bytes chunk",entry,remaining,outputLength);
 
             boolean needsCompress = true;
+
+            Deflater deflater = getDeflater();
 
             if (deflater.needsInput() && !supplyInput(deflater,data))
             {