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 35 matching lines @@
         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")
+    // Only relevant if mDisableAutoFlush is true.
+    private final LinkedList<ByteBuffer> mPendingData;
+
+    @GuardedBy("mNativeStreamLock")
+    // Only relevant if mDisableAutoFlush is true.

[kapishnikov, 2016/04/11 23:09:41]

Would it be worth to unify the implementation of autoflush and non-autoflush
cases? The only difference I see is that flush() should automatically be called
by write() method when autoflush is enabled and explicitly called by the client
when it is disabled.

The is slight performance impact but it looks negligible.

[xunjieli, 2016/04/12 18:01:48]

Done. Although partially. I think we probably should leave the underlying net
implementation having two write versions. I tried to consolidate them, but it
becomes ugly too soon. Let's handle the remaining in a follow-up.

+    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
@@ skipping 86 matching lines @@
                 }
                 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;
+        if (mDisableAutoFlush) {
+            mPendingData = new LinkedList<ByteBuffer>();
+        } else {
+            mPendingData = null;
+        }
     }
 
     @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 (isDoneLocked()) {

[kapishnikov, 2016/04/11 23:09:42]

Should we throw an IllegalStateException if there is an attempt to write data
when mEndOfStreamWritten is true?

[xunjieli, 2016/04/12 18:01:48]

Done.

+                return;
+            }
+
+            if (mDisableAutoFlush) {
+                mPendingData.add(buffer);
+                if (endOfStream) {
+                    mEndOfStreamWritten = true;
+                }
+                return;
+            }
             if (mWriteState != State.WAITING_FOR_WRITE) {
                 throw new IllegalStateException("Unexpected write attempt.");
             }
-            if (isDoneLocked()) {
-                return;
-            }
             if (mOnWriteCompletedTask == null) {
                 mOnWriteCompletedTask = new OnWriteCompletedRunnable();
             }
-            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");
             }
         }
     }
 
     @Override
+    public void flush() {
+        synchronized (mNativeStreamLock) {
+            if (!mDisableAutoFlush) {
+                throw new IllegalStateException("flush is called when auto flush is not disabled.");
+            }
+
+            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");
+            }
+        }
+    }
+
+    @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 16 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) {
+    private void onWriteCompleted(final ByteBuffer byteBuffer, int initialPosition,
+            int initialLimit, boolean endOfStream) {
         if (byteBuffer.position() != initialPosition || byteBuffer.limit() != initialLimit) {
             failWithException(
                     new CronetException("ByteBuffer modified externally during write", null));
             return;
         }
         // Current implementation always writes the complete buffer.
         byteBuffer.position(byteBuffer.limit());
         assert mOnWriteCompletedTask.mByteBuffer == null;
         mOnWriteCompletedTask.mByteBuffer = byteBuffer;
+        mOnWriteCompletedTask.mEndOfStream = endOfStream;
         postTaskToExecutor(mOnWriteCompletedTask);
     }
 
     @SuppressWarnings("unused")
     @CalledByNative
+    private void onWritevCompleted(final ByteBuffer[] byteBuffers, int[] initialPositions,

[kapishnikov, 2016/04/11 23:09:42]

It looks that onWritevCompleted() is called after all byteBuffers have been
written. Would it be better to call onWriteCompleted() for every written buffer;
thus, the client can start filling individual buffers with new data without
waiting for all writes to complete.

[xunjieli, 2016/04/12 18:01:48]

As discussed, we don't have the signal about each buffer being passed through
the wire. 

+            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);
+        }
+    }
+
+    @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) {
                     if (isDoneLocked()) {
                         return;
                     }
                 }
@@ skipping 191 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);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native boolean nativeWritevData(long nativePtr, ByteBuffer[] buffers, int[] positions,
+            int[] limits, boolean endOfStream);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native void nativeDestroy(long nativePtr, boolean sendOnCanceled);
 }