1 /*

     2  * Copyright (c) 2008, 2009, Oracle and/or its affiliates. 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.  Oracle designates this

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

     9  * by Oracle 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 Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package sun.nio.ch;

    27 

    28 import java.nio.ByteBuffer;

    29 import java.nio.channels.*;

    30 import java.net.*;

    31 import java.io.IOException;

    32 import java.util.*;

    33 import java.util.concurrent.*;

    34 import java.security.AccessController;

    35 import java.security.AccessControlContext;

    36 import java.security.PrivilegedExceptionAction;

    37 import java.security.PrivilegedActionException;

    38 

    39 /**

    40  * A prototype implementation of AsynchronousDatagramChannel, used to aid

    41  * test and spec development.

    42  */

    43 

    44 class SimpleAsynchronousDatagramChannelImpl

    45     extends AsynchronousDatagramChannel implements Groupable, Cancellable

    46 {

    47     private final DatagramChannel dc;

    48     private final AsynchronousChannelGroupImpl group;

    49     private final Object attachKey;

    50     private boolean closed;

    51 

    52     // used to coordinate timed and blocking reads

    53     private final Object readLock = new Object();

    54 

    55     // channel blocking mode (requires readLock)

    56     private boolean isBlocking = true;

    57 

    58     // number of blocking readers (requires readLock)

    59     private int blockingReaderCount;

    60 

    61     // true if timed read attempted while blocking read in progress (requires readLock)

    62     private boolean transitionToNonBlocking;

    63 

    64     // true if a blocking read is cancelled (requires readLock)

    65     private boolean blockingReadKilledByCancel;

    66 

    67     // temporary Selectors used by timed reads (requires readLock)

    68     private Selector firstReader;

    69     private Set<Selector> otherReaders;

    70 

    71     SimpleAsynchronousDatagramChannelImpl(ProtocolFamily family,

    72                                           AsynchronousChannelGroupImpl group)

    73         throws IOException

    74     {

    75         super(group.provider());

    76         this.dc = (family == null) ?

    77             DatagramChannel.open() : DatagramChannel.open(family);

    78         this.group = group;

    79 

    80         // attach this channel to the group as foreign channel

    81         boolean registered = false;

    82         try {

    83             if (!(dc instanceof DatagramChannelImpl))

    84                 throw new UnsupportedOperationException();

    85             attachKey = group

    86                 .attachForeignChannel(this, ((DatagramChannelImpl)dc).getFD());

    87             registered = true;

    88         } finally {

    89             if (!registered)

    90                 dc.close();

    91         }

    92     }

    93 

    94     // throws RuntimeException if blocking read has been cancelled

    95     private void ensureBlockingReadNotKilled() {

    96         assert Thread.holdsLock(readLock);

    97         if (blockingReadKilledByCancel)

    98             throw new RuntimeException("Reading not allowed due to cancellation");

    99     }

   100 

   101     // invoke prior to non-timed read/receive

   102     private void beginNoTimeoutRead() {

   103         synchronized (readLock) {

   104             ensureBlockingReadNotKilled();

   105             if (isBlocking)

   106                 blockingReaderCount++;

   107         }

   108     }

   109 

   110     // invoke after non-timed read/receive has completed

   111     private void endNoTimeoutRead() {

   112         synchronized (readLock) {

   113             if (isBlocking) {

   114                 if (--blockingReaderCount == 0 && transitionToNonBlocking) {

   115                     // notify any threads waiting to make channel non-blocking

   116                     readLock.notifyAll();

   117                 }

   118             }

   119         }

   120     }

   121 

   122     // invoke prior to timed read

   123     // returns the timeout remaining

   124     private long prepareForTimedRead(PendingFuture<?,?> result, long timeout)

   125         throws IOException

   126     {

   127         synchronized (readLock) {

   128             ensureBlockingReadNotKilled();

   129             if (isBlocking) {

   130                 transitionToNonBlocking = true;

   131                 while (blockingReaderCount > 0 &&

   132                        timeout > 0L &&

   133                        !result.isCancelled())

   134                 {

   135                     long st = System.currentTimeMillis();

   136                     try {

   137                         readLock.wait(timeout);

   138                     } catch (InterruptedException e) { }

   139                     timeout -= System.currentTimeMillis() - st;

   140                 }

   141                 if (blockingReaderCount == 0) {

   142                     // re-check that blocked read wasn't cancelled

   143                     ensureBlockingReadNotKilled();

   144                     // no blocking reads so change channel to non-blocking

   145                     dc.configureBlocking(false);

   146                     isBlocking = false;

   147                 }

   148             }

   149             return timeout;

   150         }

   151     }

   152 

   153     // returns a temporary Selector

   154     private Selector getSelector() throws IOException {

   155         Selector sel = Util.getTemporarySelector(dc);

   156         synchronized (readLock) {

   157             if (firstReader == null) {

   158                 firstReader = sel;

   159             } else {

   160                 if (otherReaders == null)

   161                     otherReaders = new HashSet<Selector>();

   162                 otherReaders.add(sel);

   163             }

   164         }

   165         return sel;

   166     }

   167 

   168     // releases a temporary Selector

   169     private void releaseSelector(Selector sel) throws IOException {

   170         synchronized (readLock) {

   171             if (firstReader == sel) {

   172                 firstReader = null;

   173             } else {

   174                 otherReaders.remove(sel);

   175             }

   176         }

   177         Util.releaseTemporarySelector(sel);

   178     }

   179 

   180     // wakeup all Selectors currently in use

   181     private void wakeupSelectors() {

   182         synchronized (readLock) {

   183             if (firstReader != null)

   184                 firstReader.wakeup();

   185             if (otherReaders != null) {

   186                 for (Selector sel: otherReaders) {

   187                     sel.wakeup();

   188                 }

   189             }

   190         }

   191     }

   192 

   193     @Override

   194     public AsynchronousChannelGroupImpl group() {

   195         return group;

   196     }

   197 

   198     @Override

   199     public boolean isOpen() {

   200         return dc.isOpen();

   201     }

   202 

   203     @Override

   204     public void onCancel(PendingFuture<?,?> task) {

   205         synchronized (readLock) {

   206             if (blockingReaderCount > 0) {

   207                 blockingReadKilledByCancel = true;

   208                 readLock.notifyAll();

   209                 return;

   210             }

   211         }

   212         wakeupSelectors();

   213     }

   214 

   215     @Override

   216     public void close() throws IOException {

   217         synchronized (dc) {

   218             if (closed)

   219                 return;

   220             closed = true;

   221         }

   222         // detach from group and close underlying channel

   223         group.detachForeignChannel(attachKey);

   224         dc.close();

   225 

   226         // wakeup any threads blocked in timed read/receives

   227         wakeupSelectors();

   228     }

   229 

   230     @Override

   231     public AsynchronousDatagramChannel connect(SocketAddress remote)

   232         throws IOException

   233     {

   234         dc.connect(remote);

   235         return this;

   236     }

   237 

   238     @Override

   239     public AsynchronousDatagramChannel disconnect() throws IOException {

   240         dc.disconnect();

   241         return this;

   242     }

   243 

   244     private static class WrappedMembershipKey extends MembershipKey {

   245         private final MulticastChannel channel;

   246         private final MembershipKey key;

   247 

   248         WrappedMembershipKey(MulticastChannel channel, MembershipKey key) {

   249             this.channel = channel;

   250             this.key = key;

   251         }

   252 

   253         @Override

   254         public boolean isValid() {

   255             return key.isValid();

   256         }

   257 

   258         @Override

   259         public void drop() {

   260             key.drop();

   261         }

   262 

   263         @Override

   264         public MulticastChannel channel() {

   265             return channel;

   266         }

   267 

   268         @Override

   269         public InetAddress group() {

   270             return key.group();

   271         }

   272 

   273         @Override

   274         public NetworkInterface networkInterface() {

   275             return key.networkInterface();

   276         }

   277 

   278         @Override

   279         public InetAddress sourceAddress() {

   280             return key.sourceAddress();

   281         }

   282 

   283         @Override

   284         public MembershipKey block(InetAddress toBlock) throws IOException {

   285             key.block(toBlock);

   286             return this;

   287         }

   288 

   289         @Override

   290         public MembershipKey unblock(InetAddress toUnblock) {

   291             key.unblock(toUnblock);

   292             return this;

   293         }

   294 

   295         @Override

   296         public String toString() {

   297             return key.toString();

   298         }

   299     }

   300 

   301     @Override

   302     public MembershipKey join(InetAddress group,

   303                               NetworkInterface interf)

   304         throws IOException

   305     {

   306         MembershipKey key = ((MulticastChannel)dc).join(group, interf);

   307         return new WrappedMembershipKey(this, key);

   308     }

   309 

   310     @Override

   311     public MembershipKey join(InetAddress group,

   312                               NetworkInterface interf,

   313                               InetAddress source)

   314         throws IOException

   315     {

   316         MembershipKey key = ((MulticastChannel)dc).join(group, interf, source);

   317         return new WrappedMembershipKey(this, key);

   318     }

   319 

   320     private <A> Future<Integer> implSend(ByteBuffer src,

   321                                          SocketAddress target,

   322                                          A attachment,

   323                                          CompletionHandler<Integer,? super A> handler)

   324     {

   325         int n = 0;

   326         Throwable exc = null;

   327         try {

   328             n = dc.send(src, target);

   329         } catch (IOException ioe) {

   330             exc = ioe;

   331         }

   332         if (handler == null)

   333             return CompletedFuture.withResult(n, exc);

   334         Invoker.invoke(this, handler, attachment, n, exc);

   335         return null;

   336     }

   337 

   338     @Override

   339     public Future<Integer> send(ByteBuffer src, SocketAddress target) {

   340         return implSend(src, target, null, null);

   341     }

   342 

   343     @Override

   344     public <A> void send(ByteBuffer src,

   345                          SocketAddress target,

   346                          A attachment,

   347                          CompletionHandler<Integer,? super A> handler)

   348     {

   349         if (handler == null)

   350             throw new NullPointerException("'handler' is null");

   351         implSend(src, target, attachment, handler);

   352     }

   353 

   354     private <A> Future<Integer> implWrite(ByteBuffer src,

   355                                           A attachment,

   356                                           CompletionHandler<Integer,? super A> handler)

   357     {

   358         int n = 0;

   359         Throwable exc = null;

   360         try {

   361             n = dc.write(src);

   362         } catch (IOException ioe) {

   363             exc = ioe;

   364         }

   365         if (handler == null)

   366             return CompletedFuture.withResult(n, exc);

   367         Invoker.invoke(this, handler, attachment, n, exc);

   368         return null;

   369 

   370     }

   371 

   372     @Override

   373     public Future<Integer> write(ByteBuffer src) {

   374         return implWrite(src, null, null);

   375     }

   376 

   377     @Override

   378     public <A> void write(ByteBuffer src,

   379                           A attachment,

   380                           CompletionHandler<Integer,? super A> handler)

   381     {

   382         if (handler == null)

   383             throw new NullPointerException("'handler' is null");

   384         implWrite(src, attachment, handler);

   385     }

   386 

   387     /**

   388      * Receive into the given buffer with privileges enabled and restricted by

   389      * the given AccessControlContext (can be null).

   390      */

   391     private SocketAddress doRestrictedReceive(final ByteBuffer dst,

   392                                               AccessControlContext acc)

   393         throws IOException

   394     {

   395         if (acc == null) {

   396             return dc.receive(dst);

   397         } else {

   398             try {

   399                 return AccessController.doPrivileged(

   400                     new PrivilegedExceptionAction<SocketAddress>() {

   401                         public SocketAddress run() throws IOException {

   402                             return dc.receive(dst);

   403                         }}, acc);

   404             } catch (PrivilegedActionException pae) {

   405                 Exception cause = pae.getException();

   406                 if (cause instanceof SecurityException)

   407                     throw (SecurityException)cause;

   408                 throw (IOException)cause;

   409             }

   410         }

   411     }

   412 

   413     private <A> Future<SocketAddress> implReceive(final ByteBuffer dst,

   414                                                   final long timeout,

   415                                                   final TimeUnit unit,

   416                                                   A attachment,

   417                                                   final CompletionHandler<SocketAddress,? super A> handler)

   418     {

   419         if (dst.isReadOnly())

   420             throw new IllegalArgumentException("Read-only buffer");

   421         if (timeout < 0L)

   422             throw new IllegalArgumentException("Negative timeout");

   423         if (unit == null)

   424             throw new NullPointerException();

   425 

   426         // complete immediately if channel closed

   427         if (!isOpen()) {

   428             Throwable exc = new ClosedChannelException();

   429             if (handler == null)

   430                 return CompletedFuture.withFailure(exc);

   431             Invoker.invoke(this, handler, attachment, null, exc);

   432             return null;

   433         }

   434 

   435         final AccessControlContext acc = (System.getSecurityManager() == null) ?

   436             null : AccessController.getContext();

   437         final PendingFuture<SocketAddress,A> result =

   438             new PendingFuture<SocketAddress,A>(this, handler, attachment);

   439         Runnable task = new Runnable() {

   440             public void run() {

   441                 try {

   442                     SocketAddress remote = null;

   443                     long to;

   444                     if (timeout == 0L) {

   445                         beginNoTimeoutRead();

   446                         try {

   447                             remote = doRestrictedReceive(dst, acc);

   448                         } finally {

   449                             endNoTimeoutRead();

   450                         }

   451                         to = 0L;

   452                     } else {

   453                         to = prepareForTimedRead(result, unit.toMillis(timeout));

   454                         if (to <= 0L)

   455                             throw new InterruptedByTimeoutException();

   456                         remote = doRestrictedReceive(dst, acc);

   457                     }

   458                     if (remote == null) {

   459                         Selector sel = getSelector();

   460                         SelectionKey sk = null;

   461                         try {

   462                             sk = dc.register(sel, SelectionKey.OP_READ);

   463                             for (;;) {

   464                                 if (!dc.isOpen())

   465                                     throw new AsynchronousCloseException();

   466                                 if (result.isCancelled())

   467                                     break;

   468                                 long st = System.currentTimeMillis();

   469                                 int ns = sel.select(to);

   470                                 if (ns > 0) {

   471                                     remote = doRestrictedReceive(dst, acc);

   472                                     if (remote != null)

   473                                         break;

   474                                 }

   475                                 sel.selectedKeys().remove(sk);

   476                                 if (timeout != 0L) {

   477                                     to -= System.currentTimeMillis() - st;

   478                                     if (to <= 0)

   479                                         throw new InterruptedByTimeoutException();

   480                                 }

   481                             }

   482                         } finally {

   483                             if (sk != null)

   484                                 sk.cancel();

   485                             releaseSelector(sel);

   486                         }

   487                     }

   488                     result.setResult(remote);

   489                 } catch (Throwable x) {

   490                     if (x instanceof ClosedChannelException)

   491                         x = new AsynchronousCloseException();

   492                     result.setFailure(x);

   493                 }

   494                 Invoker.invokeUnchecked(result);

   495             }

   496         };

   497         try {

   498             group.executeOnPooledThread(task);

   499         } catch (RejectedExecutionException ree) {

   500             throw new ShutdownChannelGroupException();

   501         }

   502         return result;

   503     }

   504 

   505     @Override

   506     public Future<SocketAddress> receive(ByteBuffer dst) {

   507         return implReceive(dst, 0L, TimeUnit.MILLISECONDS, null, null);

   508     }

   509 

   510     @Override

   511     public <A> void receive(ByteBuffer dst,

   512                             long timeout,

   513                             TimeUnit unit,

   514                             A attachment,

   515                             CompletionHandler<SocketAddress,? super A> handler)

   516     {

   517         if (handler == null)

   518             throw new NullPointerException("'handler' is null");

   519         implReceive(dst, timeout, unit, attachment, handler);

   520     }

   521 

   522     private <A> Future<Integer> implRead(final ByteBuffer dst,

   523                                          final long timeout,

   524                                          final TimeUnit unit,

   525                                          A attachment,

   526                                          final CompletionHandler<Integer,? super A> handler)

   527     {

   528         if (dst.isReadOnly())

   529             throw new IllegalArgumentException("Read-only buffer");

   530         if (timeout < 0L)

   531             throw new IllegalArgumentException("Negative timeout");

   532         if (unit == null)

   533             throw new NullPointerException();

   534 

   535         // complete immediately if channel closed

   536         if (!isOpen()) {

   537             Throwable exc = new ClosedChannelException();

   538             if (handler == null)

   539                 return CompletedFuture.withFailure(exc);

   540             Invoker.invoke(this, handler, attachment, null, exc);

   541             return null;

   542         }

   543 

   544         // another thread may disconnect before read is initiated

   545         if (!dc.isConnected())

   546             throw new NotYetConnectedException();

   547 

   548         final PendingFuture<Integer,A> result =

   549             new PendingFuture<Integer,A>(this, handler, attachment);

   550         Runnable task = new Runnable() {

   551             public void run() {

   552                 try {

   553                     int n = 0;

   554                     long to;

   555                     if (timeout == 0L) {

   556                         beginNoTimeoutRead();

   557                         try {

   558                             n = dc.read(dst);

   559                         } finally {

   560                             endNoTimeoutRead();

   561                         }

   562                         to = 0L;

   563                     } else {

   564                         to = prepareForTimedRead(result, unit.toMillis(timeout));

   565                         if (to <= 0L)

   566                             throw new InterruptedByTimeoutException();

   567                         n = dc.read(dst);

   568                     }

   569                     if (n == 0) {

   570                         Selector sel = getSelector();

   571                         SelectionKey sk = null;

   572                         try {

   573                             sk = dc.register(sel, SelectionKey.OP_READ);

   574                             for (;;) {

   575                                 if (!dc.isOpen())

   576                                     throw new AsynchronousCloseException();

   577                                 if (result.isCancelled())

   578                                     break;

   579                                 long st = System.currentTimeMillis();

   580                                 int ns = sel.select(to);

   581                                 if (ns > 0) {

   582                                     if ((n = dc.read(dst)) != 0)

   583                                         break;

   584                                 }

   585                                 sel.selectedKeys().remove(sk);

   586                                 if (timeout != 0L) {

   587                                     to -= System.currentTimeMillis() - st;

   588                                     if (to <= 0)

   589                                         throw new InterruptedByTimeoutException();

   590                                 }

   591                             }

   592                         } finally {

   593                             if (sk != null)

   594                                 sk.cancel();

   595                             releaseSelector(sel);

   596                         }

   597                     }

   598                     result.setResult(n);

   599                 } catch (Throwable x) {

   600                     if (x instanceof ClosedChannelException)

   601                         x = new AsynchronousCloseException();

   602                     result.setFailure(x);

   603                 }

   604                 Invoker.invokeUnchecked(result);

   605             }

   606         };

   607         try {

   608             group.executeOnPooledThread(task);

   609         } catch (RejectedExecutionException ree) {

   610             throw new ShutdownChannelGroupException();

   611         }

   612         return result;

   613     }

   614 

   615     @Override

   616     public Future<Integer> read(ByteBuffer dst) {

   617         return implRead(dst, 0L, TimeUnit.MILLISECONDS, null, null);

   618     }

   619 

   620     @Override

   621     public <A> void read(ByteBuffer dst,

   622                             long timeout,

   623                             TimeUnit unit,

   624                             A attachment,

   625                             CompletionHandler<Integer,? super A> handler)

   626     {

   627         if (handler == null)

   628             throw new NullPointerException("'handler' is null");

   629         implRead(dst, timeout, unit, attachment, handler);

   630     }

   631 

   632     @Override

   633     public  AsynchronousDatagramChannel bind(SocketAddress local)

   634         throws IOException

   635     {

   636         dc.bind(local);

   637         return this;

   638     }

   639 

   640     @Override

   641     public SocketAddress getLocalAddress() throws IOException {

   642         return dc.getLocalAddress();

   643     }

   644 

   645     @Override

   646     public <T> AsynchronousDatagramChannel setOption(SocketOption<T> name, T value)

   647         throws IOException

   648     {

   649         dc.setOption(name, value);

   650         return this;

   651     }

   652 

   653     @Override

   654     public  <T> T getOption(SocketOption<T> name) throws IOException {

   655         return dc.getOption(name);

   656     }

   657 

   658     @Override

   659     public Set<SocketOption<?>> supportedOptions() {

   660         return dc.supportedOptions();

   661     }

   662 

   663     @Override

   664     public SocketAddress getRemoteAddress() throws IOException {

   665         return dc.getRemoteAddress();

   666     }

   667 }