lib/jetty/org/eclipse/jetty/util/SharedBlockingCallback.java

   1 //
   2 //  ========================================================================
   3 //  Copyright (c) 1995-2014 Mort Bay Consulting Pty. Ltd.
   4 //  ------------------------------------------------------------------------
   5 //  All rights reserved. This program and the accompanying materials
   6 //  are made available under the terms of the Eclipse Public License v1.0
   7 //  and Apache License v2.0 which accompanies this distribution.
   8 //
   9 //      The Eclipse Public License is available at
  10 //      http://www.eclipse.org/legal/epl-v10.html
  11 //
  12 //      The Apache License v2.0 is available at
  13 //      http://www.opensource.org/licenses/apache2.0.php
  14 //
  15 //  You may elect to redistribute this code under either of these licenses.
  16 //  ========================================================================
  17 //
  18
  19 package org.eclipse.jetty.util;
  20
  21 import java.io.Closeable;
  22 import java.io.IOException;
  23 import java.io.InterruptedIOException;
  24 import java.util.concurrent.CancellationException;
  25 import java.util.concurrent.TimeUnit;
  26 import java.util.concurrent.TimeoutException;
  27 import java.util.concurrent.locks.Condition;
  28 import java.util.concurrent.locks.ReentrantLock;
  29
  30 import org.eclipse.jetty.util.log.Log;
  31 import org.eclipse.jetty.util.log.Logger;
  32 import org.eclipse.jetty.util.thread.NonBlockingThread;
  33
  34
  35 /* ------------------------------------------------------------ */
  36 /** Provides a reusable BlockingCallback.
  37  * A typical usage pattern is:
  38  * <pre>
  39  * void someBlockingCall(Object... args) throws IOException
  40  * {
  41  *   try(Blocker blocker=sharedBlockingCallback.acquire())
  42  *   {
  43  *     someAsyncCall(args,blocker);
  44  *     blocker.block();
  45  *   }
  46  * }
  47  * </pre>
  48  */
  49 public class SharedBlockingCallback
  50 {
  51     private static final Logger LOG = Log.getLogger(SharedBlockingCallback.class);
  52
  53
  54     private static Throwable IDLE = new Throwable()
  55     {
  56         @Override
  57         public String toString()
  58         {
  59             return "IDLE";
  60         }
  61     };
  62
  63     private static Throwable SUCCEEDED = new Throwable()
  64     {
  65         @Override
  66         public String toString()
  67         {
  68             return "SUCCEEDED";
  69         }
  70     };
  71
  72     private static Throwable FAILED = new Throwable()
  73     {
  74         @Override
  75         public String toString()
  76         {
  77             return "FAILED";
  78         }
  79     };
  80
  81     final Blocker _blocker;
  82
  83     public SharedBlockingCallback()
  84     {
  85         this(new Blocker());
  86     }
  87
  88     protected SharedBlockingCallback(Blocker blocker)
  89     {
  90         _blocker=blocker;
  91     }
  92
  93     public Blocker acquire() throws IOException
  94     {
  95         _blocker._lock.lock();
  96         try
  97         {
  98             while (_blocker._state != IDLE)
  99                 _blocker._idle.await();
 100             _blocker._state = null;
 101         }
 102         catch (final InterruptedException e)
 103         {
 104             throw new InterruptedIOException()
 105             {
 106                 {
 107                     initCause(e);
 108                 }
 109             };
 110         }
 111         finally
 112         {
 113             _blocker._lock.unlock();
 114         }
 115         return _blocker;
 116     }
 117
 118
 119     /* ------------------------------------------------------------ */
 120     /** A Closeable Callback.
 121      * Uses the auto close mechanism to check block has been called OK.
 122      */
 123     public static class Blocker implements Callback, Closeable
 124     {
 125         final ReentrantLock _lock = new ReentrantLock();
 126         final Condition _idle = _lock.newCondition();
 127         final Condition _complete = _lock.newCondition();
 128         Throwable _state = IDLE;
 129
 130         public Blocker()
 131         {
 132         }
 133
 134         @Override
 135         public void succeeded()
 136         {
 137             _lock.lock();
 138             try
 139             {
 140                 if (_state == null)
 141                 {
 142                     _state = SUCCEEDED;
 143                     _complete.signalAll();
 144                 }
 145                 else if (_state == IDLE)
 146                     throw new IllegalStateException("IDLE");
 147             }
 148             finally
 149             {
 150                 _lock.unlock();
 151             }
 152         }
 153
 154         @Override
 155         public void failed(Throwable cause)
 156         {
 157             _lock.lock();
 158             try
 159             {
 160                 if (_state == null)
 161                 {
 162                     // TODO remove when feedback received on 435322
 163                     if (cause==null)
 164                         LOG.warn("null failed cause (please report stack trace) ",new Throwable());
 165                     _state = cause==null?FAILED:cause;
 166                     _complete.signalAll();
 167                 }
 168                 else if (_state == IDLE)
 169                     throw new IllegalStateException("IDLE");
 170             }
 171             finally
 172             {
 173                 _lock.unlock();
 174             }
 175         }
 176
 177         /**
 178          * 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
 179          * repeatable use a FutureCallback to convert an asynchronous API to a blocking API.
 180          *
 181          * @throws IOException
 182          *             if exception was caught during blocking, or callback was cancelled
 183          */
 184         public void block() throws IOException
 185         {
 186             if (NonBlockingThread.isNonBlockingThread())
 187                 LOG.warn("Blocking a NonBlockingThread: ",new Throwable());
 188
 189             _lock.lock();
 190             try
 191             {
 192                 while (_state == null)
 193                 {
 194                     // TODO remove this debug timout!
 195                     // This is here to help debug 435322,
 196                     if (!_complete.await(10,TimeUnit.MINUTES))
 197                     {
 198                         IOException x = new IOException("DEBUG timeout");
 199                         LOG.warn("Blocked too long (please report!!!) "+this, x);
 200                         _state=x;
 201                     }
 202                 }
 203
 204                 if (_state == SUCCEEDED)
 205                     return;
 206                 if (_state == IDLE)
 207                     throw new IllegalStateException("IDLE");
 208                 if (_state instanceof IOException)
 209                     throw (IOException)_state;
 210                 if (_state instanceof CancellationException)
 211                     throw (CancellationException)_state;
 212                 if (_state instanceof RuntimeException)
 213                     throw (RuntimeException)_state;
 214                 if (_state instanceof Error)
 215                     throw (Error)_state;
 216                 throw new IOException(_state);
 217             }
 218             catch (final InterruptedException e)
 219             {
 220                 throw new InterruptedIOException()
 221                 {
 222                     {
 223                         initCause(e);
 224                     }
 225                 };
 226             }
 227             finally
 228             {
 229                 _lock.unlock();
 230             }
 231         }
 232
 233         /**
 234          * Check the Callback has succeeded or failed and after the return leave in the state to allow reuse.
 235          *
 236          * @throws IOException
 237          *             if exception was caught during blocking, or callback was cancelled
 238          */
 239         @Override
 240         public void close() throws IOException
 241         {
 242             _lock.lock();
 243             try
 244             {
 245                 if (_state == IDLE)
 246                     throw new IllegalStateException("IDLE");
 247                 if (_state == null)
 248                     LOG.debug("Blocker not complete",new Throwable());
 249             }
 250             finally
 251             {
 252                 _state = IDLE;
 253                 _idle.signalAll();
 254                 _lock.unlock();
 255             }
 256         }
 257
 258         @Override
 259         public String toString()
 260         {
 261             _lock.lock();
 262             try
 263             {
 264                 return String.format("%s@%x{%s}",SharedBlockingCallback.class.getSimpleName(),hashCode(),_state);
 265             }
 266             finally
 267             {
 268                 _lock.unlock();
 269             }
 270         }
 271     }
 272 }