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 socketchannel vector IO (use -Dseed=X to set PRNG seed)

    27  * @requires (os.family == "linux")

    28  * @library .. /test/lib /test/jdk/java/nio/channels

    29  * @build jdk.test.lib.RandomFactory

    30  * @build RsocketTest

    31  * @run main/othervm VectorIO

    32  * @key randomness

    33  */

    34 

    35 import java.io.IOException;

    36 import java.net.InetAddress;

    37 import java.net.InetSocketAddress;

    38 import java.net.StandardProtocolFamily;

    39 import java.nio.ByteBuffer;

    40 import java.nio.channels.ServerSocketChannel;

    41 import java.nio.channels.SocketChannel;

    42 import java.util.Random;

    43 import jdk.test.lib.RandomFactory;

    44 import jdk.net.RdmaSockets;

    45 

    46 import jtreg.SkippedException;

    47 

    48 public class VectorIO {

    49 

    50     private static Random generator = RandomFactory.getRandom();

    51 

    52     static int testSize;

    53 

    54     // whether to use the write/read variant with a length parameter

    55     static boolean setLength;

    56 

    57     public static void main(String[] args) throws Exception {

    58         if (!RsocketTest.isRsocketAvailable())

    59             throw new SkippedException("rsocket is not available");

    60 

    61         testSize = 1;

    62         setLength = false;

    63         runTest();

    64         for(int i = 15; i < 18; i++) {

    65             testSize = i;

    66             setLength = !setLength;

    67             runTest();

    68         }

    69     }

    70 

    71     static void runTest() throws Exception {

    72         System.err.println("Length " + testSize);

    73         Server sv = new Server(testSize);

    74         sv.start();

    75         bufferTest(sv.port());

    76         if (sv.finish(8000) == 0)

    77             throw new Exception("Failed: Length = " + testSize);

    78     }

    79 

    80     static void bufferTest(int port) throws Exception {

    81         ByteBuffer[] bufs = new ByteBuffer[testSize];

    82         long total = 0L;

    83         for(int i = 0; i < testSize; i++) {

    84             String source = "buffer" + i;

    85             if (generator.nextBoolean())

    86                 bufs[i] = ByteBuffer.allocateDirect(source.length());

    87             else

    88                 bufs[i] = ByteBuffer.allocate(source.length());

    89 

    90             bufs[i].put(source.getBytes("8859_1"));

    91             bufs[i].flip();

    92             total += bufs[i].remaining();

    93         }

    94 

    95         ByteBuffer[] bufsPlus1 = new ByteBuffer[bufs.length + 1];

    96         System.arraycopy(bufs, 0, bufsPlus1, 0, bufs.length);

    97 

    98         // Get a connection to the server

    99         InetAddress lh = InetAddress.getLocalHost();

   100         InetSocketAddress isa = new InetSocketAddress(lh, port);

   101         SocketChannel sc = RdmaSockets.openSocketChannel(StandardProtocolFamily.INET);

   102         sc.connect(isa);

   103         sc.configureBlocking(generator.nextBoolean());

   104 

   105         // Write the data out

   106         long rem = total;

   107         while (rem > 0L) {

   108             long bytesWritten;

   109             if (setLength) {

   110                 bytesWritten = sc.write(bufsPlus1, 0, bufs.length);

   111             } else {

   112                 bytesWritten = sc.write(bufs);

   113             }

   114             if (bytesWritten == 0) {

   115                 if (sc.isBlocking()) {

   116                     throw new RuntimeException("write did not block");

   117                 } else {

   118                     System.err.println("Non-blocking write() wrote zero bytes");

   119                 }

   120                 Thread.sleep(50);

   121             } else {

   122                 rem -= bytesWritten;

   123             }

   124         }

   125 

   126         // Clean up

   127         sc.close();

   128     }

   129 

   130     static class Server

   131         extends TestThread

   132     {

   133         final int testSize;

   134         final ServerSocketChannel ssc;

   135 

   136         Server(int testSize) throws IOException {

   137             super("Server " + testSize);

   138             this.testSize = testSize;

   139             this.ssc = RdmaSockets.openServerSocketChannel(

   140                 StandardProtocolFamily.INET).bind(new InetSocketAddress(0));

   141         }

   142 

   143         int port() {

   144             return ssc.socket().getLocalPort();

   145         }

   146 

   147         void go() throws Exception {

   148             bufferTest();

   149         }

   150 

   151         void bufferTest() throws Exception {

   152             long total = 0L;

   153             ByteBuffer[] bufs = new ByteBuffer[testSize];

   154             for(int i=0; i<testSize; i++) {

   155                 String source = "buffer" + i;

   156                 if (generator.nextBoolean())

   157                     bufs[i] = ByteBuffer.allocateDirect(source.length());

   158                 else

   159                     bufs[i] = ByteBuffer.allocate(source.length());

   160                 total += bufs[i].capacity();

   161             }

   162 

   163             ByteBuffer[] bufsPlus1 = new ByteBuffer[bufs.length + 1];

   164             System.arraycopy(bufs, 0, bufsPlus1, 0, bufs.length);

   165 

   166             // Get a connection from client

   167             SocketChannel sc = null;

   168 

   169             try {

   170 

   171                 ssc.configureBlocking(false);

   172 

   173                 for (;;) {

   174                     sc = ssc.accept();

   175                     if (sc != null) {

   176                         System.err.println("accept() succeeded");

   177                         break;

   178                     }

   179                     Thread.sleep(50);

   180                 }

   181 

   182                 sc.configureBlocking(generator.nextBoolean());

   183 

   184                 // Read data into multiple buffers

   185                 long avail = total;

   186                 while (avail > 0) {

   187                     long bytesRead;

   188                     if (setLength) {

   189                         bytesRead = sc.read(bufsPlus1, 0, bufs.length);

   190                     } else {

   191                         bytesRead = sc.read(bufs);

   192                     }

   193                     if (bytesRead < 0)

   194                         break;

   195                     if (bytesRead == 0) {

   196                         if (sc.isBlocking()) {

   197                             throw new RuntimeException("read did not block");

   198                         } else {

   199                             System.err.println

   200                                 ("Non-blocking read() read zero bytes");

   201                         }

   202                         Thread.sleep(50);

   203                     }

   204                     avail -= bytesRead;

   205                 }

   206 

   207                 // Check results

   208                 for(int i=0; i<testSize; i++) {

   209                     String expected = "buffer" + i;

   210                     bufs[i].flip();

   211                     int size = bufs[i].capacity();

   212                     byte[] data = new byte[size];

   213                     for(int j=0; j<size; j++)

   214                         data[j] = bufs[i].get();

   215                     String message = new String(data, "8859_1");

   216                     if (!message.equals(expected))

   217                         throw new Exception("Wrong data: Got "

   218                                             + message + ", expected "

   219                                             + expected);

   220                 }

   221 

   222             } finally {

   223                 // Clean up

   224                 ssc.close();

   225                 if (sc != null)

   226                     sc.close();

   227             }

   228 

   229         }

   230 

   231     }

   232 }