1 /*

     2  * Copyright 2008-2009 Sun Microsystems, Inc.  All Rights Reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Sun designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Sun in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    22  * CA 95054 USA or visit www.sun.com if you need additional information or

    23  * have any questions.

    24  */

    25 

    26 package sun.nio.ch;

    27 

    28 import java.nio.channels.*;

    29 import java.util.concurrent.*;

    30 import java.io.IOException;

    31 

    32 /**

    33  * A Future for a pending I/O operation. A PendingFuture allows for the

    34  * attachment of an additional arbitrary context object and a timer task.

    35  */

    36 

    37 final class PendingFuture<V,A>

    38     extends AbstractFuture<V,A>

    39 {

    40     private static final CancellationException CANCELLED =

    41         new CancellationException();

    42 

    43     private final CompletionHandler<V,? super A> handler;

    44 

    45     // true if result (or exception) is available

    46     private volatile boolean haveResult;

    47     private volatile V result;

    48     private volatile Throwable exc;

    49 

    50     // latch for waiting (created lazily if needed)

    51     private CountDownLatch latch;

    52 

    53     // optional timer task that is cancelled when result becomes available

    54     private Future<?> timeoutTask;

    55 

    56     // optional context object

    57     private volatile Object context;

    58 

    59 

    60     PendingFuture(AsynchronousChannel channel,

    61                   CompletionHandler<V,? super A> handler,

    62                   A attachment,

    63                   Object context)

    64     {

    65         super(channel, attachment);

    66         this.handler = handler;

    67         this.context = context;

    68     }

    69 

    70     PendingFuture(AsynchronousChannel channel,

    71                   CompletionHandler<V,? super A> handler,

    72                   A attachment)

    73     {

    74         super(channel, attachment);

    75         this.handler = handler;

    76     }

    77 

    78     CompletionHandler<V,? super A> handler() {

    79         return handler;

    80     }

    81 

    82     void setContext(Object context) {

    83         this.context = context;

    84     }

    85 

    86     Object getContext() {

    87         return context;

    88     }

    89 

    90     void setTimeoutTask(Future<?> task) {

    91         synchronized (this) {

    92             if (haveResult) {

    93                 task.cancel(false);

    94             } else {

    95                 this.timeoutTask = task;

    96             }

    97         }

    98     }

    99 

   100     // creates latch if required; return true if caller needs to wait

   101     private boolean prepareForWait() {

   102         synchronized (this) {

   103             if (haveResult) {

   104                 return false;

   105             } else {

   106                 if (latch == null)

   107                     latch = new CountDownLatch(1);

   108                 return true;

   109             }

   110         }

   111     }

   112 

   113     /**

   114      * Sets the result, or a no-op if the result or exception is already set.

   115      */

   116     boolean setResult(V res) {

   117         synchronized (this) {

   118             if (haveResult)

   119                 return false;

   120             result = res;

   121             haveResult = true;

   122             if (timeoutTask != null)

   123                 timeoutTask.cancel(false);

   124             if (latch != null)

   125                 latch.countDown();

   126             return true;

   127         }

   128     }

   129 

   130     /**

   131      * Sets the result, or a no-op if the result or exception is already set.

   132      */

   133     boolean setFailure(Throwable x) {

   134         if (!(x instanceof IOException) && !(x instanceof SecurityException))

   135             x = new IOException(x);

   136         synchronized (this) {

   137             if (haveResult)

   138                 return false;

   139             exc = x;

   140             haveResult = true;

   141             if (timeoutTask != null)

   142                 timeoutTask.cancel(false);

   143             if (latch != null)

   144                 latch.countDown();

   145             return true;

   146         }

   147     }

   148 

   149     @Override

   150     public V get() throws ExecutionException, InterruptedException {

   151         if (!haveResult) {

   152             boolean needToWait = prepareForWait();

   153             if (needToWait)

   154                 latch.await();

   155         }

   156         if (exc != null) {

   157             if (exc == CANCELLED)

   158                 throw new CancellationException();

   159             throw new ExecutionException(exc);

   160         }

   161         return result;

   162     }

   163 

   164     @Override

   165     public V get(long timeout, TimeUnit unit)

   166         throws ExecutionException, InterruptedException, TimeoutException

   167     {

   168         if (!haveResult) {

   169             boolean needToWait = prepareForWait();

   170             if (needToWait)

   171                 if (!latch.await(timeout, unit)) throw new TimeoutException();

   172         }

   173         if (exc != null) {

   174             if (exc == CANCELLED)

   175                 throw new CancellationException();

   176             throw new ExecutionException(exc);

   177         }

   178         return result;

   179     }

   180 

   181     @Override

   182     Throwable exception() {

   183         return (exc != CANCELLED) ? exc : null;

   184     }

   185 

   186     @Override

   187     V value() {

   188         return result;

   189     }

   190 

   191     @Override

   192     public boolean isCancelled() {

   193         return (exc == CANCELLED);

   194     }

   195 

   196     @Override

   197     public boolean isDone() {

   198         return haveResult;

   199     }

   200 

   201     @Override

   202     public boolean cancel(boolean mayInterruptIfRunning) {

   203         synchronized (this) {

   204             if (haveResult)

   205                 return false;    // already completed

   206 

   207             // A shutdown of the channel group will close all channels and

   208             // shutdown the executor. To ensure that the completion handler

   209             // is executed we queue the task while holding the lock.

   210             if (handler != null) {

   211                 prepareForWait();

   212                 Runnable cancelTask = new Runnable() {

   213                     public void run() {

   214                         while (!haveResult) {

   215                             try {

   216                                 latch.await();

   217                             } catch (InterruptedException ignore) { }

   218                         }

   219                         handler.cancelled(attachment());

   220                     }

   221                 };

   222                 AsynchronousChannel ch = channel();

   223                 if (ch instanceof Groupable) {

   224                     ((Groupable)ch).group().executeOnPooledThread(cancelTask);

   225                 } else {

   226                     if (ch instanceof AsynchronousFileChannelImpl) {

   227                         ((AsynchronousFileChannelImpl)ch).executor().execute(cancelTask);

   228                     } else {

   229                         throw new AssertionError("Should not get here");

   230                     }

   231                 }

   232             }

   233 

   234             // notify channel

   235             if (channel() instanceof Cancellable)

   236                 ((Cancellable)channel()).onCancel(this);

   237 

   238             // set result and cancel timer

   239             exc = CANCELLED;

   240             haveResult = true;

   241             if (timeoutTask != null)

   242                 timeoutTask.cancel(false);

   243         }

   244 

   245         // close channel if forceful cancel

   246         if (mayInterruptIfRunning) {

   247             try {

   248                 channel().close();

   249             } catch (IOException ignore) { }

   250         }

   251 

   252         // release waiters (this also releases the invoker)

   253         if (latch != null)

   254             latch.countDown();

   255         return true;

   256     }

   257 }