[Cronet] Fix BidirectionalStream.flush()

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 66 matching lines @@
     private final boolean mDisableAutoFlush;
     private final boolean mDelayRequestHeadersUntilFirstFlush;
 
     /*
      * Synchronizes access to mNativeStream, mReadState and mWriteState.
      */
     private final Object mNativeStreamLock = new Object();
 
     @GuardedBy("mNativeStreamLock")
     // Pending write data.
-    private LinkedList<ByteBuffer> mPendingData;
+    private LinkedList<ByteBuffer> mPendingQueue;

[mef, 2016/06/21 21:12:27]

Any particular reason for XyzData -> XyzQueue rename?

[xunjieli, 2016/06/22 14:26:47]

I felt "Data" is a bit overloaded. I changed it to "Queue" so that it is clearer
that they queues the data/ByteBuffer to be sent.

getFlush[Queue]SizeForTesting() is clear, but getFlush[Data]SizeForTesting()
isn't. The DataSize might be number of bytes, but we really want the size of the
queue.

[mef, 2016/06/24 19:36:30]

Yeah, unfortunately Queue is also very generic, and also not very accurate.
More accurate name would be a 'XyzList', right?
Maybe getFlushDataCountForTesting?

 
     @GuardedBy("mNativeStreamLock")
     // Flush data queue that should be pushed to the native stack when the previous
     // nativeWritevData completes.
-    private LinkedList<ByteBuffer> mFlushData;
+    private LinkedList<ByteBuffer> mFlushQueue;
 
     @GuardedBy("mNativeStreamLock")
     // Whether an end-of-stream flag is passed in through write().
     private boolean mEndOfStreamWritten;
 
     @GuardedBy("mNativeStreamLock")
     // Whether request headers have been sent.
     private boolean mRequestHeadersSent;
 
     /* Native BidirectionalStream object, owned by CronetBidirectionalStream. */
@@ skipping 111 matching lines @@
             boolean disableAutoFlush, boolean delayRequestHeadersUntilNextFlush) {
         mRequestContext = requestContext;
         mInitialUrl = url;
         mInitialPriority = convertStreamPriority(priority);
         mCallback = callback;
         mExecutor = executor;
         mInitialMethod = httpMethod;
         mRequestHeaders = stringsFromHeaderList(requestHeaders);
         mDisableAutoFlush = disableAutoFlush;
         mDelayRequestHeadersUntilFirstFlush = delayRequestHeadersUntilNextFlush;
-        mPendingData = new LinkedList<>();
-        mFlushData = new LinkedList<>();
+        mPendingQueue = new LinkedList<>();
+        mFlushQueue = new LinkedList<>();
     }
 
     @Override
     public void start() {
         synchronized (mNativeStreamLock) {
             if (mReadState != State.NOT_STARTED) {
                 throw new IllegalStateException("Stream is already started.");
             }
             try {
                 mNativeStream = nativeCreateBidirectionalStream(
@@ skipping 51 matching lines @@
             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 (isDoneLocked()) {
                 return;
             }
-            mPendingData.add(buffer);
+            mPendingQueue.add(buffer);
             if (endOfStream) {
                 mEndOfStreamWritten = true;
             }
             if (!mDisableAutoFlush) {
                 flushLocked();
             }
         }
     }
 
     @Override
     public void flush() {
         synchronized (mNativeStreamLock) {
             flushLocked();
         }
     }
 
     @SuppressWarnings("GuardedByChecker")
     private void flushLocked() {
         if (isDoneLocked()
                 || (mWriteState != State.WAITING_FOR_FLUSH && mWriteState != State.WRITING)) {
             return;
         }
-        if (mPendingData.isEmpty() && mFlushData.isEmpty()) {
+        if (mPendingQueue.isEmpty() && mFlushQueue.isEmpty()) {
             // If there is no pending write when flush() is called, see if
             // request headers need to be flushed.
             if (!mRequestHeadersSent) {
                 mRequestHeadersSent = true;
                 nativeSendRequestHeaders(mNativeStream);
                 if (!doesMethodAllowWriteData(mInitialMethod)) {
                     mWriteState = State.WRITING_DONE;
                 }
             }
             return;
         }
 
-        assert !mPendingData.isEmpty() || !mFlushData.isEmpty();
+        assert !mPendingQueue.isEmpty() || !mFlushQueue.isEmpty();
 
-        // Move buffers from mPendingData to the flushing queue.
-        if (!mPendingData.isEmpty()) {
-            mFlushData.addAll(mPendingData);
-            mPendingData.clear();
+        // Move buffers from mPendingQueue to the flushing queue.
+        if (!mPendingQueue.isEmpty()) {
+            mFlushQueue.addAll(mPendingQueue);
+            mPendingQueue.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;
         }
+        sendFlushDataLocked();
+    }
+
+    // Helper method to send buffers in mFlushQueue. Caller needs to acquire
+    // mNativeStreamLock and make sure mWriteState is WAITING_FOR_FLUSH and
+    // mFlushQueue queue isn't empty.
+    @SuppressWarnings("GuardedByChecker")
+    private void sendFlushDataLocked() {
         assert mWriteState == State.WAITING_FOR_FLUSH;
-        int size = mFlushData.size();
+        int size = mFlushQueue.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();
+            ByteBuffer buffer = mFlushQueue.poll();
             buffers[i] = buffer;
             positions[i] = buffer.position();
             limits[i] = buffer.limit();
         }
-        assert mFlushData.isEmpty();
+        assert mFlushQueue.isEmpty();
         assert buffers.length >= 1;
         mWriteState = State.WRITING;
-        if (!nativeWritevData(mNativeStream, buffers, positions, limits, mEndOfStreamWritten)) {
+        if (!nativeWritevData(mNativeStream, buffers, positions, limits,
+                    mEndOfStreamWritten && mPendingQueue.isEmpty())) {
             // Still waiting on flush. This is just to have consistent
             // behavior with the other error cases.
             mWriteState = State.WAITING_FOR_FLUSH;
             throw new IllegalArgumentException("Unable to call native writev.");
         }
     }
 
+    @VisibleForTesting
+    public int getPendingQueueSizeForTesting() {
+        synchronized (mNativeStreamLock) {
+            return mPendingQueue.size();
+        }
+    }
+
+    @VisibleForTesting
+    public int getFlushQueueSizeForTesting() {
+        synchronized (mNativeStreamLock) {
+            return mFlushQueue.size();
+        }
+    }
+
     @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 128 matching lines @@
     }
 
     @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_FLUSH;
-            // Flush if there is anything in the flush queue mFlushData.
-            if (!mFlushData.isEmpty()) {
-                flushLocked();
+            // Flush if there is anything in the flush queue mFlushQueue.
+            if (!mFlushQueue.isEmpty()) {
+                sendFlushDataLocked();
             }
         }
         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());
-            postTaskToExecutor(new OnWriteCompletedRunnable(buffer, endOfStream));
+            postTaskToExecutor(new OnWriteCompletedRunnable(
+                    buffer, endOfStream && i == byteBuffers.length - 1));

[mef, 2016/06/21 21:12:27]

this is a bit hard to understand, maybe it deserves a comment?

[xunjieli, 2016/06/22 14:26:47]

Done.

         }
     }
 
     @SuppressWarnings("unused")
     @CalledByNative
     private void onResponseTrailersReceived(String[] trailers) {
         final UrlResponseInfo.HeaderBlock trailersBlock =
                 new UrlResponseInfo.HeaderBlock(headersListFromStrings(trailers));
         postTaskToExecutor(new Runnable() {
             public void run() {
@@ skipping 185 matching lines @@
     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);
 }