Revert "430242 - added SharedBlockingCallback to support threadsafe blocking"

This reverts commit 9c30e7a303.

jetty-server/src/main/java/org/eclipse/jetty/server/HttpChannel.java

@@ -53,7 +53,6 @@ import org.eclipse.jetty.util.BlockingCallback;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.URIUtil;
-import org.eclipse.jetty.util.SharedBlockingCallback.Blocker;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.util.thread.Scheduler;
@@ -729,10 +728,10 @@ public class HttpChannel<T> implements HttpParser.RequestHandler<T>, Runnable
 
     protected boolean sendResponse(ResponseInfo info, ByteBuffer content, boolean complete) throws IOException
     {
-        try(Blocker blocker = _response.getHttpOutput().acquireWriteBlockingCallback())
+        BlockingCallback writeBlock = _response.getHttpOutput().acquireWriteBlockingCallback();
-        {
+        boolean committing=sendResponse(info,content,complete,writeBlock);
-            return sendResponse(info,content,complete,blocker);
+        writeBlock.block();
-        }
+        return committing;
     }
 
     public boolean isCommitted()

jetty-server/src/main/java/org/eclipse/jetty/server/HttpInputOverHTTP.java

@@ -24,7 +24,6 @@ import java.nio.ByteBuffer;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.SharedBlockingCallback;
-import org.eclipse.jetty.util.SharedBlockingCallback.Blocker;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 
@@ -58,11 +57,10 @@ public class HttpInputOverHTTP extends HttpInput<ByteBuffer> implements Callback
     {
         while(true)
         {
-            try (Blocker blocker=_readBlocker.acquire())
+            _readBlocker.acquire();
-            {            
+            _httpConnection.fillInterested(_readBlocker);
-                _httpConnection.fillInterested(blocker);
+            LOG.debug("{} block readable on {}",this,_readBlocker);
-                LOG.debug("{} block readable on {}",this,blocker);
+            _readBlocker.block();
-            }
 
             Object content=getNextContent();
             if (content!=null || isFinished())

jetty-server/src/main/java/org/eclipse/jetty/server/HttpOutput.java

@@ -35,7 +35,6 @@ import org.eclipse.jetty.http.HttpContent;
 import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.util.BlockingCallback;
 import org.eclipse.jetty.util.SharedBlockingCallback;
-import org.eclipse.jetty.util.SharedBlockingCallback.Blocker;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.IteratingCallback;
@@ -117,17 +116,17 @@ public class HttpOutput extends ServletOutputStream implements Runnable
         return _channel.getResponse().isAllContentWritten(_written);
     }
 
-    protected Blocker acquireWriteBlockingCallback() throws IOException
+    protected BlockingCallback acquireWriteBlockingCallback() throws IOException
     {
-        return _writeblock.acquire();
+        _writeblock.acquire();
+        return _writeblock;
     }
     
     protected void write(ByteBuffer content, boolean complete) throws IOException
     {
-        try (Blocker blocker=_writeblock.acquire())
+        _writeblock.acquire();
-        {        
+        write(content,complete,_writeblock);
-            write(content,complete,blocker);
+        _writeblock.block();
-        }
     }
     
     protected void write(ByteBuffer content, boolean complete, Callback callback)
@@ -440,10 +439,9 @@ public class HttpOutput extends ServletOutputStream implements Runnable
                     // Check if all written or full
                     if (complete || BufferUtil.isFull(_aggregate))
                     {
-                        try(Blocker blocker=_writeblock.acquire())
+                        _writeblock.acquire();
-                        {
+                        write(_aggregate, complete, _writeblock);
-                            write(_aggregate, complete, blocker);
+                        _writeblock.block();
-                        }
                         if (complete)
                             closed();
                     }
@@ -499,10 +497,9 @@ public class HttpOutput extends ServletOutputStream implements Runnable
      */
     public void sendContent(ByteBuffer content) throws IOException
     {
-        try(Blocker blocker=_writeblock.acquire())
+        _writeblock.acquire();
-        {
+        write(content,true,_writeblock);
-            write(content,true,blocker);
+        _writeblock.block();
-        }
     }
 
     /* ------------------------------------------------------------ */
@@ -512,10 +509,9 @@ public class HttpOutput extends ServletOutputStream implements Runnable
      */
     public void sendContent(InputStream in) throws IOException
     {
-        try(Blocker blocker=_writeblock.acquire())
+        _writeblock.acquire();
-        {
+        new InputStreamWritingCB(in,_writeblock).iterate();
-            new InputStreamWritingCB(in,blocker).iterate();
+        _writeblock.block();
-        }
     }
 
     /* ------------------------------------------------------------ */
@@ -525,10 +521,9 @@ public class HttpOutput extends ServletOutputStream implements Runnable
      */
     public void sendContent(ReadableByteChannel in) throws IOException
     {
-        try(Blocker blocker=_writeblock.acquire())
+        _writeblock.acquire();
-        {
+        new ReadableByteChannelWritingCB(in,_writeblock).iterate();
-            new ReadableByteChannelWritingCB(in,blocker).iterate();
+        _writeblock.block();
-        }
     }
 
 
@@ -539,10 +534,9 @@ public class HttpOutput extends ServletOutputStream implements Runnable
      */
     public void sendContent(HttpContent content) throws IOException
     {
-        try(Blocker blocker=_writeblock.acquire())
+        _writeblock.acquire();
-        {
+        sendContent(content,_writeblock);
-            sendContent(content,blocker);
+        _writeblock.block();
-        }
     }
 
     /* ------------------------------------------------------------ */

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

@@ -26,7 +26,30 @@ import java.util.concurrent.atomic.AtomicReference;
 
 /* ------------------------------------------------------------ */
 /**
- * TODO
+ * A Callback for simple reusable conversion of an 
+ * asynchronous API to blocking.
+ * <p>
+ * To avoid late redundant calls to {@link #succeeded()} or {@link #failed(Throwable)} from
+ * interfering with later reuses of this class, the callback context is used to hold pass a phase indicated
+ * and only a single callback per phase is allowed.
+ * <p>
+ * A typical usage pattern is:
+ * <pre>
+ * public class MyClass
+ * {
+ *     BlockingCallback cb = new BlockingCallback();
+ *     
+ *     public void blockingMethod(Object args) throws Exception
+ *     {
+ *         asyncMethod(args,cb);
+ *         cb.block();
+ *     }
+ *     
+ *     public <C>void asyncMethod(Object args, Callback callback)
+ *     {
+ *         ...
+ *     }
+ *  }
  */
 public class BlockingCallback implements Callback
 {

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

@@ -18,102 +18,82 @@
 
 package org.eclipse.jetty.util;
 
-import java.io.Closeable;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
-
 /* ------------------------------------------------------------ */
-/** Provides a reusable BlockingCallback.
+/**
+ * A Callback for simple reusable conversion of an 
+ * asynchronous API to blocking.
+ * <p>
+ * To avoid late redundant calls to {@link #succeeded()} or {@link #failed(Throwable)} from
+ * interfering with later reuses of this class, the callback context is used to hold pass a phase indicated
+ * and only a single callback per phase is allowed.
+ * <p>
  * A typical usage pattern is:
  * <pre>
- * void someBlockingCall(Object... args) throws IOException
+ * public class MyClass
  * {
- *   try(Blocker blocker=sharedBlockingCallback.acquire())
+ *     BlockingCallback cb = new BlockingCallback();
+ *     
+ *     public void blockingMethod(Object args) throws Exception
  *     {
- *     someAsyncCall(args,blocker);
+ *         asyncMethod(args,cb);
+ *         cb.block();
  *     }
- *   catch(Throwable e)
+ *     
+ *     public <C>void asyncMethod(Object args, Callback callback)
  *     {
- *     blocker.fail(e);
+ *         ...
  *     }
  *  }
- * </pre>
  */
-public class SharedBlockingCallback
+public class SharedBlockingCallback extends BlockingCallback
 {
     private static Throwable IDLE=new Throwable()
     {
         @Override
-        public String toString()
+        public String toString() { return "IDLE"; }
-        {
-            return "IDLE";
-        }
     };
     
     private static Throwable SUCCEEDED=new Throwable()
     {
         @Override
-        public String toString()
+        public String toString() { return "SUCCEEDED"; }
-        {
-            return "SUCCEEDED";
-        }
     };
     
-    final Blocker _blocker;
+
+    final ReentrantLock _lock = new ReentrantLock();
+    Condition _idle = _lock.newCondition();
+    Condition _complete = _lock.newCondition();
+    Throwable _state = IDLE;
+    
     
     public SharedBlockingCallback()
-    {
+    {}
-        this(new Blocker());
-    }
 
-    protected SharedBlockingCallback(Blocker blocker)
+    public void acquire() throws IOException
     {
-        _blocker=blocker;
+        _lock.lock();
-    }
-    
-    public Blocker acquire() throws IOException
-    {
-        _blocker._lock.lock();
         try
         {
-            while (_blocker._state != IDLE)
+            while (_state!=IDLE)
-                _blocker._idle.await();
+                _idle.await();
-            _blocker._state = null;
+            _state=null;
         }
         catch (final InterruptedException e)
         {
-            throw new InterruptedIOException()
+            throw new InterruptedIOException(){{initCause(e);}};
-            {
-                {
-                    initCause(e);
-                }
-            };
         }
         finally
         {
-            _blocker._lock.unlock();
+            _lock.unlock();
         }
-        return _blocker;
-    }
-
-    
-    /* ------------------------------------------------------------ */
-    /** A Closeable Callback.
-     * Uses the auto close mechanism to block until the collback is complete.
-     */
-    public static class Blocker implements Callback, Closeable
-    {
-        final ReentrantLock _lock = new ReentrantLock();
-        final Condition _idle = _lock.newCondition();
-        final Condition _complete = _lock.newCondition();
-        Throwable _state = IDLE;
-
-        public Blocker()
-        {
     }
     
     @Override
@@ -156,15 +136,14 @@ public class SharedBlockingCallback
         }
     }
 
-        /**
+    /** Block until the Callback has succeeded or failed and 
-         * Block until the Callback has succeeded or failed and after the return leave in the state to allow reuse. This is useful for code that wants to
+     * after the return leave in the state to allow reuse.
-         * repeatable use a FutureCallback to convert an asynchronous API to a blocking API.
+     * This is useful for code that wants to repeatable use a FutureCallback to convert
-         * 
+     * an asynchronous API to a blocking API. 
-         * @throws IOException
+     * @throws IOException if exception was caught during blocking, or callback was cancelled 
-         *             if exception was caught during blocking, or callback was cancelled
      */
     @Override
-        public void close() throws IOException
+    public void block() throws IOException
     {
         _lock.lock();
         try
@@ -180,20 +159,11 @@ public class SharedBlockingCallback
                 throw (IOException) _state;
             if (_state instanceof CancellationException)
                 throw (CancellationException) _state;
-                if (_state instanceof RuntimeException)
-                    throw (RuntimeException)_state;
-                if (_state instanceof Error)
-                    throw (Error)_state;
             throw new IOException(_state);
         }
         catch (final InterruptedException e)
         {
-                throw new InterruptedIOException()
+            throw new InterruptedIOException(){{initCause(e);}};
-                {
-                    {
-                        initCause(e);
-                    }
-                };
         }
         finally
         {
@@ -203,6 +173,7 @@ public class SharedBlockingCallback
         }
     }
     
+    
     @Override
     public String toString()
     {
@@ -216,5 +187,5 @@ public class SharedBlockingCallback
             _lock.unlock();
         }
     }
-    }
+
 }

jetty-util/src/test/java/org/eclipse/jetty/util/SharedBlockingCallbackTest.java

@@ -19,17 +19,22 @@
 package org.eclipse.jetty.util;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.eclipse.jetty.util.SharedBlockingCallback.Blocker;
 import org.hamcrest.Matchers;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
 
 public class SharedBlockingCallbackTest
 {
-    final SharedBlockingCallback sbcb= new SharedBlockingCallback();
+    final SharedBlockingCallback fcb= new SharedBlockingCallback();
     
     public SharedBlockingCallbackTest()
     {
@@ -39,21 +44,17 @@ public class SharedBlockingCallbackTest
     @Test
     public void testDone() throws Exception
     {
-        long start;
+        fcb.acquire();
-        try (Blocker blocker=sbcb.acquire())
+        fcb.succeeded();
-        {
+        long start=System.currentTimeMillis();
-            blocker.succeeded();
+        fcb.block();
-            start=System.currentTimeMillis();
-        }
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(500L));     
     }
     
     @Test
     public void testGetDone() throws Exception
     {
-        long start;
+        fcb.acquire();
-        try (final Blocker blocker=sbcb.acquire())
-        {
         final CountDownLatch latch = new CountDownLatch(1);
         
         new Thread(new Runnable()
@@ -63,13 +64,13 @@ public class SharedBlockingCallbackTest
             {
                 latch.countDown();
                 try{TimeUnit.MILLISECONDS.sleep(100);}catch(Exception e){e.printStackTrace();}
-                    blocker.succeeded();
+                fcb.succeeded();
             }
         }).start();
         
         latch.await();
-            start=System.currentTimeMillis();
+        long start=System.currentTimeMillis();
-        }
+        fcb.block();
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.greaterThan(10L)); 
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(1000L)); 
     }
@@ -77,19 +78,18 @@ public class SharedBlockingCallbackTest
     @Test
     public void testFailed() throws Exception
     {
-        final Exception ex = new Exception("FAILED");
+        fcb.acquire();
-        long start=Long.MIN_VALUE;
+        Exception ex=new Exception("FAILED");
+        fcb.failed(ex);
+        
+        long start=System.currentTimeMillis();
         try
         {
-            try (final Blocker blocker=sbcb.acquire())
+            fcb.block();
-            {
-                blocker.failed(ex);
-            }
             Assert.fail();
         }
         catch(IOException ee)
         {
-            start=System.currentTimeMillis();
             Assert.assertEquals(ex,ee.getCause());
         }
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(100L));     
@@ -98,15 +98,10 @@ public class SharedBlockingCallbackTest
     @Test
     public void testGetFailed() throws Exception
     {
+        fcb.acquire();
         final Exception ex=new Exception("FAILED");
-        long start=Long.MIN_VALUE;
         final CountDownLatch latch = new CountDownLatch(1);
         
-        try
-        {
-            try (final Blocker blocker=sbcb.acquire())
-            {
-
         new Thread(new Runnable()
         {
             @Override
@@ -114,13 +109,15 @@ public class SharedBlockingCallbackTest
             {
                 latch.countDown();
                 try{TimeUnit.MILLISECONDS.sleep(100);}catch(Exception e){e.printStackTrace();}
-                        blocker.failed(ex);
+                fcb.failed(ex);
             }
         }).start();
         
         latch.await();
-                start=System.currentTimeMillis();
+        long start=System.currentTimeMillis();
-            }
+        try
+        {
+            fcb.block();
             Assert.fail();
         }
         catch(IOException ee)
@@ -144,12 +141,11 @@ public class SharedBlockingCallbackTest
             {
                 try
                 {
-                    try (Blocker blocker=sbcb.acquire())
+                    fcb.acquire();
-                    {
                     latch.countDown();
                     TimeUnit.MILLISECONDS.sleep(100);
-                        blocker.succeeded();
+                    fcb.succeeded();
-                    }
+                    fcb.block();
                 }
                 catch(Exception e)
                 {
@@ -161,13 +157,12 @@ public class SharedBlockingCallbackTest
         
         latch.await();
         long start=System.currentTimeMillis();
-        try (Blocker blocker=sbcb.acquire())
+        fcb.acquire();
-        {
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.greaterThan(10L)); 
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(500L)); 
 
-            blocker.succeeded();
+        fcb.succeeded();
-        };
+        fcb.block();
         Assert.assertThat(System.currentTimeMillis()-start,Matchers.lessThan(600L));   
     }
 }

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/BlockingWriteCallback.java

@@ -18,29 +18,14 @@
 
 package org.eclipse.jetty.websocket.common;
 
-import java.io.IOException;
-
 import org.eclipse.jetty.util.SharedBlockingCallback;
 import org.eclipse.jetty.websocket.api.WriteCallback;
 
-
+public class BlockingWriteCallback extends SharedBlockingCallback implements WriteCallback
-/* ------------------------------------------------------------ */
-/** extend a SharedlBlockingCallback to an websocket WriteCallback
- */
-public class BlockingWriteCallback extends SharedBlockingCallback
 {
     public BlockingWriteCallback()
-    {
+    {}
-        super(new WriteBlocker());
-    }
     
-    public WriteBlocker acquireWriteBlocker() throws IOException
-    {
-        return (WriteBlocker)acquire();
-    }
-    
-    public static class WriteBlocker extends Blocker implements WriteCallback
-    {
     @Override
     public void writeFailed(Throwable x)
     {
@@ -53,4 +38,3 @@ public class BlockingWriteCallback extends SharedBlockingCallback
         succeeded();
     }
 }
-}

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/WebSocketRemoteEndpoint.java

@@ -32,7 +32,6 @@ import org.eclipse.jetty.websocket.api.BatchMode;
 import org.eclipse.jetty.websocket.api.RemoteEndpoint;
 import org.eclipse.jetty.websocket.api.WriteCallback;
 import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
-import org.eclipse.jetty.websocket.common.BlockingWriteCallback.WriteBlocker;
 import org.eclipse.jetty.websocket.common.frames.BinaryFrame;
 import org.eclipse.jetty.websocket.common.frames.ContinuationFrame;
 import org.eclipse.jetty.websocket.common.frames.DataFrame;
@@ -101,10 +100,9 @@ public class WebSocketRemoteEndpoint implements RemoteEndpoint
 
     private void blockingWrite(WebSocketFrame frame) throws IOException
     {
-        try(WriteBlocker b=blocker.acquireWriteBlocker())
+        blocker.acquire();
-        {
+        uncheckedSendFrame(frame, blocker);
-            uncheckedSendFrame(frame, b);
+        blocker.block();
-        }
     }
 
     private boolean lockMsg(MsgType type)
@@ -443,13 +441,14 @@ public class WebSocketRemoteEndpoint implements RemoteEndpoint
         this.batchMode = batchMode;
     }
 
-    @Override
     public void flush() throws IOException
     {
         lockMsg(MsgType.ASYNC);
-        try (WriteBlocker b = blocker.acquireWriteBlocker())
+        try
         {
-            uncheckedSendFrame(FrameFlusher.FLUSH_FRAME, b);
+            blocker.acquire();
+            uncheckedSendFrame(FrameFlusher.FLUSH_FRAME, blocker);
+            blocker.block();
         }
         finally
         {

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageOutputStream.java

@@ -31,7 +31,6 @@ import org.eclipse.jetty.websocket.api.WriteCallback;
 import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
 import org.eclipse.jetty.websocket.common.BlockingWriteCallback;
 import org.eclipse.jetty.websocket.common.WebSocketSession;
-import org.eclipse.jetty.websocket.common.BlockingWriteCallback.WriteBlocker;
 import org.eclipse.jetty.websocket.common.frames.BinaryFrame;
 
 /**
@@ -143,10 +142,9 @@ public class MessageOutputStream extends OutputStream
             frame.setPayload(buffer);
             frame.setFin(fin);
 
-            try(WriteBlocker b=blocker.acquireWriteBlocker())
+            blocker.acquire();
-            {
+            outgoing.outgoingFrame(frame, blocker, BatchMode.OFF);
-                outgoing.outgoingFrame(frame, b, BatchMode.OFF);
+            blocker.block();
-            }
 
             ++frameCount;
             // Any flush after the first will be a CONTINUATION frame.

jetty-websocket/websocket-common/src/main/java/org/eclipse/jetty/websocket/common/message/MessageWriter.java

@@ -31,7 +31,6 @@ import org.eclipse.jetty.websocket.api.WriteCallback;
 import org.eclipse.jetty.websocket.api.extensions.OutgoingFrames;
 import org.eclipse.jetty.websocket.common.BlockingWriteCallback;
 import org.eclipse.jetty.websocket.common.WebSocketSession;
-import org.eclipse.jetty.websocket.common.BlockingWriteCallback.WriteBlocker;
 import org.eclipse.jetty.websocket.common.frames.TextFrame;
 
 /**
@@ -147,10 +146,9 @@ public class MessageWriter extends Writer
             frame.setPayload(data);
             frame.setFin(fin);
 
-            try (WriteBlocker b = blocker.acquireWriteBlocker())
+            blocker.acquire();
-            {
+            outgoing.outgoingFrame(frame, blocker, BatchMode.OFF);
-                outgoing.outgoingFrame(frame, b, BatchMode.OFF);
+            blocker.block();
-            }
 
             ++frameCount;
             // Any flush after the first will be a CONTINUATION frame.