1 /* 2 * Copyright (c) 2000, 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 socketchannel vector IO (use -Dseed=X to set PRNG seed) 27 * @requires (os.family == "linux") 28 * @library .. /test/lib 29 * @build jdk.test.lib.RandomFactory 30 * @build RsocketTest 31 * @run main/othervm -Djava.net.preferIPv4Stack=true VectorIO 32 * @key randomness 33 */ 34 import java.io.*; 35 import java.net.*; 36 import java.nio.*; 37 import java.nio.channels.*; 38 import java.util.*; 39 import jdk.test.lib.RandomFactory; 40 import jdk.net.Sockets; 41 42 public class VectorIO { 43 44 private static Random generator = RandomFactory.getRandom(); 45 46 static int testSize; 47 48 // whether to use the write/read variant with a length parameter 49 static boolean setLength; 50 51 public static void main(String[] args) throws Exception { 52 if (!RsocketTest.isRsocketAvailable()) 53 return; 54 55 testSize = 1; 56 setLength = false; 57 runTest(); 58 for(int i = 15; i < 18; i++) { 59 testSize = i; 60 setLength = !setLength; 61 runTest(); 62 } 63 } 64 65 static void runTest() throws Exception { 66 System.err.println("Length " + testSize); 67 Server sv = new Server(testSize); 68 sv.start(); 69 bufferTest(sv.port()); 70 if (sv.finish(8000) == 0) 71 throw new Exception("Failed: Length = " + testSize); 72 } 73 74 static void bufferTest(int port) throws Exception { 75 ByteBuffer[] bufs = new ByteBuffer[testSize]; 76 long total = 0L; 77 for(int i = 0; i < testSize; i++) { 78 String source = "buffer" + i; 79 if (generator.nextBoolean()) 80 bufs[i] = ByteBuffer.allocateDirect(source.length()); 81 else 82 bufs[i] = ByteBuffer.allocate(source.length()); 83 84 bufs[i].put(source.getBytes("8859_1")); 85 bufs[i].flip(); 86 total += bufs[i].remaining(); 87 } 88 89 ByteBuffer[] bufsPlus1 = new ByteBuffer[bufs.length + 1]; 90 System.arraycopy(bufs, 0, bufsPlus1, 0, bufs.length); 91 92 // Get a connection to the server 93 InetAddress lh = InetAddress.getLocalHost(); 94 InetSocketAddress isa = new InetSocketAddress(lh, port); 95 SocketChannel sc = Sockets.openRdmaSocketChannel(); 96 sc.connect(isa); 97 sc.configureBlocking(generator.nextBoolean()); 98 99 // Write the data out 100 long rem = total; 101 while (rem > 0L) { 102 long bytesWritten; 103 if (setLength) { 104 bytesWritten = sc.write(bufsPlus1, 0, bufs.length); 105 } else { 106 bytesWritten = sc.write(bufs); 107 } 108 if (bytesWritten == 0) { 109 if (sc.isBlocking()) { 110 throw new RuntimeException("write did not block"); 111 } else { 112 System.err.println("Non-blocking write() wrote zero bytes"); 113 } 114 Thread.sleep(50); 115 } else { 116 rem -= bytesWritten; 117 } 118 } 119 120 // Clean up 121 sc.close(); 122 } 123 124 static class Server 125 extends TestThread 126 { 127 final int testSize; 128 final ServerSocketChannel ssc; 129 130 Server(int testSize) throws IOException { 131 super("Server " + testSize); 132 this.testSize = testSize; 133 this.ssc = Sockets.openRdmaServerSocketChannel().bind(new InetSocketAddress(0)); 134 } 135 136 int port() { 137 return ssc.socket().getLocalPort(); 138 } 139 140 void go() throws Exception { 141 bufferTest(); 142 } 143 144 void bufferTest() throws Exception { 145 long total = 0L; 146 ByteBuffer[] bufs = new ByteBuffer[testSize]; 147 for(int i=0; i<testSize; i++) { 148 String source = "buffer" + i; 149 if (generator.nextBoolean()) 150 bufs[i] = ByteBuffer.allocateDirect(source.length()); 151 else 152 bufs[i] = ByteBuffer.allocate(source.length()); 153 total += bufs[i].capacity(); 154 } 155 156 ByteBuffer[] bufsPlus1 = new ByteBuffer[bufs.length + 1]; 157 System.arraycopy(bufs, 0, bufsPlus1, 0, bufs.length); 158 159 // Get a connection from client 160 SocketChannel sc = null; 161 162 try { 163 164 ssc.configureBlocking(false); 165 166 for (;;) { 167 sc = ssc.accept(); 168 if (sc != null) { 169 System.err.println("accept() succeeded"); 170 break; 171 } 172 Thread.sleep(50); 173 } 174 175 sc.configureBlocking(generator.nextBoolean()); 176 177 // Read data into multiple buffers 178 long avail = total; 179 while (avail > 0) { 180 long bytesRead; 181 if (setLength) { 182 bytesRead = sc.read(bufsPlus1, 0, bufs.length); 183 } else { 184 bytesRead = sc.read(bufs); 185 } 186 if (bytesRead < 0) 187 break; 188 if (bytesRead == 0) { 189 if (sc.isBlocking()) { 190 throw new RuntimeException("read did not block"); 191 } else { 192 System.err.println 193 ("Non-blocking read() read zero bytes"); 194 } 195 Thread.sleep(50); 196 } 197 avail -= bytesRead; 198 } 199 200 // Check results 201 for(int i=0; i<testSize; i++) { 202 String expected = "buffer" + i; 203 bufs[i].flip(); 204 int size = bufs[i].capacity(); 205 byte[] data = new byte[size]; 206 for(int j=0; j<size; j++) 207 data[j] = bufs[i].get(); 208 String message = new String(data, "8859_1"); 209 if (!message.equals(expected)) 210 throw new Exception("Wrong data: Got " 211 + message + ", expected " 212 + expected); 213 } 214 215 } finally { 216 // Clean up 217 ssc.close(); 218 if (sc != null) 219 sc.close(); 220 } 221 222 } 223 224 } 225 }