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