1 /*

     2  * Copyright (c) 2017, 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 package jdk.internal.net.http;

    25 

    26 import java.io.IOException;

    27 import java.io.InputStream;

    28 import java.io.OutputStream;

    29 import java.io.UncheckedIOException;

    30 import java.net.ServerSocket;

    31 import java.net.Socket;

    32 import java.net.URI;

    33 import java.nio.ByteBuffer;

    34 import java.nio.channels.SelectionKey;

    35 import java.util.Random;

    36 import java.util.concurrent.CountDownLatch;

    37 import java.util.concurrent.TimeUnit;

    38 import java.util.concurrent.atomic.AtomicInteger;

    39 import java.util.concurrent.atomic.AtomicLong;

    40 import java.net.http.HttpClient;

    41 import java.net.http.HttpRequest;

    42 import java.net.http.HttpResponse;

    43 import jdk.internal.net.http.websocket.RawChannel;

    44 import jdk.internal.net.http.websocket.WebSocketRequest;

    45 import org.testng.annotations.Test;

    46 import static java.net.http.HttpResponse.BodyHandler.discard;

    47 import static org.testng.Assert.assertEquals;

    48 

    49 /*

    50  * This test exercises mechanics of _independent_ reads and writes on the

    51  * RawChannel. It verifies that the underlying implementation can manage more

    52  * than a single type of notifications at the same time.

    53  */

    54 public class RawChannelTest {

    55 

    56     private final AtomicLong clientWritten = new AtomicLong();

    57     private final AtomicLong serverWritten = new AtomicLong();

    58     private final AtomicLong clientRead = new AtomicLong();

    59     private final AtomicLong serverRead = new AtomicLong();

    60 

    61     /*

    62      * Since at this level we don't have any control over the low level socket

    63      * parameters, this latch ensures a write to the channel will stall at least

    64      * once (socket's send buffer filled up).

    65      */

    66     private final CountDownLatch writeStall = new CountDownLatch(1);

    67     private final CountDownLatch initialWriteStall = new CountDownLatch(1);

    68 

    69     /*

    70      * This one works similarly by providing means to ensure a read from the

    71      * channel will stall at least once (no more data available on the socket).

    72      */

    73     private final CountDownLatch readStall = new CountDownLatch(1);

    74     private final CountDownLatch initialReadStall = new CountDownLatch(1);

    75 

    76     private final AtomicInteger writeHandles = new AtomicInteger();

    77     private final AtomicInteger readHandles = new AtomicInteger();

    78 

    79     private final CountDownLatch exit = new CountDownLatch(1);

    80 

    81     @Test

    82     public void test() throws Exception {

    83         try (ServerSocket server = new ServerSocket(0)) {

    84             int port = server.getLocalPort();

    85             new TestServer(server).start();

    86 

    87             final RawChannel chan = channelOf(port);

    88             print("RawChannel is %s", String.valueOf(chan));

    89             initialWriteStall.await();

    90 

    91             // It's very important not to forget the initial bytes, possibly

    92             // left from the HTTP thingy

    93             int initialBytes = chan.initialByteBuffer().remaining();

    94             print("RawChannel has %s initial bytes", initialBytes);

    95             clientRead.addAndGet(initialBytes);

    96 

    97             // tell the server we have read the initial bytes, so

    98             // that it makes sure there is something for us to

    99             // read next in case the initialBytes have already drained the

   100             // channel dry.

   101             initialReadStall.countDown();

   102 

   103             chan.registerEvent(new RawChannel.RawEvent() {

   104 

   105                 private final ByteBuffer reusableBuffer = ByteBuffer.allocate(32768);

   106 

   107                 @Override

   108                 public int interestOps() {

   109                     return SelectionKey.OP_WRITE;

   110                 }

   111 

   112                 @Override

   113                 public void handle() {

   114                     int i = writeHandles.incrementAndGet();

   115                     print("OP_WRITE #%s", i);

   116                     if (i > 3) { // Fill up the send buffer not more than 3 times

   117                         try {

   118                             chan.shutdownOutput();

   119                         } catch (IOException e) {

   120                             e.printStackTrace();

   121                         }

   122                         return;

   123                     }

   124                     long total = 0;

   125                     try {

   126                         long n;

   127                         do {

   128                             ByteBuffer[] array = {reusableBuffer.slice()};

   129                             n = chan.write(array, 0, 1);

   130                             total += n;

   131                         } while (n > 0);

   132                         print("OP_WRITE clogged SNDBUF with %s bytes", total);

   133                         clientWritten.addAndGet(total);

   134                         chan.registerEvent(this);

   135                         writeStall.countDown(); // signal send buffer is full

   136                     } catch (IOException e) {

   137                         throw new UncheckedIOException(e);

   138                     }

   139                 }

   140             });

   141 

   142             chan.registerEvent(new RawChannel.RawEvent() {

   143 

   144                 @Override

   145                 public int interestOps() {

   146                     return SelectionKey.OP_READ;

   147                 }

   148 

   149                 @Override

   150                 public void handle() {

   151                     int i = readHandles.incrementAndGet();

   152                     print("OP_READ #%s", i);

   153                     ByteBuffer read = null;

   154                     long total = 0;

   155                     while (true) {

   156                         try {

   157                             read = chan.read();

   158                         } catch (IOException e) {

   159                             e.printStackTrace();

   160                         }

   161                         if (read == null) {

   162                             print("OP_READ EOF");

   163                             break;

   164                         } else if (!read.hasRemaining()) {

   165                             print("OP_READ stall");

   166                             try {

   167                                 chan.registerEvent(this);

   168                             } catch (IOException e) {

   169                                 e.printStackTrace();

   170                             }

   171                             readStall.countDown();

   172                             break;

   173                         }

   174                         int r = read.remaining();

   175                         total += r;

   176                         clientRead.addAndGet(r);

   177                     }

   178                     print("OP_READ read %s bytes (%s total)", total, clientRead.get());

   179                 }

   180             });

   181             exit.await(); // All done, we need to compare results:

   182             assertEquals(clientRead.get(), serverWritten.get());

   183             assertEquals(serverRead.get(), clientWritten.get());

   184         }

   185     }

   186 

   187     private static RawChannel channelOf(int port) throws Exception {

   188         URI uri = URI.create("http://127.0.0.1:" + port + "/");

   189         print("raw channel to %s", uri.toString());

   190         HttpRequest req = HttpRequest.newBuilder(uri).build();

   191         // Switch on isWebSocket flag to prevent the connection from

   192         // being returned to the pool.

   193         ((WebSocketRequest)req).isWebSocket(true);

   194         HttpClient client = HttpClient.newHttpClient();

   195         try {

   196             HttpResponse<?> r = client.send(req, discard());

   197             r.body();

   198             return ((HttpResponseImpl) r).rawChannel();

   199         } finally {

   200            // Need to hold onto the client until the RawChannel is

   201            // created. This would not be needed if we had created

   202            // a WebSocket, but here we are fiddling directly

   203            // with the internals of HttpResponseImpl!

   204            java.lang.ref.Reference.reachabilityFence(client);

   205         }

   206     }

   207 

   208     private class TestServer extends Thread { // Powered by Slowpokes

   209 

   210         private final ServerSocket server;

   211 

   212         TestServer(ServerSocket server) throws IOException {

   213             this.server = server;

   214         }

   215 

   216         @Override

   217         public void run() {

   218             try (Socket s = server.accept()) {

   219                 InputStream is = s.getInputStream();

   220                 OutputStream os = s.getOutputStream();

   221 

   222                 processHttp(is, os);

   223 

   224                 Thread reader = new Thread(() -> {

   225                     try {

   226                         long n = readSlowly(is);

   227                         print("Server read %s bytes", n);

   228                         serverRead.addAndGet(n);

   229                         s.shutdownInput();

   230                     } catch (Exception e) {

   231                         e.printStackTrace();

   232                     }

   233                 });

   234 

   235                 Thread writer = new Thread(() -> {

   236                     try {

   237                         long n = writeSlowly(os);

   238                         print("Server written %s bytes", n);

   239                         serverWritten.addAndGet(n);

   240                         s.shutdownOutput();

   241                     } catch (Exception e) {

   242                         e.printStackTrace();

   243                     }

   244                 });

   245 

   246                 reader.start();

   247                 writer.start();

   248 

   249                 reader.join();

   250                 writer.join();

   251             } catch (Exception e) {

   252                 e.printStackTrace();

   253             } finally {

   254                 exit.countDown();

   255             }

   256         }

   257 

   258         private void processHttp(InputStream is, OutputStream os)

   259                 throws IOException

   260         {

   261             os.write("HTTP/1.1 200 OK\r\nContent-length: 0\r\n\r\n".getBytes());

   262 

   263             // write some initial bytes

   264             byte[] initial = byteArrayOfSize(1024);

   265             os.write(initial);

   266             os.flush();

   267             serverWritten.addAndGet(initial.length);

   268             initialWriteStall.countDown();

   269 

   270             byte[] buf = new byte[1024];

   271             String s = "";

   272             while (true) {

   273                 int n = is.read(buf);

   274                 if (n <= 0) {

   275                     throw new RuntimeException("Unexpected end of request");

   276                 }

   277                 s = s + new String(buf, 0, n);

   278                 if (s.contains("\r\n\r\n")) {

   279                     break;

   280                 }

   281             }

   282         }

   283 

   284         private long writeSlowly(OutputStream os) throws Exception {

   285             byte[] first = byteArrayOfSize(1024);

   286             long total = first.length;

   287             os.write(first);

   288             os.flush();

   289 

   290             // wait until initial bytes were read

   291             initialReadStall.await();

   292 

   293             // make sure there is something to read, otherwise readStall

   294             // will never be counted down.

   295             first = byteArrayOfSize(1024);

   296             os.write(first);

   297             os.flush();

   298             total += first.length;

   299 

   300             // Let's wait for the signal from the raw channel that its read has

   301             // stalled, and then continue sending a bit more stuff

   302             readStall.await();

   303             for (int i = 0; i < 32; i++) {

   304                 byte[] b = byteArrayOfSize(1024);

   305                 os.write(b);

   306                 os.flush();

   307                 total += b.length;

   308                 TimeUnit.MILLISECONDS.sleep(1);

   309             }

   310             return total;

   311         }

   312 

   313         private long readSlowly(InputStream is) throws Exception {

   314             // Wait for the raw channel to fill up its send buffer

   315             writeStall.await();

   316             long overall = 0;

   317             byte[] array = new byte[1024];

   318             for (int n = 0; n != -1; n = is.read(array)) {

   319                 TimeUnit.MILLISECONDS.sleep(1);

   320                 overall += n;

   321             }

   322             return overall;

   323         }

   324     }

   325 

   326     private static void print(String format, Object... args) {

   327         System.out.println(Thread.currentThread() + ": " + String.format(format, args));

   328     }

   329 

   330     private static byte[] byteArrayOfSize(int bound) {

   331         return new byte[new Random().nextInt(1 + bound)];

   332     }

   333 }