Identity.java revision 893:f06f30b29f36
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 29import java.nio.ByteBuffer; 30import java.nio.channels.*; 31import java.net.*; 32import java.util.*; 33import java.util.concurrent.*; 34import java.util.concurrent.atomic.*; 35 36/** 37 * Tests that the completion handler is invoked by a thread with 38 * the expected identity. 39 */ 40 41public 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} 167