1 /*
2 * Copyright (c) 2018, Oracle and/or its affiliates. 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 Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20 * or visit www.oracle.com if you need additional information or have any
21 * questions.
22 */
23
24 /* @test
25 * @bug 8195160
26 * @summary Test RdmaSelector with RdmaServerSocketChannels
27 * @requires (os.family == "linux")
28 * @library .. /test/lib /test/jdk/java/nio/channels
29 * @build RsocketTest
30 * @run main/othervm BasicAccept
31 */
32
33 import java.io.IOException;
34 import java.net.InetAddress;
35 import java.net.InetSocketAddress;
36 import java.net.StandardProtocolFamily;
37 import java.nio.ByteBuffer;
38 import java.nio.channels.Selector;
39 import java.nio.channels.SelectionKey;
40 import java.nio.channels.ServerSocketChannel;
41 import java.nio.channels.SocketChannel;
42 import java.nio.channels.spi.SelectorProvider;
43 import java.util.Iterator;
44 import java.util.Set;
45 import jdk.net.RdmaSockets;
46
47 import jtreg.SkippedException;
48
49 public class SelectorTest {
50 private static List clientList = new LinkedList();
51 private static Random rnd = new Random();
52 public static int NUM_CLIENTS = 5;
53 public static int TEST_PORT = 31452;
54 static PrintStream log = System.err;
55 private static int FINISH_TIME = 30000;
56
57 /*
58 * Usage note
59 *
60 * java SelectorTest [server] [client <host>] [<port>]
61 *
62 * No arguments runs both client and server in separate threads
63 * using the default port of 31452.
64 *
65 * client runs the client on this machine and connects to server
66 * at the given IP address.
67 *
68 * server runs the server on localhost.
69 */
70 public static void main(String[] args) throws Exception {
71 if (!RsocketTest.isRsocketAvailable())
72 throw new SkippedException("rsocket is not available");
73
74 if (args.length == 0) {
75 Server server = new Server(0);
76 server.start();
77 try {
78 Thread.sleep(1000);
79 } catch (InterruptedException e) { }
80 InetSocketAddress isa
81 = new InetSocketAddress(InetAddress.getLocalHost(), server.port());
82 Client client = new Client(isa);
83 client.start();
84 if ((server.finish(FINISH_TIME) & client.finish(FINISH_TIME)) == 0)
85 throw new Exception("Failure");
86 log.println();
87
88 } else if (args[0].equals("server")) {
89
90 if (args.length > 1)
91 TEST_PORT = Integer.parseInt(args[1]);
92 Server server = new Server(TEST_PORT);
93 server.start();
94 if (server.finish(FINISH_TIME) == 0)
95 throw new Exception("Failure");
96 log.println();
97
98 } else if (args[0].equals("client")) {
99
100 if (args.length < 2) {
101 log.println("No host specified: terminating.");
102 return;
103 }
104 String ip = args[1];
105 if (args.length > 2)
106 TEST_PORT = Integer.parseInt(args[2]);
107 InetAddress ia = InetAddress.getByName(ip);
108 InetSocketAddress isa = new InetSocketAddress(ia, TEST_PORT);
109 Client client = new Client(isa);
110 client.start();
111 if (client.finish(FINISH_TIME) == 0)
112 throw new Exception("Failure");
113 log.println();
114
115 } else {
116 System.out.println("Usage note:");
117 System.out.println("java SelectorTest [server] [client <host>] [<port>]");
118 System.out.println("No arguments runs both client and server in separate threads using the default port of 31452.");
119 System.out.println("client runs the client on this machine and connects to the server specified.");
120 System.out.println("server runs the server on localhost.");
121 }
122 }
123
124 static class Client extends TestThread {
125 InetSocketAddress isa;
126 Client(InetSocketAddress isa) {
127 super("Client", SelectorTest.log);
128 this.isa = isa;
129 }
130
131 public void go() throws Exception {
132 log.println("starting client...");
133 for (int i=0; i<NUM_CLIENTS; i++)
134 clientList.add(new RemoteEntity(i, isa, log));
135
136 Collections.shuffle(clientList);
137
138 log.println("created "+NUM_CLIENTS+" clients");
139 do {
140 for (Iterator i = clientList.iterator(); i.hasNext(); ) {
141 RemoteEntity re = (RemoteEntity) i.next();
142 if (re.cycle()) {
143 i.remove();
144 }
145 }
146 Collections.shuffle(clientList);
147 } while (clientList.size() > 0);
148 }
149 }
150
151 static class Server extends TestThread {
152 private final ServerSocketChannel ssc;
153 private List socketList = new ArrayList();
154 private ServerSocket ss;
155 private int connectionsAccepted = 0;
156 private Selector pollSelector;
157 private Selector acceptSelector;
158 private Set pkeys;
159 private Set pskeys;
160
161 Server(int port) throws IOException {
162 super("Server", SelectorTest.log);
163 this.ssc = RdmaSockets.openServerSocketChannel(
164 StandardProtocolFamily.INET);
165 ssc.bind(new InetSocketAddress(InetAddress.getLocalHost(), port));
166 }
167
168 int port() {
169 return ssc.socket().getLocalPort();
170 }
171
172 public void go() throws Exception {
173 log.println("starting server...");
174 acceptSelector = RdmaSockets.openSelector();
175 pollSelector = RdmaSockets.openSelector();
176 pkeys = pollSelector.keys();
177 pskeys = pollSelector.selectedKeys();
178 Set readyKeys = acceptSelector.selectedKeys();
179 RequestHandler rh = new RequestHandler(pollSelector, log);
180 Thread requestThread = new Thread(rh);
181
182 requestThread.start();
183
184 ssc.configureBlocking(false);
185 SelectionKey acceptKey = ssc.register(acceptSelector,
186 SelectionKey.OP_ACCEPT);
187 while(connectionsAccepted < SelectorTest.NUM_CLIENTS) {
188 int keysAdded = acceptSelector.select(100);
189 if (keysAdded > 0) {
190 Iterator i = readyKeys.iterator();
191 while(i.hasNext()) {
192 SelectionKey sk = (SelectionKey)i.next();
193 i.remove();
194 ServerSocketChannel nextReady =
195 (ServerSocketChannel)sk.channel();
196 SocketChannel sc = nextReady.accept();
197 connectionsAccepted++;
198 if (sc != null) {
199 sc.configureBlocking(false);
200 synchronized (pkeys) {
201 sc.register(pollSelector, SelectionKey.OP_READ);
202 }
203 } else {
204 throw new RuntimeException(
205 "Socket does not support Channels");
206 }
207 }
208 }
209 }
210 acceptKey.cancel();
211 requestThread.join();
212 acceptSelector.close();
213 pollSelector.close();
214 }
215 }
216 }
217
218 class RemoteEntity {
219 private static Random rnd = new Random();
220 int id;
221 ByteBuffer data;
222 int dataWrittenIndex;
223 int totalDataLength;
224 boolean initiated = false;
225 boolean connected = false;
226 boolean written = false;
227 boolean acked = false;
228 boolean closed = false;
229 private SocketChannel sc;
230 ByteBuffer ackBuffer;
231 PrintStream log;
232 InetSocketAddress server;
233
234 RemoteEntity(int id, InetSocketAddress server, PrintStream log)
235 throws Exception
236 {
237 int connectFailures = 0;
238 this.id = id;
239 this.log = log;
240 this.server = server;
241
242 sc = RdmaSockets.openSocketChannel(StandardProtocolFamily.INET);
243 sc.configureBlocking(false);
244
245 // Prepare the data buffer to write out from this entity
246 // Let's use both slow and fast buffers
247 if (rnd.nextBoolean())
248 data = ByteBuffer.allocateDirect(100);
249 else
250 data = ByteBuffer.allocate(100);
251 String number = Integer.toString(id);
252 if (number.length() == 1)
253 number = "0"+number;
254 String source = "Testing from " + number;
255 data.put(source.getBytes("8859_1"));
256 data.flip();
257 totalDataLength = source.length();
258
259 // Allocate an ack buffer
260 ackBuffer = ByteBuffer.allocateDirect(10);
261 }
262
263 private void reset() throws Exception {
264 sc.close();
265 sc = RdmaSockets.openSocketChannel(StandardProtocolFamily.INET);
266 sc.configureBlocking(false);
267 }
268
269 private void connect() throws Exception {
270 try {
271 connected = sc.connect(server);
272 initiated = true;
273 } catch (ConnectException e) {
274 initiated = false;
275 reset();
276 }
277 }
278
279 private void finishConnect() throws Exception {
280 try {
281 connected = sc.finishConnect();
282 } catch (IOException e) {
283 initiated = false;
284 reset();
285 }
286 }
287
288 int id() {
289 return id;
290 }
291
292 boolean cycle() throws Exception {
293 if (!initiated)
294 connect();
295 else if (!connected)
296 finishConnect();
297 else if (!written)
298 writeCycle();
299 else if (!acked)
300 ackCycle();
301 else if (!closed)
302 close();
303 return closed;
304 }
305
306 private void ackCycle() throws Exception {
307 //log.println("acking from "+id);
308 int bytesRead = sc.read(ackBuffer);
309 if (bytesRead > 0) {
310 acked = true;
311 }
312 }
313
314 private void close() throws Exception {
315 sc.close();
316 closed = true;
317 }
318
319 private void writeCycle() throws Exception {
320 log.println("writing from "+id);
321 int numBytesToWrite = rnd.nextInt(10)+1;
322 int newWriteTarget = dataWrittenIndex + numBytesToWrite;
323 if (newWriteTarget > totalDataLength)
324 newWriteTarget = totalDataLength;
325 data.limit(newWriteTarget);
326 int bytesWritten = sc.write(data);
327 if (bytesWritten > 0)
328 dataWrittenIndex += bytesWritten;
329 if (dataWrittenIndex == totalDataLength) {
330 written = true;
331 sc.socket().shutdownOutput();
332 }
333 }
334
335 }
336
337
338 class RequestHandler implements Runnable {
339 private static Random rnd = new Random();
340 private Selector selector;
341 private int connectionsHandled = 0;
342 private HashMap dataBin = new HashMap();
343 PrintStream log;
344
345 public RequestHandler(Selector selector, PrintStream log) {
346 this.selector = selector;
347 this.log = log;
348 }
349
350 public void run() {
351 log.println("starting request handler...");
352 int connectionsAccepted = 0;
353
354 Set nKeys = selector.keys();
355 Set readyKeys = selector.selectedKeys();
356
357 try {
358 while(connectionsHandled < SelectorTest.NUM_CLIENTS) {
359 int numKeys = selector.select(100);
360
361 // Process channels with data
362 synchronized (nKeys) {
363 if (readyKeys.size() > 0) {
364 Iterator i = readyKeys.iterator();
365 while(i.hasNext()) {
366 SelectionKey sk = (SelectionKey)i.next();
367 i.remove();
368 SocketChannel sc = (SocketChannel)sk.channel();
369 if (sc.isOpen())
370 read(sk, sc);
371 }
372 }
373 }
374
375 // Give other threads a chance to run
376 if (numKeys == 0) {
377 try {
378 Thread.sleep(1);
379 } catch (Exception x) {}
380 }
381 }
382 } catch (Exception e) {
383 log.println("Unexpected error 1: "+e);
384 e.printStackTrace();
385 }
386 }
387
388 private void read(SelectionKey sk, SocketChannel sc) throws Exception {
389 ByteBuffer bin = (ByteBuffer)dataBin.get(sc);
390 if (bin == null) {
391 if (rnd.nextBoolean())
392 bin = ByteBuffer.allocateDirect(100);
393 else
394 bin = ByteBuffer.allocate(100);
395 dataBin.put(sc, bin);
396 }
397
398 int bytesRead = 0;
399 do {
400 bytesRead = sc.read(bin);
401 } while(bytesRead > 0);
402
403 if (bytesRead == -1) {
404 sk.interestOps(0);
405 bin.flip();
406 int size = bin.limit();
407 byte[] data = new byte[size];
408 for(int j=0; j<size; j++)
409 data[j] = bin.get();
410 String message = new String(data, "8859_1");
411 connectionsHandled++;
412 acknowledge(sc);
413 log.println("Received >>>"+message + "<<<");
414 log.println("Handled: "+connectionsHandled);
415 }
416 }
417
418 private void acknowledge(SocketChannel sc) throws Exception {
419 ByteBuffer ackBuffer = ByteBuffer.allocateDirect(10);
420 String s = "ack";
421 ackBuffer.put(s.getBytes("8859_1"));
422 ackBuffer.flip();
423 int bytesWritten = 0;
424 while(bytesWritten == 0) {
425 bytesWritten += sc.write(ackBuffer);
426 }
427 sc.close();
428 }
429 }