893N/A

893N/A

893N/Aimport java.nio.ByteBuffer;

893N/Aimport java.nio.channels.*;

893N/Aimport java.net.*;

893N/Aimport java.util.*;

893N/Aimport java.util.concurrent.*;

893N/Aimport java.util.concurrent.atomic.*;

2546N/Aimport java.io.IOException;

893N/A

893N/A

893N/Apublic class Identity {

893N/A static final Random rand = new Random();

893N/A static final CountDownLatch done = new CountDownLatch(1);

893N/A static final AtomicBoolean failed = new AtomicBoolean(false);

893N/A

893N/A static void fail(String msg) {

893N/A failed.set(true);

893N/A done.countDown();

893N/A throw new RuntimeException(msg);

893N/A }

893N/A

893N/A // thread-local identifies the thread

893N/A private static final ThreadLocal<Integer> myGroup =

893N/A new ThreadLocal<Integer>() {

893N/A @Override protected Integer initialValue() {

893N/A return Integer.valueOf(-1);

893N/A }

893N/A };

893N/A

893N/A // creates a ThreadFactory that constructs groups with the given identity

893N/A static final ThreadFactory createThreadFactory(final int groupId) {

893N/A return new ThreadFactory() {

893N/A @Override

893N/A public Thread newThread(final Runnable r) {

893N/A Thread t = new Thread(new Runnable() {

893N/A public void run() {

893N/A myGroup.set(groupId);

893N/A r.run();

893N/A }});

893N/A t.setDaemon(true);

893N/A return t;

893N/A }

893N/A };

893N/A }

893N/A

893N/A public static void main(String[] args) throws Exception {

893N/A // create listener to accept connections

893N/A final AsynchronousServerSocketChannel listener =

893N/A AsynchronousServerSocketChannel.open()

893N/A .bind(new InetSocketAddress(0));

1434N/A listener.accept((Void)null, new CompletionHandler<AsynchronousSocketChannel,Void>() {

893N/A public void completed(final AsynchronousSocketChannel ch, Void att) {

1434N/A listener.accept((Void)null, this);

893N/A final ByteBuffer buf = ByteBuffer.allocate(100);

2546N/A ch.read(buf, ch, new CompletionHandler<Integer,AsynchronousSocketChannel>() {

2546N/A public void completed(Integer bytesRead, AsynchronousSocketChannel ch) {

2546N/A if (bytesRead < 0) {

2546N/A try { ch.close(); } catch (IOException ignore) { }

2546N/A } else {

2546N/A buf.clear();

2546N/A ch.read(buf, ch, this);

2546N/A }

893N/A }

2546N/A public void failed(Throwable exc, AsynchronousSocketChannel ch) {

2546N/A try { ch.close(); } catch (IOException ignore) { }

893N/A }

893N/A });

893N/A }

893N/A public void failed(Throwable exc, Void att) {

893N/A }

893N/A });

893N/A int port = ((InetSocketAddress)(listener.getLocalAddress())).getPort();

893N/A SocketAddress sa = new InetSocketAddress(InetAddress.getLocalHost(), port);

893N/A

893N/A // create 3-10 channels, each in its own group

893N/A final int groupCount = 3 + rand.nextInt(8);

2546N/A AsynchronousChannelGroup[] groups = new AsynchronousChannelGroup[groupCount];

2546N/A final AsynchronousSocketChannel[] channels = new AsynchronousSocketChannel[groupCount];

893N/A for (int i=0; i<groupCount; i++) {

893N/A ThreadFactory factory = createThreadFactory(i);

893N/A AsynchronousChannelGroup group;

893N/A if (rand.nextBoolean()) {

893N/A int nThreads = 1 + rand.nextInt(10);

893N/A group = AsynchronousChannelGroup.withFixedThreadPool(nThreads, factory);

893N/A } else {

893N/A ExecutorService pool = Executors.newCachedThreadPool(factory);

893N/A group = AsynchronousChannelGroup.withCachedThreadPool(pool, rand.nextInt(5));

893N/A }

2546N/A groups[i] = group;

893N/A

893N/A // create channel in group and connect it to the server

893N/A AsynchronousSocketChannel ch = AsynchronousSocketChannel.open(group);

893N/A ch.connect(sa).get();

2546N/A channels[i] = ch;

893N/A }

893N/A

893N/A // randomly write to each channel, ensuring that the completion handler

893N/A // is always invoked by a thread with the right identity.

893N/A final AtomicInteger writeCount = new AtomicInteger(100);

2546N/A channels[0].write(getBuffer(), 0, new CompletionHandler<Integer,Integer>() {

893N/A public void completed(Integer bytesWritten, Integer groupId) {

893N/A if (bytesWritten != 1)

893N/A fail("Expected 1 byte to be written");

893N/A if (!myGroup.get().equals(groupId))

893N/A fail("Handler invoked by thread with the wrong identity");

893N/A if (writeCount.decrementAndGet() > 0) {

893N/A int id = rand.nextInt(groupCount);

2546N/A channels[id].write(getBuffer(), id, this);

893N/A } else {

893N/A done.countDown();

893N/A }

893N/A }

893N/A public void failed(Throwable exc, Integer groupId) {

893N/A fail(exc.getMessage());

893N/A }

893N/A });

893N/A

2546N/A // wait until done

893N/A done.await();

2546N/A

2546N/A // clean-up

2546N/A for (AsynchronousSocketChannel ch: channels)

2546N/A ch.close();

2546N/A for (AsynchronousChannelGroup group: groups)

2546N/A group.shutdownNow();

2546N/A listener.close();

2546N/A

893N/A if (failed.get())

893N/A throw new RuntimeException("Test failed - see log for details");

893N/A }

893N/A

893N/A static ByteBuffer getBuffer() {

893N/A ByteBuffer buf;

893N/A if (rand.nextBoolean()) {

893N/A buf = ByteBuffer.allocateDirect(1);

893N/A } else {

893N/A buf = ByteBuffer.allocate(1);

893N/A }

893N/A buf.put((byte)0);

893N/A buf.flip();

893N/A return buf;

893N/A }

893N/A}