OutboundMessageQueue refactorings:

- extracting a method
- renaming methods

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/VertxHttp2Stream.java

@@ -15,7 +15,6 @@
 import io.netty.buffer.Unpooled;
 import io.netty.channel.EventLoop;
 import io.netty.handler.codec.http2.EmptyHttp2Headers;
-import io.netty.handler.codec.http2.Http2Exception;
 import io.netty.handler.codec.http2.Http2Headers;
 import io.netty.handler.codec.http2.Http2Stream;
 import io.netty.util.concurrent.FutureListener;
@@ -93,15 +92,15 @@ public boolean test(MessageWrite msg) {
         }
       }
       @Override
-      protected void disposeMessage(MessageWrite messageWrite) {
+      protected void handleDispose(MessageWrite messageWrite) {
         Throwable cause = failure;
         if (cause == null) {
           cause = HttpUtils.STREAM_CLOSED_EXCEPTION;
         }
         messageWrite.cancel(cause);
       }
       @Override
-      protected void afterDrain() {
+      protected void handleDrained() {
         context.emit(VertxHttp2Stream.this, VertxHttp2Stream::handleWriteQueueDrained);
       }
     };
@@ -156,7 +155,7 @@ void onData(Buffer data) {
   void onWritabilityChanged() {
     writable = !writable;
     if (writable) {
-      outboundQueue.drain();
+      outboundQueue.tryDrain();
     }
   }
 

vertx-core/src/main/java/io/vertx/core/internal/concurrent/OutboundMessageChannel.java

@@ -20,28 +20,30 @@ public class OutboundMessageChannel<M> implements Predicate<M> {
   private volatile boolean eventuallyClosed;
 
   // State accessed exclusively by the event loop thread
-  private boolean overflow;
+  private boolean overflow; // Indicates channel ownership
+  private int draining = 0; // Indicates drain in progress
   private boolean closed;
-  private int reentrant = 0;
 
   /**
    * Create a channel.
    *
    * @param eventLoop the channel event-loop
    */
-  public OutboundMessageChannel(EventLoop eventLoop, Predicate<M> predicate) {
+  public OutboundMessageChannel(EventLoop eventLoop) {
     this.eventLoop = eventLoop;
-    this.messageChannel = new MessageChannel.MpSc<>(predicate);
+    this.messageChannel = new MessageChannel.MpSc<>(this);
   }
 
   /**
    * Create a channel.
    *
    * @param eventLoop the channel event-loop
+   * @param lowWaterMark the low-water mark, must be positive
+   * @param highWaterMark the high-water mark, must be greater than the low-water mark
    */
-  public OutboundMessageChannel(EventLoop eventLoop) {
+  public OutboundMessageChannel(EventLoop eventLoop, int lowWaterMark, int highWaterMark) {
     this.eventLoop = eventLoop;
-    this.messageChannel = new MessageChannel.MpSc<>(this);
+    this.messageChannel = new MessageChannel.MpSc<>(this, lowWaterMark, highWaterMark);
   }
 
   @Override
@@ -68,68 +70,76 @@ public final boolean write(M message) {
     int flags;
     if (inEventLoop) {
       if (closed) {
-        disposeMessage(message);
+        handleDispose(message);
         return true;
       }
-      reentrant++;
-      try {
-        flags = messageChannel.add(message);
-        if ((flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0) {
-          flags = messageChannel.drain();
-          overflow |= (flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0;
-          if ((flags & MessageChannel.WRITABLE_MASK) != 0) {
-            handleDrained(numberOfUnwritableSignals(flags));
-          }
-        }
-      } finally {
-        reentrant--;
-      }
-      if (reentrant == 0 && closed) {
-        releaseMessages();
+      flags = messageChannel.add(message);
+      if (draining == 0 && (flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0) {
+        flags = drainMessageQueue();
       }
     } else {
       if (eventuallyClosed) {
-        disposeMessage(message);
+        handleDispose(message);
         return true;
       }
       flags = messageChannel.add(message);
       if ((flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0) {
-        eventLoop.execute(this::drainMessageChannel);
+        eventLoop.execute(this::drain);
       }
     }
+    int val;
     if ((flags & MessageChannel.UNWRITABLE_MASK) != 0) {
-      int val = numberOfUnwritableSignals.incrementAndGet();
-      return val <= 0;
+      val = numberOfUnwritableSignals.incrementAndGet();
     } else {
-      return numberOfUnwritableSignals.get() <= 0;
+      val = numberOfUnwritableSignals.get();
     }
+    return val <= 0;
   }
 
   /**
-   * Attempt to drain the queue Drain the queue.
+   * Synchronous message queue drain.
    */
-  public void drain() {
-    assert(eventLoop.inEventLoop());
-    if (overflow) {
-      startDraining();
-      reentrant++;
-      int flags;
-      try {
-        flags = messageChannel.drain();
-        overflow = (flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0;
-        if ((flags & MessageChannel.WRITABLE_MASK) != 0) {
-          handleDrained(numberOfUnwritableSignals(flags));
-        }
-      } finally {
-        reentrant--;
+  private int drainMessageQueue() {
+    draining++;
+    try {
+      int flags = messageChannel.drain();
+      overflow |= (flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0;
+      if ((flags & MessageChannel.WRITABLE_MASK) != 0) {
+        handleDrained(numberOfUnwritableSignals(flags));
       }
-      stopDraining();
-      if (reentrant == 0 && closed) {
+      return flags;
+    } finally {
+      draining--;
+      if (draining == 0 && closed) {
         releaseMessages();
       }
     }
   }
 
+  private void drain() {
+    if (closed) {
+      return;
+    }
+    assert(draining == 0);
+    startDraining();
+    drainMessageQueue();
+    stopDraining();
+  }
+
+  /**
+   * Attempts to drain the queue.
+   */
+  public final boolean tryDrain() {
+    assert(eventLoop.inEventLoop());
+    if (overflow) {
+      overflow = false;
+      drain();
+      return true;
+    } else {
+      return false;
+    }
+  }
+
   /**
    * Close the queue.
    */
@@ -140,52 +150,30 @@ public final void close() {
     }
     closed = true;
     eventuallyClosed = true;
-    if (reentrant > 0) {
+    if (draining > 0) {
       return;
     }
     releaseMessages();
   }
 
-  private void drainMessageChannel() {
-    if (closed) {
-      return;
-    }
-    startDraining();
-    reentrant++;
-    int flags;
-    try {
-      flags = messageChannel.drain();
-      overflow = (flags & MessageChannel.DRAIN_REQUIRED_MASK) != 0;
-      if ((flags & MessageChannel.WRITABLE_MASK) != 0) {
-        handleDrained(numberOfUnwritableSignals(flags));
-      }
-    } finally {
-      reentrant--;
-    }
-    stopDraining();
-    if (reentrant == 0 && closed) {
-      releaseMessages();
-    }
-  }
-
   private void handleDrained(int numberOfSignals) {
     int val = numberOfUnwritableSignals.addAndGet(-numberOfSignals);
     if ((val + numberOfSignals) > 0 && val <= 0) {
-      afterDrain();
+      eventLoop.execute(this::handleDrained);
     }
   }
 
   private void releaseMessages() {
     List<M> messages = messageChannel.clear();
     for (M elt : messages) {
-      disposeMessage(elt);
+      handleDispose(elt);
     }
   }
 
   /**
    * Called when the channel becomes writable again.
    */
-  protected void afterDrain() {
+  protected void handleDrained() {
   }
 
   protected void startDraining() {
@@ -199,6 +187,6 @@ protected void stopDraining() {
    *
    * @param msg the message
    */
-  protected void disposeMessage(M msg) {
+  protected void handleDispose(M msg) {
   }
 }

vertx-core/src/main/java/io/vertx/core/net/impl/VertxConnection.java

@@ -210,7 +210,7 @@ protected void handleMessage(Object msg) {
   void channelWritabilityChanged() {
     channelWritable = chctx.channel().isWritable();
     if (channelWritable) {
-      messageQueue.drain();
+      messageQueue.tryDrain();
     }
   }
 
@@ -468,7 +468,7 @@ public boolean test(MessageWrite msg) {
     }
 
     @Override
-    protected void disposeMessage(MessageWrite write) {
+    protected void handleDispose(MessageWrite write) {
       write.cancel(CLOSED_EXCEPTION);
     }
 
@@ -486,7 +486,7 @@ protected void stopDraining() {
     }
 
     @Override
-    protected void afterDrain() {
+    protected void handleDrained() {
       VertxConnection.this.handleWriteQueueDrained();
     }
   }

vertx-core/src/main/java/io/vertx/core/streams/impl/MessageChannel.java

@@ -435,6 +435,7 @@ public static class MpSc<E> extends MessageChannel<E> {
     public MpSc(Predicate<E> consumer) {
       this(consumer, DEFAULT_LOW_WATER_MARK, DEFAULT_HIGH_WATER_MARK);
     }
+
     public MpSc(Predicate<E> consumer, int lowWaterMark, int highWaterMark) {
       super(PlatformDependent.newMpscQueue(), consumer, lowWaterMark, highWaterMark);
     }