Streams WriteResult.

Motivation:

The stream writability API requires a poll after write (writeQueueFull) which can lead to a data race.

A stream write could provide the writability as a result of the write operation, improving the implementation of writability poll.

Changes:

Introduce an internal WriteResult future, aimed to eventually replace stream writeQueueFull.

This API is provided currently by HTTP server writes.

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/http2/DefaultHttp2Stream.java

@@ -269,7 +269,8 @@ public final boolean isWritable() {
   }
 
   public final Future<Void> write(WritePromise write) {
-    outboundQueue.write(write);
+    boolean writable = outboundQueue.write(write);
+    write.enqueued(writable);
     return write;
   }
 

vertx-core/src/main/java/io/vertx/core/internal/streams/WriteResult.java

@@ -0,0 +1,17 @@
+package io.vertx.core.internal.streams;
+
+import io.vertx.core.Future;
+
+/**
+ * Result of a write operation.
+ *
+ * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
+ */
+public interface WriteResult<T> extends Future<T> {
+
+  /**
+   * @return whether the stream was writable after the write operation was submitted
+   */
+  boolean isWritable();
+
+}

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

@@ -21,6 +21,9 @@ public interface MessageWrite {
    */
   void write();
 
+  default void enqueued(boolean writable) {
+  }
+
   /**
    * Cancel the write operation.
    *

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

@@ -481,7 +481,9 @@ public void cancel(Throwable cause) {
 
   // Write to channel boolean return for now is not used so avoids reading a volatile
   public final boolean writeToChannel(MessageWrite msg) {
-    return outboundMessageQueue.write(msg);
+    boolean writable = outboundMessageQueue.write(msg);
+    msg.enqueued(writable);
+    return writable;
   }
 
   /**

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

@@ -12,20 +12,32 @@
 
 import io.vertx.core.impl.future.PromiseImpl;
 import io.vertx.core.internal.ContextInternal;
+import io.vertx.core.internal.streams.WriteResult;
 
 /**
  * A write promise.
  *
  * @author <a href="mailto:julien@julienviet.com">Julien Viet</a>
  */
-public abstract class WritePromise extends PromiseImpl<Void> implements MessageWrite {
+public abstract class WritePromise extends PromiseImpl<Void> implements MessageWrite, WriteResult<Void> {
+
+  private boolean writable;
 
   public WritePromise(ContextInternal context) {
     super(context);
   }
 
+  @Override
+  public void enqueued(boolean writable) {
+    this.writable = writable;
+  }
+
   @Override
   public void cancel(Throwable cause) {
     fail(cause);
   }
+
+  public boolean isWritable() {
+    return writable;
+  }
 }

vertx-core/src/test/java/io/vertx/tests/http/HttpTest.java

@@ -32,6 +32,7 @@
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.http.HttpClientInternal;
 import io.vertx.core.internal.net.endpoint.EndpointResolverInternal;
+import io.vertx.core.internal.streams.WriteResult;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.*;
 import io.vertx.core.net.impl.MessageWrite;
@@ -6686,4 +6687,54 @@ public void testServerMessageWrite() throws Exception {
     assertTrue(ref1.get() instanceof MessageWrite);
     assertTrue(ref2.get() instanceof MessageWrite);
   }
+
+  private void stableUnwritable(Buffer chunk, HttpServerResponse response, WriteResult<?> prev, Consumer<WriteResult<?>> done) {
+    WriteResult<?> last = null;
+    while (true) {
+      if (response.writeQueueFull()) {
+        WriteResult<?> w = last;
+        if (w == null) {
+          // Done => use prev
+          done.accept(prev);
+        } else {
+          // Give some time to fill the window
+          vertx.setTimer(10, v -> {
+            stableUnwritable(chunk, response, w, done);
+          });
+        }
+        break;
+      } else {
+        last = (WriteResult<?>)response.write(chunk);
+      }
+    }
+  }
+
+  @Test
+  public void testServerMessageWriteWritability(Checkpoint checkpoint1, Checkpoint checkpoint2, Checkpoint checkpoint3) throws Exception {
+    server.requestHandler(request -> {
+      HttpServerResponse response = request.response();
+      response.setChunked(true);
+      vertx.runOnContext(v1 -> {
+        Buffer chunk = Buffer.buffer(TestUtils.randomAlphaString(1024));
+        stableUnwritable(chunk, response, null, last -> {
+          assertEquals(last.isWritable(), !response.writeQueueFull());
+          last.onComplete(v -> {
+            checkpoint2.succeed();
+          });
+          response.drainHandler(v2 -> {
+            checkpoint3.succeed();
+          });
+          checkpoint1.succeed();
+        });
+      });
+    });
+    startServer();
+    HttpClientResponse response = client.request(new RequestOptions(requestOptions).setPort(server.actualPort()))
+      .compose(req -> req
+        .send()
+        .expecting(HttpResponseExpectation.SC_OK)
+        .andThen(onSuccess(HttpClientResponse::pause))).await();
+    checkpoint1.awaitSuccess();
+    response.resume();
+  }
 }