d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

4b5c8e93cab28d3c65ba9d407fd8f46e3be1db1cMatthew Ahrens

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.nio.ByteBuffer;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.nio.channels.*;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.net.*;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.util.*;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.util.concurrent.*;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.util.concurrent.atomic.*;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedyimport java.io.IOException;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedypublic class Identity {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy static final Random rand = new Random();

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy static final CountDownLatch done = new CountDownLatch(1);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy static final AtomicBoolean failed = new AtomicBoolean(false);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy static void fail(String msg) {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy failed.set(true);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy done.countDown();

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy throw new RuntimeException(msg);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy }

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy // thread-local identifies the thread

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy private static final ThreadLocal<Integer> myGroup =

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy new ThreadLocal<Integer>() {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy @Override protected Integer initialValue() {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy return Integer.valueOf(-1);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy }

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy };

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy // creates a ThreadFactory that constructs groups with the given identity

4b5c8e93cab28d3c65ba9d407fd8f46e3be1db1cMatthew Ahrens static final ThreadFactory createThreadFactory(final int groupId) {

4b5c8e93cab28d3c65ba9d407fd8f46e3be1db1cMatthew Ahrens return new ThreadFactory() {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy @Override

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy public Thread newThread(final Runnable r) {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy Thread t = new Thread(new Runnable() {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy public void run() {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy myGroup.set(groupId);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy r.run();

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy }});

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy t.setDaemon(true);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy return t;

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy }

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy };

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy }

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy public static void main(String[] args) throws Exception {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy // create listener to accept connections

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy final AsynchronousServerSocketChannel listener =

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy AsynchronousServerSocketChannel.open()

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy .bind(new InetSocketAddress(0));

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy listener.accept((Void)null, new CompletionHandler<AsynchronousSocketChannel,Void>() {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy public void completed(final AsynchronousSocketChannel ch, Void att) {

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy listener.accept((Void)null, this);

d583b39bfb4e2571d3e41097c5c357ffe353ad45John Wren Kennedy final ByteBuffer buf = ByteBuffer.allocate(100);

ch.read(buf, ch, new CompletionHandler<Integer,AsynchronousSocketChannel>() {

public void completed(Integer bytesRead, AsynchronousSocketChannel ch) {

if (bytesRead < 0) {

try { ch.close(); } catch (IOException ignore) { }

} else {

buf.clear();

ch.read(buf, ch, this);

}

}

public void failed(Throwable exc, AsynchronousSocketChannel ch) {

try { ch.close(); } catch (IOException ignore) { }

}

});

}

public void failed(Throwable exc, Void att) {

}

});

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

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

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

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

AsynchronousChannelGroup[] groups = new AsynchronousChannelGroup[groupCount];

final AsynchronousSocketChannel[] channels = new AsynchronousSocketChannel[groupCount];

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

ThreadFactory factory = createThreadFactory(i);

AsynchronousChannelGroup group;

if (rand.nextBoolean()) {

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

group = AsynchronousChannelGroup.withFixedThreadPool(nThreads, factory);

} else {

ExecutorService pool = Executors.newCachedThreadPool(factory);

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

}

groups[i] = group;

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

AsynchronousSocketChannel ch = AsynchronousSocketChannel.open(group);

ch.connect(sa).get();

channels[i] = ch;

}

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

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

final AtomicInteger writeCount = new AtomicInteger(100);

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

public void completed(Integer bytesWritten, Integer groupId) {

if (bytesWritten != 1)

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

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

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

if (writeCount.decrementAndGet() > 0) {

int id = rand.nextInt(groupCount);

channels[id].write(getBuffer(), id, this);

} else {

done.countDown();

}

}

public void failed(Throwable exc, Integer groupId) {

fail(exc.getMessage());

}

});

// wait until done

done.await();

// clean-up

for (AsynchronousSocketChannel ch: channels)

ch.close();

for (AsynchronousChannelGroup group: groups)

group.shutdownNow();

listener.close();

if (failed.get())

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

}

static ByteBuffer getBuffer() {

ByteBuffer buf;

if (rand.nextBoolean()) {

buf = ByteBuffer.allocateDirect(1);

} else {

buf = ByteBuffer.allocate(1);

}

buf.put((byte)0);

buf.flip();

return buf;

}

}