Coalesce small buffers in net::BidirectionalStream

components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java

Index: components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
diff --git a/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java b/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
index 4c0fae741dbf49d5534a516ada1703a9e03fb8b0..60fdf6ad3ea95426308e3254150d776b34260bd1 100644
--- a/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
+++ b/components/cronet/android/java/src/org/chromium/net/CronetBidirectionalStream.java
@@ -14,6 +14,7 @@ import java.nio.ByteBuffer;
 import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Executor;
@@ -56,7 +57,10 @@ class CronetBidirectionalStream extends BidirectionalStream {
         SUCCESS,
         /* Waiting for {@code write()} to be called. */
         WAITING_FOR_WRITE,
-        /* Writing to the remote, {@code onWriteCompleted()} callback will be called when done. */
+        /*
+         * Writing to the remote, {@code onWriteCompleted()} callback will be called when done.
+         * This state is only applicable when {@code mDisableAutoFlush} is false.
+         */
         WRITING,

[kapishnikov, 2016/04/19 03:22:22]

Do we need this state? If it is not really used, I think removing it will make
enable and disable auto flush cases more consistent.

[xunjieli, 2016/04/19 19:30:03]

Yes, we need this state to detect that there is only one Write() in fight when
mDisableAutoFlush is false, because the native code can only handle one
in-flight Write() or Writev().

         /* There is no more data to write and stream is half-closed by the local side. */
         WRITING_DONE,
@@ -69,12 +73,20 @@ class CronetBidirectionalStream extends BidirectionalStream {
     private final int mInitialPriority;
     private final String mInitialMethod;
     private final String mRequestHeaders[];
+    private final boolean mDisableAutoFlush;
 
     /*
      * Synchronizes access to mNativeStream, mReadState and mWriteState.
      */
     private final Object mNativeStreamLock = new Object();
 
+    @GuardedBy("mNativeStreamLock")
+    private final LinkedList<ByteBuffer> mPendingData;
+
+    @GuardedBy("mNativeStreamLock")
+    // Whether an end-of-stream flag is passed in through write().
+    private boolean mEndOfStreamWritten;
+
     /* Native BidirectionalStream object, owned by CronetBidirectionalStream. */
     @GuardedBy("mNativeStreamLock")
     private long mNativeStream;
@@ -91,10 +103,10 @@ class CronetBidirectionalStream extends BidirectionalStream {
 
     /**
      * Write state is tracking writing flow.
-     *                         / <--- WRITING <---  \
-     *                         |                    |
-     *                         \                    /
-     * NOT_STARTED -> STARTED --> WAITING_FOR_WRITE -> WRITING_DONE -> SUCCESS
+     *                         / <--- WRITING (if !mDisableAutoFlush)<--- \
+     *                         |                                          |
+     *                         \                                          /
+     * NOT_STARTED -> STARTED --> -------------WAITING_FOR_WRITE --------> WRITING_DONE -> SUCCESS
      */
     @GuardedBy("mNativeStreamLock")
     private State mWriteState = State.NOT_STARTED;
@@ -107,12 +119,6 @@ class CronetBidirectionalStream extends BidirectionalStream {
      */
     private OnReadCompletedRunnable mOnReadCompletedTask;
 
-    /*
-     * OnWriteCompleted callback is repeatedly invoked when each write is completed, so it
-     * is cached as a member variable.
-     */
-    private OnWriteCompletedRunnable mOnWriteCompletedTask;
-
     private Runnable mOnDestroyedCallbackForTesting;
 
     private final class OnReadCompletedRunnable implements Runnable {
@@ -165,9 +171,8 @@ class CronetBidirectionalStream extends BidirectionalStream {
                     }
                     if (mEndOfStream) {
                         mWriteState = State.WRITING_DONE;
-                        if (maybeSucceedLocked()) {
-                            return;
-                        }
+                        // Maybe post an onSucceeded callback to be executed after run() completes.
+                        maybeSucceedLocked();

[kapishnikov, 2016/04/19 03:22:22]

The sequence of calls should be mCallback.onWriteCompleted(), then
mCallback.onSucceeded(). Currently, it is possible that mCallback.onSucceeded()
will be called first and then followed by mCallback.onWriteCompleted().

[xunjieli, 2016/04/19 19:30:03]

No, it isn't possible. onSucceeded is posted asynchronously, so it is run after
onWriteCompleted.

[kapishnikov, 2016/04/19 19:56:39]

Do we specify anywhere that the executor should be single-threaded? If it can be
multithreaded then onSucceeded() can be posted on a different thread. The lock
is released before mCallback.onWriteCompleted(...) is called. So, is it possible
that the second thread will execute onSucceeded() faster than the first thread
executes onWriteCompleted().

                     } else {
                         mWriteState = State.WAITING_FOR_WRITE;
                     }
@@ -181,7 +186,8 @@ class CronetBidirectionalStream extends BidirectionalStream {
 
     CronetBidirectionalStream(CronetUrlRequestContext requestContext, String url,
             @BidirectionalStream.Builder.StreamPriority int priority, Callback callback,
-            Executor executor, String httpMethod, List<Map.Entry<String, String>> requestHeaders) {
+            Executor executor, String httpMethod, List<Map.Entry<String, String>> requestHeaders,
+            boolean disableAutoFlush) {
         mRequestContext = requestContext;
         mInitialUrl = url;
         mInitialPriority = convertStreamPriority(priority);
@@ -189,6 +195,8 @@ class CronetBidirectionalStream extends BidirectionalStream {
         mExecutor = executor;
         mInitialMethod = httpMethod;
         mRequestHeaders = stringsFromHeaderList(requestHeaders);
+        mDisableAutoFlush = disableAutoFlush;
+        mPendingData = new LinkedList<ByteBuffer>();
     }
 
     @Override
@@ -199,7 +207,7 @@ class CronetBidirectionalStream extends BidirectionalStream {
             }
             try {
                 mNativeStream = nativeCreateBidirectionalStream(
-                        mRequestContext.getUrlRequestContextAdapter());
+                        mRequestContext.getUrlRequestContextAdapter(), mDisableAutoFlush);
                 mRequestContext.onRequestStarted();
                 // Non-zero startResult means an argument error.
                 int startResult = nativeStart(mNativeStream, mInitialUrl, mInitialPriority,
@@ -253,24 +261,53 @@ class CronetBidirectionalStream extends BidirectionalStream {
             if (!buffer.hasRemaining() && !endOfStream) {
                 throw new IllegalArgumentException("Empty buffer before end of stream.");
             }
+            if (mEndOfStreamWritten) {
+                throw new IllegalArgumentException("Write after writing end of stream.");
+            }
             if (mWriteState != State.WAITING_FOR_WRITE) {
                 throw new IllegalStateException("Unexpected write attempt.");
             }
             if (isDoneLocked()) {
                 return;
             }
-            if (mOnWriteCompletedTask == null) {
-                mOnWriteCompletedTask = new OnWriteCompletedRunnable();
+            mPendingData.add(buffer);
+            if (endOfStream) {
+                mEndOfStreamWritten = true;
             }
-            mOnWriteCompletedTask.mEndOfStream = endOfStream;
-            mWriteState = State.WRITING;
-            if (!nativeWriteData(
-                        mNativeStream, buffer, buffer.position(), buffer.limit(), endOfStream)) {
-                // Still waiting on write. This is just to have consistent
-                // behavior with the other error cases.
-                mWriteState = State.WAITING_FOR_WRITE;
-                throw new IllegalArgumentException("Unable to call native write");
+            if (mDisableAutoFlush) {
+                return;
             }
+            mWriteState = State.WRITING;

[kapishnikov, 2016/04/19 03:22:22]

If we remove WRITING state, we can write:
if (!mDisableAutoFlush) {
    flushLocked();
}
That should be the difference between auto flush enabled/disabled cases.

[xunjieli, 2016/04/19 19:30:03]

State.WRITING is needed to detect that there isn't more than one Write() in
flight when mDisableAutoFlush is false. When mDisableAutoFlush() is false, we
flush() automatically after every write(). However, the native code can only
handle one write() or writev() in flight.

+            flushLocked();
+        }
+    }
+
+    @Override
+    public void flush() {
+        synchronized (mNativeStreamLock) {
+            flushLocked();
+        }
+    }
+
+    @SuppressWarnings("GuardedByChecker")
+    private void flushLocked() {
+        if (mPendingData.isEmpty()) {
+            // No-op if there is nothing to write.
+            return;
+        }
+        int size = mPendingData.size();
+        ByteBuffer[] buffers = new ByteBuffer[size];
+        int[] positions = new int[size];
+        int[] limits = new int[size];
+        for (int i = 0; i < size; i++) {
+            ByteBuffer buffer = mPendingData.poll();
+            buffers[i] = buffer;
+            positions[i] = buffer.position();
+            limits[i] = buffer.limit();
+        }
+        assert mPendingData.isEmpty();
+        if (!nativeWritevData(mNativeStream, buffers, positions, limits, mEndOfStreamWritten)) {
+            throw new IllegalArgumentException("Unable to call native write");

[kapishnikov, 2016/04/19 03:22:22]

Should we change the state to ERROR here? 

[xunjieli, 2016/04/19 19:30:03]

Done.

         }
     }
 
@@ -311,7 +348,7 @@ class CronetBidirectionalStream extends BidirectionalStream {
 
     @SuppressWarnings("unused")
     @CalledByNative
-    private void onRequestHeadersSent() {
+    private void onStreamReady() {
         postTaskToExecutor(new Runnable() {
             public void run() {
                 synchronized (mNativeStreamLock) {
@@ -326,7 +363,7 @@ class CronetBidirectionalStream extends BidirectionalStream {
                 }
 
                 try {
-                    mCallback.onRequestHeadersSent(CronetBidirectionalStream.this);
+                    mCallback.onStreamReady(CronetBidirectionalStream.this);
                 } catch (Exception e) {
                     onCallbackException(e);
                 }
@@ -391,18 +428,25 @@ class CronetBidirectionalStream extends BidirectionalStream {
 
     @SuppressWarnings("unused")
     @CalledByNative
-    private void onWriteCompleted(
-            final ByteBuffer byteBuffer, int initialPosition, int initialLimit) {
-        if (byteBuffer.position() != initialPosition || byteBuffer.limit() != initialLimit) {
-            failWithException(
-                    new CronetException("ByteBuffer modified externally during write", null));
-            return;
+    private void onWritevCompleted(final ByteBuffer[] byteBuffers, int[] initialPositions,
+            int[] initialLimits, boolean endOfStream) {
+        assert byteBuffers.length == initialPositions.length;
+        assert byteBuffers.length == initialLimits.length;
+        assert initialPositions.length == initialLimits.length;
+        for (int i = 0; i < byteBuffers.length; i++) {
+            ByteBuffer buffer = byteBuffers[i];
+            if (buffer.position() != initialPositions[i] || buffer.limit() != initialLimits[i]) {
+                failWithException(
+                        new CronetException("ByteBuffer modified externally during write", null));
+                return;
+            }
+            // Current implementation always writes the complete buffer.
+            buffer.position(buffer.limit());
+            OnWriteCompletedRunnable runnable = new OnWriteCompletedRunnable();
+            runnable.mByteBuffer = buffer;
+            runnable.mEndOfStream = endOfStream;
+            postTaskToExecutor(runnable);
         }
-        // Current implementation always writes the complete buffer.
-        byteBuffer.position(byteBuffer.limit());
-        assert mOnWriteCompletedTask.mByteBuffer == null;
-        mOnWriteCompletedTask.mByteBuffer = byteBuffer;
-        postTaskToExecutor(mOnWriteCompletedTask);
     }
 
     @SuppressWarnings("unused")
@@ -618,7 +662,8 @@ class CronetBidirectionalStream extends BidirectionalStream {
     }
 
     // Native methods are implemented in cronet_bidirectional_stream_adapter.cc.
-    private native long nativeCreateBidirectionalStream(long urlRequestContextAdapter);
+    private native long nativeCreateBidirectionalStream(
+            long urlRequestContextAdapter, boolean disableAutoFlush);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native int nativeStart(long nativePtr, String url, int priority, String method,
@@ -629,8 +674,8 @@ class CronetBidirectionalStream extends BidirectionalStream {
             long nativePtr, ByteBuffer byteBuffer, int position, int limit);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
-    private native boolean nativeWriteData(
-            long nativePtr, ByteBuffer byteBuffer, int position, int limit, boolean endOfStream);
+    private native boolean nativeWritevData(long nativePtr, ByteBuffer[] buffers, int[] positions,
+            int[] limits, boolean endOfStream);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native void nativeDestroy(long nativePtr, boolean sendOnCanceled);