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) {
         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() {
             @Override
             public void execute(final Runnable command) {
                 executor.execute(new Runnable() {
                     @Override
                     public void run() {
                         int oldTag = TrafficStats.getThreadStatsTag();
                         TrafficStats.setThreadStatsTag(mTrafficStatsTag);
@@ skipping 84 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 {
         final AtomicReference<SinkState> mSinkState = new AtomicReference<>(SinkState.NOT_STARTED);
-        final Executor mUserExecutor;
+        final Executor mUserUploadExecutor;
         final Executor mExecutor;
         final HttpURLConnection mUrlConnection;
         WritableByteChannel mOutputChannel;
         final UploadDataProvider mUploadProvider;
         ByteBuffer mBuffer;
         /** This holds the total bytes to send (the content-length). -1 if unknown. */
         long mTotalBytes;
         /** This holds the bytes written so far */
         long mWrittenBytes = 0;
 
         OutputStreamDataSink(final Executor userExecutor, Executor executor,
                 HttpURLConnection urlConnection, UploadDataProvider provider) {
-            this.mUserExecutor = new Executor() {
+            this.mUserUploadExecutor = new Executor() {
                 @Override
                 public void execute(Runnable runnable) {
                     try {
                         userExecutor.execute(runnable);
                     } catch (RejectedExecutionException e) {
                         enterUploadErrorState(e);
                     }
                 }
             };
             this.mExecutor = executor;
@@ skipping 16 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 {
+                mUserUploadExecutor.execute(uploadErrorSetting(runnable));
+            } 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)) {
+                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)) {
+                execute(State.AWAITING_READ, new CheckedRunnable() {
+                    @Override
+                    public void run() throws Exception {
                         mCallback.onReadCompleted(JavaUrlRequest.this, info, byteBuffer);
                     }
-                }
-            });
+                });
+            }
         }
 
         void onCanceled(final UrlResponseInfo info) {
             closeResponseChannel();
             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) {
             closeResponseChannel();
-            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 void closeResponseChannel() {
         final Closeable closeable = mResponseChannel;
         if (closeable == null) {
             return;
         }
         mResponseChannel = null;
         mExecutor.execute(new Runnable() {
             @Override
             public void run() {
                 try {
                     closeable.close();
                 } catch (IOException e) {
                     e.printStackTrace();
                 }
             }
         });
     }
+
+    /**
+     * Executor that detects and throws if its mDelegate runs a submitted runnable inline.
+     */
+    static final class DirectPreventingExecutor implements Executor {
+        private final Executor mDelegate;
+
+        DirectPreventingExecutor(Executor delegate) {
+            this.mDelegate = delegate;
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            Thread currentThread = Thread.currentThread();
+            InlineCheckingRunnable runnable = new InlineCheckingRunnable(command, currentThread);
+            mDelegate.execute(runnable);
+            if (runnable.mExecutedInline != null) {
+                throw runnable.mExecutedInline;
+            } else {
+                // It's possible that this method is being called on an executor, and the runnable
+                // that
+                // was just queued will run on this thread after the current runnable returns. By
+                // nulling out the mCallingThread field, the InlineCheckingRunnable's current thread
+                // comparison will not fire.
+                runnable.mCallingThread = null;
+            }
+        }
+
+        private static final class InlineCheckingRunnable implements Runnable {
+            private final Runnable mCommand;
+            private Thread mCallingThread;
+            private InlineExecutionProhibitedException mExecutedInline = null;
+
+            private InlineCheckingRunnable(Runnable command, Thread callingThread) {
+                this.mCommand = command;
+                this.mCallingThread = callingThread;
+            }
+
+            @Override
+            public void run() {
+                if (Thread.currentThread() == mCallingThread) {
+                    // Can't throw directly from here, since the delegate executor could catch this
+                    // exception.
+                    mExecutedInline = new InlineExecutionProhibitedException();
+                    return;
+                }
+                mCommand.run();
+            }
+        }
+    }
 }