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.concurrent.*;

    33 import java.io.IOException;

    34 

    35 public class Unbounded {

    36     // number of concurrent completion handlers

    37     static final int CONCURRENCY_COUNT = 512;

    38 

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

    40         // all accepted connections are added to a queue

    41         final ArrayBlockingQueue<AsynchronousSocketChannel> queue =

    42             new ArrayBlockingQueue<AsynchronousSocketChannel>(CONCURRENCY_COUNT);

    43 

    44         // create listener to accept connections

    45         final AsynchronousServerSocketChannel listener =

    46             AsynchronousServerSocketChannel.open()

    47                 .bind(new InetSocketAddress(0));

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

    49             public void completed(AsynchronousSocketChannel ch, Void att) {

    50                 queue.add(ch);

    51                 listener.accept(null, this);

    52             }

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

    54             }

    55             public void cancelled(Void att) {

    56             }

    57         });

    58         System.out.println("Listener created.");

    59 

    60         // establish lots of connections

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

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

    63         AsynchronousSocketChannel[] channels =

    64             new AsynchronousSocketChannel[CONCURRENCY_COUNT];

    65         for (int i=0; i<CONCURRENCY_COUNT; i++) {

    66             int attempts = 0;

    67             for (;;) {

    68                 try {

    69                     channels[i] = AsynchronousSocketChannel.open();

    70                     channels[i].connect(sa).get();

    71                     break;

    72                 } catch (IOException x) {

    73                     // probably resource issue so back off and retry

    74                     if (++attempts >= 3)

    75                         throw x;

    76                     Thread.sleep(50);

    77                 }

    78             }

    79         }

    80         System.out.println("All connection established.");

    81 

    82         // the barrier where all threads (plus the main thread) wait

    83         final CyclicBarrier barrier = new CyclicBarrier(CONCURRENCY_COUNT+1);

    84 

    85         // initiate a read operation on each channel.

    86         for (int i=0; i<CONCURRENCY_COUNT; i++) {

    87             ByteBuffer buf = ByteBuffer.allocateDirect(100);

    88             channels[i].read( buf, channels[i],

    89                 new CompletionHandler<Integer,AsynchronousSocketChannel>() {

    90                     public void completed(Integer bytesRead, AsynchronousSocketChannel ch) {

    91                         try {

    92                             ch.close();

    93                             barrier.await();

    94                         } catch (Exception x) {

    95                             throw new AssertionError(x);

    96                         }

    97                     }

    98                     public void failed(Throwable exc, AsynchronousSocketChannel ch) {

    99                     }

   100                     public void cancelled(AsynchronousSocketChannel ch) {

   101                     }

   102                 });

   103         }

   104         System.out.println("All read operations outstanding.");

   105 

   106         // write data to each of the accepted connections

   107         int remaining = CONCURRENCY_COUNT;

   108         while (remaining > 0) {

   109             AsynchronousSocketChannel ch = queue.take();

   110             ch.write(ByteBuffer.wrap("welcome".getBytes())).get();

   111             ch.close();

   112             remaining--;

   113         }

   114 

   115         // wait for all threads to reach the barrier

   116         System.out.println("Waiting for all threads to reach barrier");

   117         barrier.await();

   118         listener.close();

   119     }

   120 }