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.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.internal.net.http.websocket;

    27 

    28 import jdk.internal.net.http.common.Utils;

    29 import jdk.internal.vm.annotation.Stable;

    30 

    31 import java.io.IOException;

    32 import java.nio.ByteBuffer;

    33 import java.nio.CharBuffer;

    34 import java.util.concurrent.CompletableFuture;

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

    36 import java.util.function.BiConsumer;

    37 

    38 /*

    39  * A FIFO message storage facility.

    40  *

    41  * The queue supports at most one consumer and an arbitrary number of producers.

    42  * Methods `peek`, `remove` and `isEmpty` must not be invoked concurrently.

    43  * Methods `addText`, `addBinary`, `addPing`, `addPong` and `addClose` may be

    44  * invoked concurrently.

    45  *

    46  * This queue is of a bounded size. The queue pre-allocates array of the said

    47  * size and fills it with `Message` elements. The resulting structure never

    48  * changes. This allows to avoid re-allocation and garbage collection of

    49  * elements and arrays thereof. For this reason `Message` elements are never

    50  * returned from the `peek` method. Instead their components passed to the

    51  * provided callback.

    52  *

    53  * The queue consists of:

    54  *

    55  *   - a ring array of n + 1 `Message` elements

    56  *   - indexes H and T denoting the head and the tail elements of the queue

    57  *     respectively

    58  *

    59  * Each `Message` element contains a boolean flag. This flag is an auxiliary

    60  * communication between the producers and the consumer. The flag shows

    61  * whether or not the element is ready to be consumed (peeked at, removed). The

    62  * flag is required since updating an element involves many fields and thus is

    63  * not an atomic action. An addition to the queue happens in two steps:

    64  *

    65  * # Step 1

    66  *

    67  * Producers race with each other to secure an index for the element they add.

    68  * T is atomically advanced [1] only if the advanced value doesn't equal to H

    69  * (a producer doesn't bump into the head of the queue).

    70  *

    71  * # Step 2

    72  *

    73  * Once T is advanced in the previous step, the producer updates the message

    74  * fields of the element at the previous value of T and then sets the flag of

    75  * this element.

    76  *

    77  * A removal happens in a single step. The consumer gets the element at index H.

    78  * If the flag of this element is set, the consumer clears the fields of the

    79  * element, clears the flag and finally advances H.

    80  *

    81  * ----------------------------------------------------------------------------

    82  * [1] To advance the index is to change it from i to (i + 1) % (n + 1).

    83  */

    84 public class MessageQueue {

    85 

    86     private final static boolean DEBUG = false;

    87 

    88     @Stable

    89     private final Message[] elements;

    90 

    91     private final AtomicInteger tail = new AtomicInteger();

    92     private volatile int head;

    93 

    94     public MessageQueue() {

    95         this(defaultSize());

    96     }

    97 

    98     /* Exposed for testing */

    99     protected MessageQueue(int size) {

   100         if (size < 1) {

   101             throw new IllegalArgumentException();

   102         }

   103         Message[] array = new Message[size + 1];

   104         for (int i = 0; i < array.length; i++) {

   105             array[i] = new Message();

   106         }

   107         elements = array;

   108     }

   109 

   110     private static int defaultSize() {

   111         String property = "jdk.httpclient.websocket.outputQueueMaxSize";

   112         int defaultSize = 128;

   113         String value = Utils.getNetProperty(property);

   114         int size;

   115         if (value == null) {

   116             size = defaultSize;

   117         } else {

   118             try {

   119                 size = Integer.parseUnsignedInt(value);

   120             } catch (NumberFormatException ignored) {

   121                 size = defaultSize;

   122             }

   123         }

   124         if (DEBUG) {

   125             System.out.printf("[MessageQueue] %s=%s, using size %s%n",

   126                               property, value, size);

   127         }

   128         return size;

   129     }

   130 

   131     public <T> void addText(CharBuffer message,

   132                             boolean isLast,

   133                             T attachment,

   134                             BiConsumer<? super T, ? super Throwable> action,

   135                             CompletableFuture<T> future)

   136             throws IOException

   137     {

   138         add(MessageQueue.Type.TEXT, null, message, isLast, -1, attachment,

   139             action, future);

   140     }

   141 

   142     private <T> void add(Type type,

   143                          ByteBuffer binary,

   144                          CharBuffer text,

   145                          boolean isLast,

   146                          int statusCode,

   147                          T attachment,

   148                          BiConsumer<? super T, ? super Throwable> action,

   149                          CompletableFuture<? super T> future)

   150             throws IOException

   151     {

   152         int h, currentTail, newTail;

   153         do {

   154             h = head;

   155             currentTail = tail.get();

   156             newTail = (currentTail + 1) % elements.length;

   157             if (newTail == h) {

   158                 throw new IOException("Queue full");

   159             }

   160         } while (!tail.compareAndSet(currentTail, newTail));

   161         Message t = elements[currentTail];

   162         if (t.ready) {

   163             throw new InternalError();

   164         }

   165         t.type = type;

   166         t.binary = binary;

   167         t.text = text;

   168         t.isLast = isLast;

   169         t.statusCode = statusCode;

   170         t.attachment = attachment;

   171         t.action = action;

   172         t.future = future;

   173         t.ready = true;

   174     }

   175 

   176     public <T> void addBinary(ByteBuffer message,

   177                               boolean isLast,

   178                               T attachment,

   179                               BiConsumer<? super T, ? super Throwable> action,

   180                               CompletableFuture<? super T> future)

   181             throws IOException

   182     {

   183         add(MessageQueue.Type.BINARY, message, null, isLast, -1, attachment,

   184             action, future);

   185     }

   186 

   187     public <T> void addPing(ByteBuffer message,

   188                             T attachment,

   189                             BiConsumer<? super T, ? super Throwable> action,

   190                             CompletableFuture<? super T> future)

   191             throws IOException

   192     {

   193         add(MessageQueue.Type.PING, message, null, false, -1, attachment,

   194             action, future);

   195     }

   196 

   197     public <T> void addPong(ByteBuffer message,

   198                             T attachment,

   199                             BiConsumer<? super T, ? super Throwable> action,

   200                             CompletableFuture<? super T> future)

   201             throws IOException

   202     {

   203         add(MessageQueue.Type.PONG, message, null, false, -1, attachment,

   204             action, future);

   205     }

   206 

   207     public <T> void addClose(int statusCode,

   208                              CharBuffer reason,

   209                              T attachment,

   210                              BiConsumer<? super T, ? super Throwable> action,

   211                              CompletableFuture<? super T> future)

   212             throws IOException

   213     {

   214         add(MessageQueue.Type.CLOSE, null, reason, false, statusCode,

   215             attachment, action, future);

   216     }

   217 

   218     @SuppressWarnings("unchecked")

   219     public <R, E extends Throwable> R peek(QueueCallback<R, E> callback)

   220             throws E

   221     {

   222         Message h = elements[head];

   223         if (!h.ready) {

   224             return callback.onEmpty();

   225         }

   226         Type type = h.type;

   227         switch (type) {

   228             case TEXT:

   229                 try {

   230                     return (R) callback.onText(h.text, h.isLast, h.attachment,

   231                                                h.action, h.future);

   232                 } catch (Throwable t) {

   233                     // Something unpleasant is going on here with the compiler.

   234                     // If this seemingly useless catch is omitted, the compiler

   235                     // reports an error:

   236                     //

   237                     //   java: unreported exception java.lang.Throwable;

   238                     //   must be caught or declared to be thrown

   239                     //

   240                     // My guess is there is a problem with both the type

   241                     // inference for the method AND @SuppressWarnings("unchecked")

   242                     // being working at the same time.

   243                     throw (E) t;

   244                 }

   245             case BINARY:

   246                 try {

   247                     return (R) callback.onBinary(h.binary, h.isLast, h.attachment,

   248                                                  h.action, h.future);

   249                 } catch (Throwable t) {

   250                     throw (E) t;

   251                 }

   252             case PING:

   253                 try {

   254                     return (R) callback.onPing(h.binary, h.attachment, h.action,

   255                                                h.future);

   256                 } catch (Throwable t) {

   257                     throw (E) t;

   258                 }

   259             case PONG:

   260                 try {

   261                     return (R) callback.onPong(h.binary, h.attachment, h.action,

   262                                                h.future);

   263                 } catch (Throwable t) {

   264                     throw (E) t;

   265                 }

   266             case CLOSE:

   267                 try {

   268                     return (R) callback.onClose(h.statusCode, h.text, h.attachment,

   269                                                 h.action, h.future);

   270                 } catch (Throwable t) {

   271                     throw (E) t;

   272                 }

   273             default:

   274                 throw new InternalError(String.valueOf(type));

   275         }

   276     }

   277 

   278     public boolean isEmpty() {

   279         return !elements[head].ready;

   280     }

   281 

   282     public void remove() {

   283         int currentHead = head;

   284         Message h = elements[currentHead];

   285         if (!h.ready) {

   286             throw new InternalError("Queue empty");

   287         }

   288         h.type = null;

   289         h.binary = null;

   290         h.text = null;

   291         h.attachment = null;

   292         h.action = null;

   293         h.future = null;

   294         h.ready = false;

   295         head = (currentHead + 1) % elements.length;

   296     }

   297 

   298     private enum Type {

   299 

   300         TEXT,

   301         BINARY,

   302         PING,

   303         PONG,

   304         CLOSE

   305     }

   306 

   307     /*

   308      * A callback for consuming a queue element's fields. Can return a result of

   309      * type T or throw an exception of type E. This design allows to avoid

   310      * "returning" results or "throwing" errors by updating some objects from

   311      * the outside of the methods.

   312      */

   313     public interface QueueCallback<R, E extends Throwable> {

   314 

   315         <T> R onText(CharBuffer message,

   316                      boolean isLast,

   317                      T attachment,

   318                      BiConsumer<? super T, ? super Throwable> action,

   319                      CompletableFuture<? super T> future) throws E;

   320 

   321         <T> R onBinary(ByteBuffer message,

   322                        boolean isLast,

   323                        T attachment,

   324                        BiConsumer<? super T, ? super Throwable> action,

   325                        CompletableFuture<? super T> future) throws E;

   326 

   327         <T> R onPing(ByteBuffer message,

   328                      T attachment,

   329                      BiConsumer<? super T, ? super Throwable> action,

   330                      CompletableFuture<? super T> future) throws E;

   331 

   332         <T> R onPong(ByteBuffer message,

   333                      T attachment,

   334                      BiConsumer<? super T, ? super Throwable> action,

   335                      CompletableFuture<? super T> future) throws E;

   336 

   337         <T> R onClose(int statusCode,

   338                       CharBuffer reason,

   339                       T attachment,

   340                       BiConsumer<? super T, ? super Throwable> action,

   341                       CompletableFuture<? super T> future) throws E;

   342 

   343         /* The queue is empty*/

   344         R onEmpty() throws E;

   345     }

   346 

   347     /*

   348      * A union of components of all WebSocket message types; also a node in a

   349      * queue.

   350      *

   351      * A `Message` never leaves the context of the queue, thus the reference to

   352      * it cannot be retained by anyone other than the queue.

   353      */

   354     private static class Message {

   355 

   356         private volatile boolean ready;

   357 

   358         // -- The source message fields --

   359 

   360         private Type type;

   361         private ByteBuffer binary;

   362         private CharBuffer text;

   363         private boolean isLast;

   364         private int statusCode;

   365         private Object attachment;

   366         @SuppressWarnings("rawtypes")

   367         private BiConsumer action;

   368         @SuppressWarnings("rawtypes")

   369         private CompletableFuture future;

   370     }

   371 }