Issue #4462 - Prevent jetty 10 WebSocket close deadlocks

jetty-util/src/main/java/org/eclipse/jetty/util/BlockingCallback.java

@@ -0,0 +1,84 @@
+//
+// ========================================================================
+// Copyright (c) 1995-2020 Mort Bay Consulting Pty Ltd and others.
+//
+// This program and the accompanying materials are made available under
+// the terms of the Eclipse Public License 2.0 which is available at
+// https://www.eclipse.org/legal/epl-2.0
+//
+// This Source Code may also be made available under the following
+// Secondary Licenses when the conditions for such availability set
+// forth in the Eclipse Public License, v. 2.0 are satisfied:
+// the Apache License v2.0 which is available at
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+// ========================================================================
+//
+
+package org.eclipse.jetty.util;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class BlockingCallback implements Callback
+{
+    private FutureCallback callback = new FutureCallback();
+    private final long timeout;
+
+    public BlockingCallback()
+    {
+        this(-1);
+    }
+
+    public BlockingCallback(long timeout)
+    {
+        this.timeout = timeout;
+    }
+
+    @Override
+    public void succeeded()
+    {
+        callback.succeeded();
+    }
+
+    @Override
+    public void failed(Throwable x)
+    {
+        callback.failed(x);
+    }
+
+    public void block() throws IOException
+    {
+        try
+        {
+            if (timeout > 0)
+                callback.get(timeout, TimeUnit.MILLISECONDS);
+            else
+                callback.get();
+        }
+        catch (InterruptedException e)
+        {
+            InterruptedIOException exception = new InterruptedIOException();
+            exception.initCause(e);
+            throw exception;
+        }
+        catch (ExecutionException e)
+        {
+            Throwable cause = e.getCause();
+            if (cause instanceof RuntimeException)
+                throw (RuntimeException)cause;
+            else if (cause instanceof IOException)
+                throw (IOException)cause;
+            else
+                throw new IOException(cause);
+        }
+        catch (TimeoutException e)
+        {
+            throw new IOException(e);
+        }
+    }
+}

jetty-websocket/websocket-core/src/main/java/org/eclipse/jetty/websocket/core/FrameHandler.java

@@ -400,16 +400,19 @@ public boolean isOutputOpen()
             @Override
             public void flush(Callback callback)
             {
+                callback.succeeded();
             }
 
             @Override
             public void close(Callback callback)
             {
+                callback.succeeded();
             }
 
             @Override
             public void close(int statusCode, String reason, Callback callback)
             {
+                callback.succeeded();
             }
 
             @Override
@@ -420,6 +423,7 @@ public void demand(long n)
             @Override
             public void sendFrame(Frame frame, Callback callback, boolean batch)
             {
+                callback.succeeded();
             }
         }
     }

jetty-websocket/websocket-core/src/main/java/org/eclipse/jetty/websocket/core/internal/WebSocketSessionState.java

@@ -142,15 +142,15 @@ public boolean onEof()
         }
     }
 
-    public boolean onOutgoingFrame(Frame frame) throws ProtocolException
+    public boolean onOutgoingFrame(Frame frame) throws Exception
     {
         byte opcode = frame.getOpCode();
         boolean fin = frame.isFin();
 
         synchronized (this)
         {
             if (!isOutputOpen())
-                throw new IllegalStateException(_sessionState.toString());
+                throw new ClosedChannelException();
 
             if (opcode == OpCode.CLOSE)
             {

jetty-websocket/websocket-javax-common/src/main/java/org/eclipse/jetty/websocket/javax/common/JavaxWebSocketBasicRemote.java

@@ -25,8 +25,8 @@
 import javax.websocket.EncodeException;
 import javax.websocket.RemoteEndpoint;
 
+import org.eclipse.jetty.util.BlockingCallback;
 import org.eclipse.jetty.util.BufferUtil;
-import org.eclipse.jetty.util.SharedBlockingCallback;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.core.Frame;
@@ -65,10 +65,10 @@ public void sendBinary(ByteBuffer data) throws IOException
         {
             LOG.debug("sendBinary({})", BufferUtil.toDetailString(data));
         }
-        try (SharedBlockingCallback.Blocker b = session.getBlocking().acquire())
-        {
-            sendFrame(new Frame(OpCode.BINARY).setPayload(data), b, false);
-        }
+
+        BlockingCallback b = newBlockingCallback();
+        sendFrame(new Frame(OpCode.BINARY).setPayload(data), b, false);
+        b.block();
     }
 
     @Override
@@ -79,37 +79,36 @@ public void sendBinary(ByteBuffer partialByte, boolean isLast) throws IOExceptio
         {
             LOG.debug("sendBinary({},{})", BufferUtil.toDetailString(partialByte), isLast);
         }
-        try (SharedBlockingCallback.Blocker b = session.getBlocking().acquire())
+
+        Frame frame;
+        switch (messageType)
         {
-            Frame frame;
-            switch (messageType)
-            {
-                case -1:
-                    // New message!
-                    frame = new Frame(OpCode.BINARY);
-                    break;
-                case OpCode.TEXT:
-                    throw new IllegalStateException("Cannot send a partial BINARY message: TEXT message in progress");
-                case OpCode.BINARY:
-                    frame = new Frame(OpCode.CONTINUATION);
-                    break;
-                default:
-                    throw new IllegalStateException("Cannot send a partial BINARY message: unrecognized active message type " + OpCode.name(messageType));
-            }
-
-            frame.setPayload(partialByte);
-            frame.setFin(isLast);
-            sendFrame(frame, b, false);
+            case -1:
+                // New message!
+                frame = new Frame(OpCode.BINARY);
+                break;
+            case OpCode.TEXT:
+                throw new IllegalStateException("Cannot send a partial BINARY message: TEXT message in progress");
+            case OpCode.BINARY:
+                frame = new Frame(OpCode.CONTINUATION);
+                break;
+            default:
+                throw new IllegalStateException("Cannot send a partial BINARY message: unrecognized active message type " + OpCode.name(messageType));
         }
+
+        frame.setPayload(partialByte);
+        frame.setFin(isLast);
+        BlockingCallback b = newBlockingCallback();
+        sendFrame(frame, b, false);
+        b.block();
     }
 
     @Override
     public void sendObject(Object data) throws IOException, EncodeException
     {
-        try (SharedBlockingCallback.Blocker b = session.getBlocking().acquire())
-        {
-            super.sendObject(data, b);
-        }
+        BlockingCallback b = newBlockingCallback();
+        super.sendObject(data, b);
+        b.block();
     }
 
     @Override
@@ -120,10 +119,11 @@ public void sendText(String text) throws IOException
         {
             LOG.debug("sendText({})", TextUtil.hint(text));
         }
-        try (SharedBlockingCallback.Blocker b = session.getBlocking().acquire())
-        {
-            sendFrame(new Frame(OpCode.TEXT).setPayload(text), b, false);
-        }
+
+
+        BlockingCallback b = newBlockingCallback();
+        sendFrame(new Frame(OpCode.TEXT).setPayload(text), b, false);
+        b.block();
     }
 
     @Override
@@ -134,27 +134,32 @@ public void sendText(String partialMessage, boolean isLast) throws IOException
         {
             LOG.debug("sendText({},{})", TextUtil.hint(partialMessage), isLast);
         }
-        try (SharedBlockingCallback.Blocker b = session.getBlocking().acquire())
+
+        Frame frame;
+        switch (messageType)
         {
-            Frame frame;
-            switch (messageType)
-            {
-                case -1:
-                    // New message!
-                    frame = new Frame(OpCode.TEXT);
-                    break;
-                case OpCode.TEXT:
-                    frame = new Frame(OpCode.CONTINUATION);
-                    break;
-                case OpCode.BINARY:
-                    throw new IllegalStateException("Cannot send a partial TEXT message: BINARY message in progress");
-                default:
-                    throw new IllegalStateException("Cannot send a partial TEXT message: unrecognized active message type " + OpCode.name(messageType));
-            }
-
-            frame.setPayload(BufferUtil.toBuffer(partialMessage, UTF_8));
-            frame.setFin(isLast);
-            sendFrame(frame, b, false);
+            case -1:
+                // New message!
+                frame = new Frame(OpCode.TEXT);
+                break;
+            case OpCode.TEXT:
+                frame = new Frame(OpCode.CONTINUATION);
+                break;
+            case OpCode.BINARY:
+                throw new IllegalStateException("Cannot send a partial TEXT message: BINARY message in progress");
+            default:
+                throw new IllegalStateException("Cannot send a partial TEXT message: unrecognized active message type " + OpCode.name(messageType));
         }
+
+        frame.setPayload(BufferUtil.toBuffer(partialMessage, UTF_8));
+        frame.setFin(isLast);
+        BlockingCallback b = newBlockingCallback();
+        sendFrame(frame, b, false);
+        b.block();
+    }
+
+    private BlockingCallback newBlockingCallback()
+    {
+        return new BlockingCallback(getIdleTimeout() + 1000);
     }
 }

jetty-websocket/websocket-javax-common/src/main/java/org/eclipse/jetty/websocket/javax/common/JavaxWebSocketRemoteEndpoint.java

@@ -25,9 +25,9 @@
 import javax.websocket.Encoder;
 import javax.websocket.SendHandler;
 
+import org.eclipse.jetty.util.BlockingCallback;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
-import org.eclipse.jetty.util.SharedBlockingCallback;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.websocket.core.Frame;
@@ -66,10 +66,9 @@ protected MessageOutputStream newMessageOutputStream()
     @Override
     public void flushBatch() throws IOException
     {
-        try (SharedBlockingCallback.Blocker blocker = session.getBlocking().acquire())
-        {
-            coreSession.flush(blocker);
-        }
+        BlockingCallback b = newBlockingCallback();
+        coreSession.flush(b);
+        b.block();
     }
 
     @Override
@@ -227,24 +226,22 @@ public void sendObject(Object data, Callback callback) throws IOException, Encod
     public void sendPing(ByteBuffer data) throws IOException, IllegalArgumentException
     {
         if (LOG.isDebugEnabled())
-        {
             LOG.debug("sendPing({})", BufferUtil.toDetailString(data));
-        }
-        // TODO: is this supposed to be a blocking call?
-        // TODO: what to do on excessively large payloads (error and close connection per RFC6455, or truncate?)
-        sendFrame(new Frame(OpCode.PING).setPayload(data), Callback.NOOP, batch);
+
+        BlockingCallback b = newBlockingCallback();
+        sendFrame(new Frame(OpCode.PING).setPayload(data), b, batch);
+        b.block();
     }
 
     @Override
     public void sendPong(ByteBuffer data) throws IOException, IllegalArgumentException
     {
         if (LOG.isDebugEnabled())
-        {
             LOG.debug("sendPong({})", BufferUtil.toDetailString(data));
-        }
-        // TODO: is this supposed to be a blocking call?
-        // TODO: what to do on excessively large payloads (error and close connection per RFC6455, or truncate?)
-        sendFrame(new Frame(OpCode.PONG).setPayload(data), Callback.NOOP, batch);
+
+        BlockingCallback b = newBlockingCallback();
+        sendFrame(new Frame(OpCode.PONG).setPayload(data), b, batch);
+        b.block();
     }
 
     protected void assertMessageNotNull(Object data)
@@ -262,4 +259,9 @@ protected void assertSendHandlerNotNull(SendHandler handler)
             throw new IllegalArgumentException("SendHandler cannot be null");
         }
     }
+
+    private BlockingCallback newBlockingCallback()
+    {
+        return new BlockingCallback(getIdleTimeout() + 1000);
+    }
 }