[Cronet] In JavaUrlRequest avoid thread-unsafe access to HttpURLConnection

components/cronet/android/java/src/org/chromium/net/impl/JavaUrlRequest.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.impl;
 
 import android.annotation.SuppressLint;
 import android.annotation.TargetApi;
 import android.net.TrafficStats;
 import android.os.Build;
@@ skipping 10 matching lines @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.HttpURLConnection;
 import java.net.URI;
 import java.net.URL;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
 import java.util.AbstractMap.SimpleEntry;
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
+import javax.annotation.concurrent.GuardedBy;
+
 /**
  * Pure java UrlRequest, backed by {@link HttpURLConnection}.
  */
 @TargetApi(Build.VERSION_CODES.ICE_CREAM_SANDWICH) // TrafficStats only available on ICS
 final class JavaUrlRequest extends UrlRequestBase {
     private static final String X_ANDROID = "X-Android";
     private static final String X_ANDROID_SELECTED_TRANSPORT = "X-Android-Selected-Transport";
     private static final String TAG = "JavaUrlConnection";
     private static final int DEFAULT_UPLOAD_BUFFER_SIZE = 8192;
     private static final int DEFAULT_CHUNK_LENGTH = DEFAULT_UPLOAD_BUFFER_SIZE;
@@ skipping 68 matching lines @@
         STARTED,
         REDIRECT_RECEIVED,
         AWAITING_FOLLOW_REDIRECT,
         AWAITING_READ,
         READING,
         ERROR,
         COMPLETE,
         CANCELLED,
     }
 
+    // Executor that runs one task at a time on an underlying Executor.
+    private final class SeriaizingExecutor implements Executor {
+        private final Executor mUnderlyingExecutor;
+        // Queue of tasks to run.  First element is the task currently executing.
+        // Task processing loop is running if queue is not empty.  Synchronized on itself.
+        @GuardedBy("mTaskQueue")
+        private final ArrayDeque<Runnable> mTaskQueue = new ArrayDeque<>();
+
+        SeriaizingExecutor(Executor underlyingExecutor) {
+            mUnderlyingExecutor = underlyingExecutor;
+        }
+
+        private void runTask(final Runnable task) {
+            try {
+                mUnderlyingExecutor.execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        try {
+                            task.run();
+                        } finally {
+                            Runnable nextTask;
+                            synchronized (mTaskQueue) {
+                                mTaskQueue.removeFirst();
+                                nextTask = mTaskQueue.peekFirst();
+                            }
+                            if (nextTask != null) {
+                                runTask(nextTask);
+                            }
+                        }
+                    }
+                });
+            } catch (RejectedExecutionException e) {
+                // This can happen if JavaCronetEngine.shutdown() was called.
+                // Ignore and cease processing this request.
+            }
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            synchronized (mTaskQueue) {
+                mTaskQueue.addLast(command);
+                // Task processing loop is already running if there are other items in mTaskQueue.
+                if (mTaskQueue.size() > 1) {
+                    return;
+                }
+            }
+            runTask(command);
+        };
+    }
+
     /**
      * @param executor The executor used for reading and writing from sockets
      * @param userExecutor The executor used to dispatch to {@code callback}
      */
     JavaUrlRequest(Callback callback, final Executor executor, Executor userExecutor, String url,
             String userAgent, boolean allowDirectExecutor) {
         if (url == null) {
             throw new NullPointerException("URL is required");
         }
         if (callback == null) {
             throw new NullPointerException("Listener is required");
         }
         if (executor == null) {
             throw new NullPointerException("Executor is required");
         }
         if (userExecutor == null) {
             throw new NullPointerException("userExecutor is required");
         }
 
         this.mAllowDirectExecutor = allowDirectExecutor;
         this.mCallbackAsync = new AsyncUrlRequestCallback(callback, userExecutor);
         this.mTrafficStatsTag = TrafficStats.getThreadStatsTag();
-        this.mExecutor = new Executor() {
+        this.mExecutor = new SeriaizingExecutor(new Executor() {
             @Override
             public void execute(final Runnable command) {
                 executor.execute(new Runnable() {
                     @Override
                     public void run() {
                         int oldTag = TrafficStats.getThreadStatsTag();
                         TrafficStats.setThreadStatsTag(mTrafficStatsTag);
                         try {
                             command.run();
                         } finally {
                             TrafficStats.setThreadStatsTag(oldTag);
                         }
                     }
                 });
             }
-        };
+        });
         this.mCurrentUrl = url;
         this.mUserAgent = userAgent;
     }
 
     @Override
     public void setHttpMethod(String method) {
         checkNotStarted();
         if (method == null) {
             throw new NullPointerException("Method is required.");
         }
@@ skipping 794 matching lines @@
                     // Can't throw directly from here, since the delegate executor could catch this
                     // exception.
                     mExecutedInline = new InlineExecutionProhibitedException();
                     return;
                 }
                 mCommand.run();
             }
         }
     }
 }