Identity.java revision 1580
2796N/A/*
2796N/A * Copyright 2008-2009 Sun Microsystems, Inc. All Rights Reserved.
2796N/A * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
2796N/A *
2796N/A * This code is free software; you can redistribute it and/or modify it
2796N/A * under the terms of the GNU General Public License version 2 only, as
2796N/A * published by the Free Software Foundation.
2796N/A *
2796N/A * This code is distributed in the hope that it will be useful, but WITHOUT
2796N/A * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
2796N/A * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
2796N/A * version 2 for more details (a copy is included in the LICENSE file that
2796N/A * accompanied this code).
2796N/A *
2796N/A * You should have received a copy of the GNU General Public License version
2796N/A * 2 along with this work; if not, write to the Free Software Foundation,
2796N/A * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
2796N/A *
2796N/A * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
2796N/A * CA 95054 USA or visit www.sun.com if you need additional information or
2796N/A * have any questions.
2796N/A */
2796N/A
2796N/A/* @test
2796N/A * @bug 4607272 6842687
2796N/A * @summary Unit test for AsynchronousChannelGroup
2796N/A */
2796N/A
2796N/Aimport java.nio.ByteBuffer;
2796N/Aimport java.nio.channels.*;
2796N/Aimport java.net.*;
2796N/Aimport java.util.*;
2796N/Aimport java.util.concurrent.*;
2796N/Aimport java.util.concurrent.atomic.*;
2796N/A
2796N/A/**
2796N/A * Tests that the completion handler is invoked by a thread with
2796N/A * the expected identity.
2796N/A */
2796N/A
2796N/Apublic class Identity {
2796N/A static final Random rand = new Random();
2796N/A static final CountDownLatch done = new CountDownLatch(1);
2796N/A static final AtomicBoolean failed = new AtomicBoolean(false);
2796N/A
2796N/A static void fail(String msg) {
2796N/A failed.set(true);
2796N/A done.countDown();
2796N/A throw new RuntimeException(msg);
2796N/A }
2796N/A
2796N/A // thread-local identifies the thread
2796N/A private static final ThreadLocal<Integer> myGroup =
2796N/A new ThreadLocal<Integer>() {
2796N/A @Override protected Integer initialValue() {
2796N/A return Integer.valueOf(-1);
2796N/A }
2796N/A };
2796N/A
2796N/A // creates a ThreadFactory that constructs groups with the given identity
2796N/A static final ThreadFactory createThreadFactory(final int groupId) {
2796N/A return new ThreadFactory() {
2796N/A @Override
2796N/A public Thread newThread(final Runnable r) {
2796N/A Thread t = new Thread(new Runnable() {
2796N/A public void run() {
2796N/A myGroup.set(groupId);
r.run();
}});
t.setDaemon(true);
return t;
}
};
}
public static void main(String[] args) throws Exception {
// create listener to accept connections
final AsynchronousServerSocketChannel listener =
AsynchronousServerSocketChannel.open()
.bind(new InetSocketAddress(0));
listener.accept((Void)null, new CompletionHandler<AsynchronousSocketChannel,Void>() {
public void completed(final AsynchronousSocketChannel ch, Void att) {
listener.accept((Void)null, this);
final ByteBuffer buf = ByteBuffer.allocate(100);
ch.read(buf, (Void)null, new CompletionHandler<Integer,Void>() {
public void completed(Integer bytesRead, Void att) {
buf.clear();
ch.read(buf, (Void)null, this);
}
public void failed(Throwable exc, Void att) {
}
});
}
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);
final AsynchronousSocketChannel[] channel = 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));
}
// create channel in group and connect it to the server
AsynchronousSocketChannel ch = AsynchronousSocketChannel.open(group);
ch.connect(sa).get();
channel[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);
channel[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);
channel[id].write(getBuffer(), id, this);
} else {
done.countDown();
}
}
public void failed(Throwable exc, Integer groupId) {
fail(exc.getMessage());
}
});
// wait until
done.await();
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;
}
}