Allow direct executors in cronet.

components/cronet/android/api/src/org/chromium/net/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;
 
 import android.annotation.TargetApi;
 import android.net.TrafficStats;
 import android.os.Build;
 import android.util.Log;
@@ skipping 38 matching lines @@
     /**
      * This is the source of thread safety in this class - no other synchronization is performed.
      * By compare-and-swapping from one state to another, we guarantee that operations aren't
      * running concurrently. Only the winner of a CAS proceeds.
      *
      * <p>A caller can lose a CAS for three reasons - user error (two calls to read() without
      * waiting for the read to succeed), runtime error (network code or user code throws an
      * 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
      * is created, so that applications doing work on behalf of another app can correctly attribute
      * 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
      * {@link Status#SENDING_REQUEST} or {@link Status#WAITING_FOR_RESPONSE}. While the distinction
      * isn't needed to implement the logic in this class, it is needed to implement
      * {@link #getStatus(StatusListener)}.
      *
      * <p>Concurrency notes - this value is not atomically updated with mState, so there is some
      * risk that we'd get an inconsistent snapshot of both - however, it also happens that this
      * value is only used with the STARTED state, so it's inconsequential.
@@ skipping 31 matching lines @@
         ERROR,
         COMPLETE,
         CANCELLED,
     }
 
     /**
      * @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) {
+            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");
         }
         if (callback == null) {
             throw new NullPointerException("Listener is required");
         }
         if (executor == null) {
             throw new NullPointerException("Executor is required");
         }
         if (userExecutor == null) {
@@ skipping 96 matching lines @@
         }
         if (!mRequestHeaders.containsKey("Content-Type")) {
             throw new IllegalArgumentException(
                     "Requests with upload data must have a Content-Type.");
         }
         checkNotStarted();
         if (mInitialMethod == null) {
             mInitialMethod = "POST";
         }
         this.mUploadDataProvider = uploadDataProvider;
-        this.mUploadExecutor = executor;
+        if (mAllowDirectExecutor) {
+            this.mUploadExecutor = executor;
+        } else {
+            this.mUploadExecutor = new DirectPreventingExecutor(executor);
+        }
     }
 
     private enum SinkState {
         AWAITING_READ_RESULT,
         AWAITING_REWIND_RESULT,
         UPLOADING,
         NOT_STARTED,
     }
 
     private final class OutputStreamDataSink implements UploadDataSink {
@@ skipping 41 matching lines @@
                                 "Read upload data length %d exceeds expected length %d",
                                 mWrittenBytes + mBuffer.remaining(), mTotalBytes)));
                         return;
                     }
                     while (mBuffer.hasRemaining()) {
                         mWrittenBytes += mOutputChannel.write(mBuffer);
                     }
                     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) {
                         finish();
                     } else {
                         enterUploadErrorState(new IllegalArgumentException(String.format(
                                 "Read upload data length %d exceeds expected length %d",
                                 mWrittenBytes, mTotalBytes)));
                     }
                 }
@@ skipping 22 matching lines @@
             mExecutor.execute(errorSetting(State.STARTED, new CheckedRunnable() {
                 @Override
                 public void run() throws Exception {
                     if (mOutputChannel == null) {
                         mAdditionalStatusDetails = Status.CONNECTING;
                         mUrlConnection.connect();
                         mAdditionalStatusDetails = Status.SENDING_REQUEST;
                         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();
             }
             fireGetHeaders();
         }
 
         void start(final boolean firstTime) {
-            mUserExecutor.execute(uploadErrorSetting(new CheckedRunnable() {
+            executeOnUploadExecutor(new CheckedRunnable() {
                 @Override
                 public void run() throws Exception {
                     mTotalBytes = mUploadProvider.getLength();
                     if (mTotalBytes == 0) {
                         finish();
                     } else {
                         // If we know how much data we have to upload, and it's small, we can save
                         // memory by allocating a reasonably sized buffer to read into.
                         if (mTotalBytes > 0 && mTotalBytes < DEFAULT_UPLOAD_BUFFER_SIZE) {
                             // Allocate one byte more than necessary, to detect callers uploading
@@ skipping 15 matching lines @@
                             mUrlConnection.setChunkedStreamingMode(DEFAULT_CHUNK_LENGTH);
                         }
                         if (firstTime) {
                             startRead();
                         } else {
                             mSinkState.set(SinkState.AWAITING_REWIND_RESULT);
                             mUploadProvider.rewind(OutputStreamDataSink.this);
                         }
                     }
                 }
-            }));
+            });
         }
     }
 
     @Override
     public void start() {
         mAdditionalStatusDetails = Status.CONNECTING;
         transitionStates(State.NOT_STARTED, State.STARTED, new Runnable() {
             @Override
             public void run() {
                 mUrlChain.add(mCurrentUrl);
@@ skipping 322 matching lines @@
                 throw new IllegalStateException("Switch is exhaustive: " + state);
         }
 
         mCallbackAsync.sendStatus(listener, status);
     }
 
     /** This wrapper ensures that callbacks are always called on the correct executor */
     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) {
             mUserExecutor.execute(new Runnable() {
                 @Override
                 public void run() {
                     listener.onStatus(status);
                 }
             });
         }
 
         void execute(State currentState, CheckedRunnable runnable) {
             try {
                 mUserExecutor.execute(userErrorSetting(currentState, runnable));
             } catch (RejectedExecutionException e) {
-                enterUserErrorState(currentState, e);
+                enterErrorState(currentState,
+                        new UrlRequestException("Exception posting task to executor", e));
             }
         }
 
         void onRedirectReceived(final UrlResponseInfo info, final String newLocationUrl) {
             execute(State.AWAITING_FOLLOW_REDIRECT, new CheckedRunnable() {
                 @Override
                 public void run() throws Exception {
                     mCallback.onRedirectReceived(JavaUrlRequest.this, info, newLocationUrl);
                 }
             });
         }
 
         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) {
             closeQuietly(mResponseChannel);
             mUserExecutor.execute(new Runnable() {
                 @Override
                 public void run() {
                     try {
                         mCallback.onCanceled(JavaUrlRequest.this, info);
                     } catch (Exception exception) {
@@ skipping 11 matching lines @@
                         mCallback.onSucceeded(JavaUrlRequest.this, info);
                     } catch (Exception exception) {
                         Log.e(TAG, "Exception in onSucceeded method", exception);
                     }
                 }
             });
         }
 
         void onFailed(final UrlResponseInfo urlResponseInfo, final UrlRequestException e) {
             closeQuietly(mResponseChannel);
-            mUserExecutor.execute(new Runnable() {
+            Runnable runnable = new Runnable() {
                 @Override
                 public void run() {
                     try {
                         mCallback.onFailed(JavaUrlRequest.this, urlResponseInfo, e);
                     } catch (Exception exception) {
                         Log.e(TAG, "Exception in onFailed method", exception);
                     }
                 }
-            });
+            };
+            try {
+                mUserExecutor.execute(runnable);
+            } catch (InlineExecutionProhibitedException wasDirect) {
+                if (mFallbackExecutor != null) {
+                    mFallbackExecutor.execute(runnable);
+                }
+            }
         }
     }
 
     private static void closeQuietly(Closeable closeable) {
         if (closeable == null) {
             return;
         }
         try {
             closeable.close();
         } catch (IOException e) {
             e.printStackTrace();
         }
     }
 }