[Cronet] Make delaying sending request headers explicit in bidirectional stream

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;
@@ skipping 23 matching lines @@
      * States of BidirectionalStream are tracked in mReadState and mWriteState.
      * The write state is separated out as it changes independently of the read state.
      * There is one initial state: State.NOT_STARTED. There is one normal final state:
      * State.SUCCESS, reached after State.READING_DONE and State.WRITING_DONE. There are two
      * exceptional final states: State.CANCELED and State.ERROR, which can be reached from
      * any other non-final state.
      */
     private enum State {
         /* Initial state, stream not started. */
         NOT_STARTED,
-        /* Stream started, request headers are being sent. */
+        /*
+         * Stream started, request headers are being sent if mDelayRequestHeadersUntilNextFlush
+         * is not set to true.
+         */
         STARTED,
         /* Waiting for {@code read()} to be called. */
         WAITING_FOR_READ,
         /* Reading from the remote, {@code onReadCompleted()} callback will be called when done. */
         READING,
         /* 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 nativeWritevData()} to be called. */
-        WAITING_FOR_WRITE,
+        /* Waiting for {@code nativeSendRequestHeaders()} or {@code nativeWritevData()} to be
+           called. */
+        WAITING_FOR_FLUSH,
         /* Writing to the remote, {@code onWritevCompleted()} callback will be called when done. */
         WRITING,
         /* 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;
+    private final boolean mDelayHeadersUntilNextFlush;
 
     /*
      * Synchronizes access to mNativeStream, mReadState and mWriteState.
      */
     private final Object mNativeStreamLock = new Object();
 
     @GuardedBy("mNativeStreamLock")
     // Pending write data.
     private LinkedList<ByteBuffer> mPendingData;
 
     @GuardedBy("mNativeStreamLock")
     // Flush data queue that should be pushed to the native stack when the previous
     // nativeWritevData completes.
     private LinkedList<ByteBuffer> mFlushData;
 
     @GuardedBy("mNativeStreamLock")
     // Whether an end-of-stream flag is passed in through write().
     private boolean mEndOfStreamWritten;
 
+    @GuardedBy("mNativeStreamLock")
+    // Whether request headers have been flushed.
+    private boolean mRequestHeadersFlushed;
+
     /* 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
+     * NOT_STARTED -> STARTED --> WAITING_FOR_FLUSH -> WRITING_DONE -> SUCCESS
      */
     @GuardedBy("mNativeStreamLock")
     private State mWriteState = State.NOT_STARTED;
 
     // Only modified on the network thread.
     private UrlResponseInfo mResponseInfo;
 
     /*
      * OnReadCompleted callback is repeatedly invoked when each read is completed, so it
      * is cached as a member variable.
@@ skipping 72 matching lines @@
                 }
             } 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,
-            boolean disableAutoFlush) {
+            boolean disableAutoFlush, boolean delayRequestHeadersUntilNextFlush) {
         mRequestContext = requestContext;
         mInitialUrl = url;
         mInitialPriority = convertStreamPriority(priority);
         mCallback = callback;
         mExecutor = executor;
         mInitialMethod = httpMethod;
         mRequestHeaders = stringsFromHeaderList(requestHeaders);
         mDisableAutoFlush = disableAutoFlush;
+        mDelayHeadersUntilNextFlush = delayRequestHeadersUntilNextFlush;
         mPendingData = new LinkedList<>();
         mFlushData = new LinkedList<>();
     }
 
     @Override
     public void start() {
         synchronized (mNativeStreamLock) {
             if (mReadState != State.NOT_STARTED) {
                 throw new IllegalStateException("Stream is already started.");
             }
             try {
                 mNativeStream = nativeCreateBidirectionalStream(
-                        mRequestContext.getUrlRequestContextAdapter(), mDisableAutoFlush);
+                        mRequestContext.getUrlRequestContextAdapter(),
+                        !mDelayHeadersUntilNextFlush);
                 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 58 matching lines @@
 
     @Override
     public void flush() {
         synchronized (mNativeStreamLock) {
             flushLocked();
         }
     }
 
     @SuppressWarnings("GuardedByChecker")
     private void flushLocked() {
-        if (isDoneLocked()) {
+        if (isDoneLocked()
+                || (mWriteState != State.WAITING_FOR_FLUSH && mWriteState != State.WRITING)) {
             return;
         }
         if (mPendingData.isEmpty() && mFlushData.isEmpty()) {
-            // No-op if there is nothing to write.
+            // If there is no pending write when flush() is called, see if
+            // request headers need to be flushed.
+            if (!mRequestHeadersFlushed) {
+                mRequestHeadersFlushed = true;
+                nativeSendRequestHeaders(mNativeStream);
+            }

[kapishnikov, 2016/05/31 18:42:12]

Should we change mWriteState to State.WRITING_DONE if the http method is GET or
HEAD?

[xunjieli, 2016/05/31 19:31:53]

Done. Good catch!

             return;
         }
 
+        assert !(mPendingData.isEmpty() && mFlushData.isEmpty());

[kapishnikov, 2016/05/31 18:42:12]

Will it prevent calling flush() two times in a roll without calling write() in
between? Should we 'return' instead?

[xunjieli, 2016/05/31 19:31:53]

We have an early return on line 327 to prevent calling flush() two times in a
roll.

[kapishnikov, 2016/05/31 19:47:24]

Acknowledged.

+
         // Move buffers from mPendingData to the flushing queue.
         if (!mPendingData.isEmpty()) {
             mFlushData.addAll(mPendingData);
             mPendingData.clear();
         }
 
         if (mWriteState == State.WRITING) {
             // If there is a write already pending, wait until onWritevCompleted is
             // called before pushing data to the native stack.
             return;
         }
+        assert mWriteState == State.WAITING_FOR_FLUSH;
         int size = mFlushData.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 = mFlushData.poll();
             buffers[i] = buffer;
             positions[i] = buffer.position();
             limits[i] = buffer.limit();
         }
         assert mFlushData.isEmpty();
+        assert buffers.length >= 1;
         mWriteState = State.WRITING;
         if (!nativeWritevData(mNativeStream, buffers, positions, limits, mEndOfStreamWritten)) {
             // Still waiting on write. This is just to have consistent
             // behavior with the other error cases.
-            mWriteState = State.WAITING_FOR_WRITE;
+            mWriteState = State.WAITING_FOR_FLUSH;
             throw new IllegalArgumentException("Unable to call native writev.");
         }
     }
 
     @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.");
     }
 
@@ skipping 44 matching lines @@
         }
         try {
             mCallback.onSucceeded(CronetBidirectionalStream.this, mResponseInfo);
         } catch (Exception e) {
             Log.e(CronetUrlRequestContext.LOG_TAG, "Exception in onSucceeded method", e);
         }
     }
 
     @SuppressWarnings("unused")
     @CalledByNative
-    private void onStreamReady() {
+    private void onStreamReady(final boolean requestHeadersFlushed) {
         postTaskToExecutor(new Runnable() {
             public void run() {
                 synchronized (mNativeStreamLock) {
                     if (isDoneLocked()) {
                         return;
                     }
-                    if (doesMethodAllowWriteData(mInitialMethod)) {
-                        mWriteState = State.WAITING_FOR_WRITE;
-                        mReadState = State.WAITING_FOR_READ;
+                    mRequestHeadersFlushed = requestHeadersFlushed;
+                    mReadState = State.WAITING_FOR_READ;
+                    if (!doesMethodAllowWriteData(mInitialMethod) && mRequestHeadersFlushed) {
+                        mWriteState = State.WRITING_DONE;
                     } else {
-                        mWriteState = State.WRITING_DONE;
+                        mWriteState = State.WAITING_FOR_FLUSH;
                     }
                 }
 
                 try {
                     mCallback.onStreamReady(CronetBidirectionalStream.this);
                 } catch (Exception e) {
                     onCallbackException(e);
                 }
             }
         });
@@ skipping 54 matching lines @@
         postTaskToExecutor(mOnReadCompletedTask);
     }
 
     @SuppressWarnings("unused")
     @CalledByNative
     private void onWritevCompleted(final ByteBuffer[] byteBuffers, int[] initialPositions,
             int[] initialLimits, boolean endOfStream) {
         assert byteBuffers.length == initialPositions.length;
         assert byteBuffers.length == initialLimits.length;
         synchronized (mNativeStreamLock) {
-            mWriteState = State.WAITING_FOR_WRITE;
+            mWriteState = State.WAITING_FOR_FLUSH;
             // Flush if there is anything in the flush queue mFlushData.
             if (!mFlushData.isEmpty()) {
                 flushLocked();
             }
         }
         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));
@@ skipping 180 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, boolean disableAutoFlush);
+            long urlRequestContextAdapter, boolean sendRequestHeadersAutomatically);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native int nativeStart(long nativePtr, String url, int priority, String method,
             String[] headers, boolean endOfStream);
 
     @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
+    private native void nativeSendRequestHeaders(long nativePtr);
+
+    @NativeClassQualifiedName("CronetBidirectionalStreamAdapter")
     private native boolean nativeReadData(
             long nativePtr, ByteBuffer byteBuffer, int position, int limit);
 
     @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);
 }