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.

     8  *

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

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

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

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

    13  * accompanied this code).

    14  *

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

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

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

    18  *

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

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

    21  * have any questions.

    22  */

    23 

    24 /* @test

    25  * @bug 4607272

    26  * @summary Unit test for AsynchronousChannelGroup

    27  */

    28 

    29 import java.nio.ByteBuffer;

    30 import java.nio.channels.*;

    31 import java.net.*;

    32 import java.util.*;

    33 import java.util.concurrent.*;

    34 import java.util.concurrent.atomic.*;

    35 

    36 /**

    37  * Tests that the completion handler is invoked by a thread with

    38  * the expected identity.

    39  */

    40 

    41 public class Identity {

    42     static final Random rand = new Random();

    43     static final CountDownLatch done = new CountDownLatch(1);

    44     static final AtomicBoolean failed = new AtomicBoolean(false);

    45 

    46     static void fail(String msg) {

    47         failed.set(true);

    48         done.countDown();

    49         throw new RuntimeException(msg);

    50     }

    51 

    52     // thread-local identifies the thread

    53     private static final ThreadLocal<Integer> myGroup =

    54         new ThreadLocal<Integer>() {

    55             @Override protected Integer initialValue() {

    56                 return Integer.valueOf(-1);

    57             }

    58         };

    59 

    60     // creates a ThreadFactory that constructs groups with the given identity

    61     static final ThreadFactory createThreadFactory(final int groupId) {

    62         return new ThreadFactory() {

    63             @Override

    64             public Thread newThread(final Runnable r) {

    65                 Thread t = new Thread(new Runnable() {

    66                     public void run() {

    67                         myGroup.set(groupId);

    68                         r.run();

    69                     }});

    70                 t.setDaemon(true);

    71                 return t;

    72             }

    73         };

    74     }

    75 

    76     public static void main(String[] args) throws Exception {

    77         // create listener to accept connections

    78         final AsynchronousServerSocketChannel listener =

    79             AsynchronousServerSocketChannel.open()

    80                 .bind(new InetSocketAddress(0));

    81         listener.accept(null, new CompletionHandler<AsynchronousSocketChannel,Void>() {

    82             public void completed(final AsynchronousSocketChannel ch, Void att) {

    83                 listener.accept(null, this);

    84 

    85                 final ByteBuffer buf = ByteBuffer.allocate(100);

    86                 ch.read(buf, null, new CompletionHandler<Integer,Void>() {

    87                     public void completed(Integer bytesRead, Void att) {

    88                         buf.clear();

    89                         ch.read(buf, null, this);

    90                     }

    91                     public void failed(Throwable exc, Void att) {

    92                     }

    93                     public void cancelled(Void att) {

    94                     }

    95                 });

    96             }

    97             public void failed(Throwable exc, Void att) {

    98             }

    99             public void cancelled(Void att) {

   100             }

   101         });

   102         int port = ((InetSocketAddress)(listener.getLocalAddress())).getPort();

   103         SocketAddress sa = new InetSocketAddress(InetAddress.getLocalHost(), port);

   104 

   105         // create 3-10 channels, each in its own group

   106         final int groupCount = 3 + rand.nextInt(8);

   107         final AsynchronousSocketChannel[] channel = new AsynchronousSocketChannel[groupCount];

   108         for (int i=0; i<groupCount; i++) {

   109             ThreadFactory factory = createThreadFactory(i);

   110             AsynchronousChannelGroup group;

   111             if (rand.nextBoolean()) {

   112                 int nThreads = 1 + rand.nextInt(10);

   113                 group = AsynchronousChannelGroup.withFixedThreadPool(nThreads, factory);

   114             } else {

   115                 ExecutorService pool = Executors.newCachedThreadPool(factory);

   116                 group = AsynchronousChannelGroup.withCachedThreadPool(pool, rand.nextInt(5));

   117             }

   118 

   119             // create channel in group and connect it to the server

   120             AsynchronousSocketChannel ch = AsynchronousSocketChannel.open(group);

   121             ch.connect(sa).get();

   122             channel[i] = ch;

   123         }

   124 

   125         // randomly write to each channel, ensuring that the completion handler

   126         // is always invoked by a thread with the right identity.

   127         final AtomicInteger writeCount = new AtomicInteger(100);

   128         channel[0].write(getBuffer(), 0, new CompletionHandler<Integer,Integer>() {

   129             public void completed(Integer bytesWritten, Integer groupId) {

   130                 if (bytesWritten != 1)

   131                     fail("Expected 1 byte to be written");

   132                 if (!myGroup.get().equals(groupId))

   133                     fail("Handler invoked by thread with the wrong identity");

   134                 if (writeCount.decrementAndGet() > 0) {

   135                     int id = rand.nextInt(groupCount);

   136                     channel[id].write(getBuffer(), id, this);

   137                 } else {

   138                     done.countDown();

   139                 }

   140             }

   141             public void failed(Throwable exc, Integer groupId) {

   142                 fail(exc.getMessage());

   143             }

   144             public void cancelled(Integer groupId) {

   145                 fail("I/O operation was cancelled");

   146             }

   147         });

   148 

   149         // wait until

   150         done.await();

   151         if (failed.get())

   152             throw new RuntimeException("Test failed - see log for details");

   153     }

   154 

   155     static ByteBuffer getBuffer() {

   156         ByteBuffer buf;

   157         if (rand.nextBoolean()) {

   158             buf = ByteBuffer.allocateDirect(1);

   159         } else {

   160             buf = ByteBuffer.allocate(1);

   161         }

   162         buf.put((byte)0);

   163         buf.flip();

   164         return buf;

   165     }

   166 }