Coalesce small buffers in net::BidirectionalStream

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

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package org.chromium.net;
 
 import org.chromium.base.Log;
 import org.chromium.base.VisibleForTesting;
 import org.chromium.base.annotations.CalledByNative;
 import org.chromium.base.annotations.JNINamespace;
 import org.chromium.base.annotations.NativeClassQualifiedName;
 
 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;
 import java.util.concurrent.RejectedExecutionException;
 
 import javax.annotation.concurrent.GuardedBy;
 
 /**
  * {@link BidirectionalStream} implementation using Chromium network stack.
  * All @CalledByNative methods are called on the native network thread
@@ skipping 22 matching lines @@
         /* There is no more data to read and stream is half-closed by the remote side. */
         READING_DONE,
         /* Stream is canceled. */
         CANCELED,
         /* Error has occured, stream is closed. */
         ERROR,
         /* Reading and writing are done, and the stream is closed successfully. */
         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,
     }
 
     private final CronetUrlRequestContext mRequestContext;
     private final Executor mExecutor;
     private final Callback mCallback;
     private final String mInitialUrl;
     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;
 
     /**
      * Read state is tracking reading flow.
      *                         / <--- READING <--- \
      *                         |                   |
      *                         \                   /
      * NOT_STARTED -> STARTED --> WAITING_FOR_READ -> READING_DONE -> SUCCESS
      */
     @GuardedBy("mNativeStreamLock")
     private State mReadState = State.NOT_STARTED;
 
     /**
      * 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;
 
     private UrlResponseInfo mResponseInfo;
 
     /*
      * OnReadCompleted callback is repeatedly invoked when each read is completed, so it
      * is cached as a member variable.
      */
     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 {
         // Buffer passed back from current invocation of onReadCompleted.
         ByteBuffer mByteBuffer;
         // End of stream flag from current invocation of onReadCompleted.
         boolean mEndOfStream;
 
         @Override
         public void run() {
             try {
                 // Null out mByteBuffer, to pass buffer ownership to callback or release if done.
                 ByteBuffer buffer = mByteBuffer;
                 mByteBuffer = null;
                 synchronized (mNativeStreamLock) {
                     if (isDoneLocked()) {
                         return;
                     }
                     if (mEndOfStream) {
                         mReadState = State.READING_DONE;
                         if (maybeSucceedLocked()) {
                             return;

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

I think we should always call mCallback.onReadCompleted() regardless whether it
was the last read or not. Otherwise the behavior is inconsistent, i.e. sometimes
the last read will trigger onReadCompleted() and sometimes not. Moreover, it is
now inconsistent with OnWriteCompletedRunnable, which always calls
mCallback.onWriteCompleted()

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

Done. I agree. Although this might break gRPC, depending on how they are calling
read(). But I think it should be fine, since this CL requires API change anyway.

                         }
                     } else {
                         mReadState = State.WAITING_FOR_READ;
                     }
                 }
                 mCallback.onReadCompleted(CronetBidirectionalStream.this, mResponseInfo, buffer);
             } catch (Exception e) {
                 onCallbackException(e);
             }
         }
@@ skipping 10 matching lines @@
             try {
                 // Null out mByteBuffer, to pass buffer ownership to callback or release if done.
                 ByteBuffer buffer = mByteBuffer;
                 mByteBuffer = null;
                 synchronized (mNativeStreamLock) {
                     if (isDoneLocked()) {
                         return;
                     }
                     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;
                     }
                 }
                 mCallback.onWriteCompleted(CronetBidirectionalStream.this, mResponseInfo, buffer);
             } catch (Exception e) {
                 onCallbackException(e);
             }
         }
     }
 
     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);
         mCallback = callback;
         mExecutor = executor;
         mInitialMethod = httpMethod;
         mRequestHeaders = stringsFromHeaderList(requestHeaders);
+        mDisableAutoFlush = disableAutoFlush;
+        mPendingData = new LinkedList<ByteBuffer>();
     }
 
     @Override
     public void start() {
         synchronized (mNativeStreamLock) {
             if (mReadState != State.NOT_STARTED) {
                 throw new IllegalStateException("Stream is already started.");
             }
             try {
                 mNativeStream = nativeCreateBidirectionalStream(
-                        mRequestContext.getUrlRequestContextAdapter());
+                        mRequestContext.getUrlRequestContextAdapter(), mDisableAutoFlush);
                 mRequestContext.onRequestStarted();
                 // Non-zero startResult means an argument error.
                 int startResult = nativeStart(mNativeStream, mInitialUrl, mInitialPriority,
                         mInitialMethod, mRequestHeaders, !doesMethodAllowWriteData(mInitialMethod));
                 if (startResult == -1) {
                     throw new IllegalArgumentException("Invalid http method " + mInitialMethod);
                 }
                 if (startResult > 0) {
                     int headerPos = startResult - 1;
                     throw new IllegalArgumentException("Invalid header "
@@ skipping 33 matching lines @@
         }
     }
 
     @Override
     public void write(ByteBuffer buffer, boolean endOfStream) {
         synchronized (mNativeStreamLock) {
             Preconditions.checkDirect(buffer);
             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;
+            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.

-            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");
-            }
+            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.

+        }
+    }
+
     @Override
     public void ping(PingCallback callback, Executor executor) {
         // TODO(mef): May be last thing to be implemented on Android.
         throw new UnsupportedOperationException("ping is not supported yet.");
     }
 
     @Override
     public void windowUpdate(int windowSizeIncrement) {
         // TODO(mef): Understand the needs and semantics of this method.
         throw new UnsupportedOperationException("windowUpdate is not supported yet.");
@@ skipping 17 matching lines @@
         }
     }
 
     @GuardedBy("mNativeStreamLock")
     private boolean isDoneLocked() {
         return mReadState != State.NOT_STARTED && mNativeStream == 0;
     }
 
     @SuppressWarnings("unused")
     @CalledByNative
-    private void onRequestHeadersSent() {
+    private void onStreamReady() {
         postTaskToExecutor(new Runnable() {
             public void run() {
                 synchronized (mNativeStreamLock) {
                     if (isDoneLocked()) {
                         return;
                     }
                     if (doesMethodAllowWriteData(mInitialMethod)) {
                         mWriteState = State.WAITING_FOR_WRITE;
                     } else {
                         mWriteState = State.WRITING_DONE;
                     }
                 }
 
                 try {
-                    mCallback.onRequestHeadersSent(CronetBidirectionalStream.this);
+                    mCallback.onStreamReady(CronetBidirectionalStream.this);
                 } catch (Exception e) {
                     onCallbackException(e);
                 }
             }
         });
     }
 
     /**
      * Called when the final set of headers, after all redirects,
      * is received. Can only be called once for each stream.
@@ skipping 44 matching lines @@
         }
         byteBuffer.position(initialPosition + bytesRead);
         assert mOnReadCompletedTask.mByteBuffer == null;
         mOnReadCompletedTask.mByteBuffer = byteBuffer;
         mOnReadCompletedTask.mEndOfStream = (bytesRead == 0);
         postTaskToExecutor(mOnReadCompletedTask);
     }
 
     @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")
     @CalledByNative
     private void onResponseTrailersReceived(String[] trailers) {
         final UrlResponseInfo.HeaderBlock trailersBlock =
                 new UrlResponseInfo.HeaderBlock(headersListFromStrings(trailers));
         postTaskToExecutor(new Runnable() {
             public void run() {
                 synchronized (mNativeStreamLock) {
@@ skipping 195 matching lines @@
      */
     private void failWithException(final CronetException exception) {
         postTaskToExecutor(new Runnable() {
             public void run() {
                 failWithExceptionOnExecutor(exception);
             }
         });
     }
 
     // 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,
             String[] headers, boolean endOfStream);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native boolean nativeReadData(
             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);
 }