Allow direct executors in cronet.

components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java

Index: components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java
diff --git a/components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java b/components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java
index fe562799cda82ac2dda37922425f7e4ca9ba1ad6..f7126c2c5916215f99265623ae94c824a762a9a0 100644
--- a/components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java
+++ b/components/cronet/android/api/src/org/chromium/net/JavaUrlRequest.java
@@ -56,6 +56,7 @@ final class JavaUrlRequest implements UrlRequest {
      * exception), or cancellation.
      */
     private final AtomicReference<State> mState = new AtomicReference<>(State.NOT_STARTED);
+    private final AtomicBoolean mUploadProviderClosed = new AtomicBoolean(false);
 
     /**
      * Traffic stats tag to associate this requests' data use with. It's captured when the request
@@ -63,12 +64,12 @@ final class JavaUrlRequest implements UrlRequest {
      * that data use.
      */
     private final int mTrafficStatsTag;
+    private final boolean mAllowDirectExecutor;
 
     /* These don't change with redirects */
     private String mInitialMethod;
     private UploadDataProvider mUploadDataProvider;
     private Executor mUploadExecutor;
-    private AtomicBoolean mUploadProviderClosed = new AtomicBoolean(false);
 
     /**
      * Holds a subset of StatusValues - {@link State#STARTED} can represent
@@ -120,7 +121,8 @@ final class JavaUrlRequest implements UrlRequest {
      * @param userExecutor The executor used to dispatch to {@code callback}
      */
     JavaUrlRequest(Callback callback, final Executor executor, Executor userExecutor, String url,
-            String userAgent) {
+            String userAgent, boolean allowDirectExecutor) {
+        this.mAllowDirectExecutor = allowDirectExecutor;

[mef, 2016/08/30 16:29:44]

nit: move it to the end, below userAgent?

[Charles, 2016/08/30 20:11:06]

This value gets read by AsyncUrlRequestCallback. Moved there.

         if (url == null) {
             throw new NullPointerException("URL is required");
         }
@@ -237,7 +239,11 @@ final class JavaUrlRequest implements UrlRequest {
             mInitialMethod = "POST";
         }
         this.mUploadDataProvider = uploadDataProvider;
-        this.mUploadExecutor = executor;
+        if (mAllowDirectExecutor) {
+            this.mUploadExecutor = executor;
+        } else {
+            this.mUploadExecutor = new DirectPreventingExecutor(executor);
+        }
     }
 
     private enum SinkState {
@@ -299,12 +305,12 @@ final class JavaUrlRequest implements UrlRequest {
                     if (mWrittenBytes < mTotalBytes || (mTotalBytes == -1 && !finalChunk)) {
                         mBuffer.clear();
                         mSinkState.set(SinkState.AWAITING_READ_RESULT);
-                        mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+                        executeOnUploadExecutor(new CheckedRunnable() {
                             @Override
                             public void run() throws Exception {
                                 mUploadProvider.read(OutputStreamDataSink.this, mBuffer);
                             }
-                        }));
+                        });
                     } else if (mTotalBytes == -1) {
                         finish();
                     } else if (mTotalBytes == mWrittenBytes) {
@@ -347,16 +353,24 @@ final class JavaUrlRequest implements UrlRequest {
                         mOutputChannel = Channels.newChannel(mUrlConnection.getOutputStream());
                     }
                     mSinkState.set(SinkState.AWAITING_READ_RESULT);
-                    mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+                    executeOnUploadExecutor(new CheckedRunnable() {
                         @Override
                         public void run() throws Exception {
                             mUploadProvider.read(OutputStreamDataSink.this, mBuffer);
                         }
-                    }));
+                    });
                 }
             }));
         }
 
+        private void executeOnUploadExecutor(CheckedRunnable runnable) {
+            try {
+                mUserExecutor.execute(uploadErrorSetting(runnable));

[mef, 2016/08/30 16:29:44]

Should this execute on mUploadExecutor instead of mUserExecutor?
We probably need a companion method to executeOnUserExecutor().

[Charles, 2016/08/30 20:11:05]

In this lexical scope mUserExecutor is referring to the mUserExecutor of
OutputStreamDataSink, which is the upload executor. There is no field for the
user executor except in AsyncUrlRequestCallback.

[mef, 2016/08/31 17:26:52]

I see. I've missed the fact that this is part of OutputStreamDataSink and not
JavaUrlRequest. It is still mildly confusing that executeOnUploadExecutor is
running tasks on |mUserExecutor|. Maybe rename mUserExecutor ->
mUserUploadExecutor or mUploadExecutor? I don't feel strongly though.

[Charles, 2016/08/31 19:08:36]

Done.

+            } catch (RejectedExecutionException e) {
+                enterUploadErrorState(e);
+            }
+        }
+
         void finish() throws IOException {
             if (mOutputChannel != null) {
                 mOutputChannel.close();
@@ -365,7 +379,7 @@ final class JavaUrlRequest implements UrlRequest {
         }
 
         void start(final boolean firstTime) {
-            mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+            executeOnUploadExecutor(new CheckedRunnable() {
                 @Override
                 public void run() throws Exception {
                     mTotalBytes = mUploadProvider.getLength();
@@ -401,7 +415,7 @@ final class JavaUrlRequest implements UrlRequest {
                         }
                     }
                 }
-            }));
+            });
         }
     }
 
@@ -744,10 +758,17 @@ final class JavaUrlRequest implements UrlRequest {
     private final class AsyncUrlRequestCallback {
         final UrlRequest.Callback mCallback;
         final Executor mUserExecutor;
+        final Executor mFallbackExecutor;
 
         AsyncUrlRequestCallback(Callback callback, final Executor userExecutor) {
             this.mCallback = callback;
-            this.mUserExecutor = userExecutor;
+            if (mAllowDirectExecutor) {
+                this.mUserExecutor = userExecutor;
+                this.mFallbackExecutor = null;
+            } else {
+                mUserExecutor = new DirectPreventingExecutor(userExecutor);
+                mFallbackExecutor = userExecutor;
+            }
         }
 
         void sendStatus(final StatusListener listener, final int status) {
@@ -763,7 +784,8 @@ final class JavaUrlRequest implements UrlRequest {
             try {
                 mUserExecutor.execute(userErrorSetting(currentState, runnable));
             } catch (RejectedExecutionException e) {
-                enterUserErrorState(currentState, e);
+                enterErrorState(currentState,
+                        new UrlRequestException("Exception posting task to executor", e));
             }
         }
 
@@ -777,25 +799,25 @@ final class JavaUrlRequest implements UrlRequest {
         }
 
         void onResponseStarted(UrlResponseInfo info) {
-            execute(State.AWAITING_READ, new CheckedRunnable() {
-                @Override
-                public void run() throws Exception {
-                    if (mState.compareAndSet(State.STARTED, State.AWAITING_READ)) {
+            if (mState.compareAndSet(State.STARTED, State.AWAITING_READ)) {

[mef, 2016/08/30 16:29:44]

This is changing thread where mState is accessed from executor.
Why is it needed and why is it safe?

[Charles, 2016/08/30 20:11:06]

mState is safe from any thread, since it's an atomic reference. This transition
is necessary to handle rejected execution exceptions properly, since our state
transition relies on knowing what the state is before we invoke something that
could trigger a user error. This code is now handling 

[mef, 2016/08/31 17:26:52]

Acknowledged.

+                execute(State.AWAITING_READ, new CheckedRunnable() {
+                    @Override
+                    public void run() throws Exception {
                         mCallback.onResponseStarted(JavaUrlRequest.this, mUrlResponseInfo);
                     }
-                }
-            });
+                });
+            }
         }
 
         void onReadCompleted(final UrlResponseInfo info, final ByteBuffer byteBuffer) {
-            execute(State.AWAITING_READ, new CheckedRunnable() {
-                @Override
-                public void run() throws Exception {
-                    if (mState.compareAndSet(State.READING, State.AWAITING_READ)) {
+            if (mState.compareAndSet(State.READING, State.AWAITING_READ)) {

[mef, 2016/08/30 16:29:44]

ditto.

[Charles, 2016/08/30 20:11:05]

This one too - we want a REE thrown when submitting to count as a user failure,
and we want to be in the correct state before calling execute().

+                execute(State.AWAITING_READ, new CheckedRunnable() {
+                    @Override
+                    public void run() throws Exception {
                         mCallback.onReadCompleted(JavaUrlRequest.this, info, byteBuffer);
                     }
-                }
-            });
+                });
+            }
         }
 
         void onCanceled(final UrlResponseInfo info) {
@@ -827,7 +849,7 @@ final class JavaUrlRequest implements UrlRequest {
 
         void onFailed(final UrlResponseInfo urlResponseInfo, final UrlRequestException e) {
             closeQuietly(mResponseChannel);
-            mUserExecutor.execute(new Runnable() {
+            Runnable runnable = new Runnable() {
                 @Override
                 public void run() {
                     try {
@@ -836,7 +858,14 @@ final class JavaUrlRequest implements UrlRequest {
                         Log.e(TAG, "Exception in onFailed method", exception);
                     }
                 }
-            });
+            };
+            try {
+                mUserExecutor.execute(runnable);
+            } catch (InlineExecutionProhibitedException wasDirect) {
+                if (mFallbackExecutor != null) {
+                    mFallbackExecutor.execute(runnable);
+                }
+            }
         }
     }