2.x: option to fail for using blockingX on the computation scheduler (#5020)

akarnokd · web-flow · commit 307f97b6fbe7 · 2017-01-27T09:08:17.000+01:00
* 2.x: option to fail for using blockingX on the computation scheduler

* Increase sleep time in XFlatMapTest

* Add a custom RxJavaPlugins callback onBeforeBlocking
diff --git a/src/main/java/io/reactivex/internal/observers/BlockingBaseObserver.java b/src/main/java/io/reactivex/internal/observers/BlockingBaseObserver.java
@@ -16,7 +16,7 @@
 
 import io.reactivex.Observer;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 
 public abstract class BlockingBaseObserver<T> extends CountDownLatch
 implements Observer<T>, Disposable {
@@ -67,6 +67,7 @@ public final boolean isDisposed() {
     public final T blockingGet() {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 await();
             } catch (InterruptedException ex) {
                 dispose();
diff --git a/src/main/java/io/reactivex/internal/observers/BlockingMultiObserver.java b/src/main/java/io/reactivex/internal/observers/BlockingMultiObserver.java
@@ -17,7 +17,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 
 /**
  * A combined Observer that awaits the success or error signal via a CountDownLatch.
@@ -79,6 +79,7 @@ public void onComplete() {
     public T blockingGet() {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 await();
             } catch (InterruptedException ex) {
                 dispose();
@@ -101,6 +102,7 @@ public T blockingGet() {
     public T blockingGet(T defaultValue) {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 await();
             } catch (InterruptedException ex) {
                 dispose();
@@ -123,6 +125,7 @@ public T blockingGet(T defaultValue) {
     public Throwable blockingGetError() {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 await();
             } catch (InterruptedException ex) {
                 dispose();
@@ -142,6 +145,7 @@ public Throwable blockingGetError() {
     public Throwable blockingGetError(long timeout, TimeUnit unit) {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 if (!await(timeout, unit)) {
                     dispose();
                     throw ExceptionHelper.wrapOrThrow(new TimeoutException());
@@ -164,6 +168,7 @@ public Throwable blockingGetError(long timeout, TimeUnit unit) {
     public boolean blockingAwait(long timeout, TimeUnit unit) {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 if (!await(timeout, unit)) {
                     dispose();
                     return false;
diff --git a/src/main/java/io/reactivex/internal/observers/FutureObserver.java b/src/main/java/io/reactivex/internal/observers/FutureObserver.java
@@ -20,6 +20,7 @@
 import io.reactivex.Observer;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.util.BlockingHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -72,6 +73,7 @@ public boolean isDone() {
     @Override
     public T get() throws InterruptedException, ExecutionException {
         if (getCount() != 0) {
+            BlockingHelper.verifyNonBlocking();
             await();
         }
 
@@ -88,6 +90,7 @@ public T get() throws InterruptedException, ExecutionException {
     @Override
     public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
         if (getCount() != 0) {
+            BlockingHelper.verifyNonBlocking();
             if (!await(timeout, unit)) {
                 throw new TimeoutException();
             }
diff --git a/src/main/java/io/reactivex/internal/observers/FutureSingleObserver.java b/src/main/java/io/reactivex/internal/observers/FutureSingleObserver.java
@@ -19,6 +19,7 @@
 import io.reactivex.SingleObserver;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.util.BlockingHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -71,6 +72,7 @@ public boolean isDone() {
     @Override
     public T get() throws InterruptedException, ExecutionException {
         if (getCount() != 0) {
+            BlockingHelper.verifyNonBlocking();
             await();
         }
 
@@ -87,6 +89,7 @@ public T get() throws InterruptedException, ExecutionException {
     @Override
     public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
         if (getCount() != 0) {
+            BlockingHelper.verifyNonBlocking();
             if (!await(timeout, unit)) {
                 throw new TimeoutException();
             }
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java b/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java
@@ -23,7 +23,7 @@
 import io.reactivex.exceptions.MissingBackpressureException;
 import io.reactivex.internal.queue.SpscArrayQueue;
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 
 public final class BlockingFlowableIterable<T> implements Iterable<T> {
     final Publisher<? extends T> source;
@@ -86,6 +86,7 @@ public boolean hasNext() {
                     }
                 }
                 if (empty) {
+                    BlockingHelper.verifyNonBlocking();
                     lock.lock();
                     try {
                         while (!done && queue.isEmpty()) {
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableLatest.java b/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableLatest.java
@@ -20,7 +20,7 @@
 import org.reactivestreams.Publisher;
 
 import io.reactivex.*;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subscribers.DisposableSubscriber;
 
@@ -79,6 +79,7 @@ public boolean hasNext() {
             if (iteratorNotification == null || iteratorNotification.isOnNext()) {
                 if (iteratorNotification == null) {
                     try {
+                        BlockingHelper.verifyNonBlocking();
                         notify.acquire();
                     } catch (InterruptedException ex) {
                         dispose();
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableNext.java b/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableNext.java
@@ -20,7 +20,7 @@
 import org.reactivestreams.Publisher;
 
 import io.reactivex.*;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subscribers.DisposableSubscriber;
 
@@ -165,6 +165,7 @@ public void onNext(Notification<T> args) {
 
         public Notification<T> takeNext() throws InterruptedException {
             setWaiting();
+            BlockingHelper.verifyNonBlocking();
             return buf.take();
         }
         void setWaiting() {
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableBlockingSubscribe.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableBlockingSubscribe.java
@@ -57,6 +57,7 @@ public static <T> void subscribe(Publisher<? extends T> o, Subscriber<? super T>
                     if (bs.isCancelled()) {
                         break;
                     }
+                    BlockingHelper.verifyNonBlocking();
                     v = queue.take();
                 }
                 if (bs.isCancelled()) {
diff --git a/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableIterable.java b/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableIterable.java
@@ -21,7 +21,7 @@
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.queue.SpscLinkedArrayQueue;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 
 public final class BlockingObservableIterable<T> implements Iterable<T> {
     final ObservableSource<? extends T> source;
@@ -78,6 +78,7 @@ public boolean hasNext() {
                 }
                 if (empty) {
                     try {
+                        BlockingHelper.verifyNonBlocking();
                         lock.lock();
                         try {
                             while (!done && queue.isEmpty()) {
diff --git a/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableLatest.java b/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableLatest.java
@@ -19,7 +19,7 @@
 
 import io.reactivex.*;
 import io.reactivex.Observable;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 import io.reactivex.observers.DisposableObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 
@@ -79,6 +79,7 @@ public boolean hasNext() {
             }
             if (iteratorNotification == null) {
                 try {
+                    BlockingHelper.verifyNonBlocking();
                     notify.acquire();
                 } catch (InterruptedException ex) {
                     dispose();
diff --git a/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableNext.java b/src/main/java/io/reactivex/internal/operators/observable/BlockingObservableNext.java
@@ -18,7 +18,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import io.reactivex.*;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 import io.reactivex.observers.DisposableObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 
@@ -162,6 +162,7 @@ public void onNext(Notification<T> args) {
 
         public Notification<T> takeNext() throws InterruptedException {
             setWaiting();
+            BlockingHelper.verifyNonBlocking();
             return buf.take();
         }
         void setWaiting() {
diff --git a/src/main/java/io/reactivex/internal/schedulers/ComputationScheduler.java b/src/main/java/io/reactivex/internal/schedulers/ComputationScheduler.java
@@ -56,7 +56,7 @@ public final class ComputationScheduler extends Scheduler {
         int priority = Math.max(Thread.MIN_PRIORITY, Math.min(Thread.MAX_PRIORITY,
                 Integer.getInteger(KEY_COMPUTATION_PRIORITY, Thread.NORM_PRIORITY)));
 
-        THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX, priority);
+        THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX, priority, true);
 
         NONE = new FixedSchedulerPool(0, THREAD_FACTORY);
         NONE.shutdown();
diff --git a/src/main/java/io/reactivex/internal/schedulers/NonBlockingThread.java b/src/main/java/io/reactivex/internal/schedulers/NonBlockingThread.java
@@ -0,0 +1,22 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.internal.schedulers;
+
+/**
+ * Marker interface to indicate blocking is not recommended while running
+ * on a Scheduler with a thread type implementing it.
+ */
+public interface NonBlockingThread {
+
+}
diff --git a/src/main/java/io/reactivex/internal/schedulers/RxThreadFactory.java b/src/main/java/io/reactivex/internal/schedulers/RxThreadFactory.java
@@ -28,15 +28,22 @@ public final class RxThreadFactory extends AtomicLong implements ThreadFactory {
 
     final int priority;
 
+    final boolean nonBlocking;
+
 //    static volatile boolean CREATE_TRACE;
 
     public RxThreadFactory(String prefix) {
-        this(prefix, Thread.NORM_PRIORITY);
+        this(prefix, Thread.NORM_PRIORITY, false);
     }
 
     public RxThreadFactory(String prefix, int priority) {
+        this(prefix, priority, false);
+    }
+
+    public RxThreadFactory(String prefix, int priority, boolean nonBlocking) {
         this.prefix = prefix;
         this.priority = priority;
+        this.nonBlocking = nonBlocking;
     }
 
     @Override
@@ -63,7 +70,8 @@ public Thread newThread(Runnable r) {
 //            }
 //        }
 
-        Thread t = new Thread(r, nameBuilder.toString());
+        String name = nameBuilder.toString();
+        Thread t = nonBlocking ? new RxCustomThread(r, name) : new Thread(r, name);
         t.setPriority(priority);
         t.setDaemon(true);
         return t;
@@ -73,4 +81,10 @@ public Thread newThread(Runnable r) {
     public String toString() {
         return "RxThreadFactory[" + prefix + "]";
     }
+
+    static final class RxCustomThread extends Thread implements NonBlockingThread {
+        RxCustomThread(Runnable run, String name) {
+            super(run, name);
+        }
+    }
 }
diff --git a/src/main/java/io/reactivex/internal/schedulers/SingleScheduler.java b/src/main/java/io/reactivex/internal/schedulers/SingleScheduler.java
@@ -44,7 +44,7 @@ public final class SingleScheduler extends Scheduler {
         int priority = Math.max(Thread.MIN_PRIORITY, Math.min(Thread.MAX_PRIORITY,
                 Integer.getInteger(KEY_SINGLE_PRIORITY, Thread.NORM_PRIORITY)));
 
-        SINGLE_THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX, priority);
+        SINGLE_THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX, priority, true);
     }
 
     public SingleScheduler() {
diff --git a/src/main/java/io/reactivex/internal/subscribers/BlockingBaseSubscriber.java b/src/main/java/io/reactivex/internal/subscribers/BlockingBaseSubscriber.java
@@ -17,7 +17,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
-import io.reactivex.internal.util.ExceptionHelper;
+import io.reactivex.internal.util.*;
 
 public abstract class BlockingBaseSubscriber<T> extends CountDownLatch
 implements Subscriber<T> {
@@ -60,6 +60,7 @@ public final void onComplete() {
     public final T blockingGet() {
         if (getCount() != 0) {
             try {
+                BlockingHelper.verifyNonBlocking();
                 await();
             } catch (InterruptedException ex) {
                 Subscription s = this.s;
diff --git a/src/main/java/io/reactivex/internal/subscribers/FutureSubscriber.java b/src/main/java/io/reactivex/internal/subscribers/FutureSubscriber.java
@@ -20,6 +20,7 @@
 import org.reactivestreams.*;
 
 import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.internal.util.BlockingHelper;
 import io.reactivex.plugins.RxJavaPlugins;
 
 /**
@@ -72,6 +73,7 @@ public boolean isDone() {
     @Override
     public T get() throws InterruptedException, ExecutionException {
         if (getCount() != 0) {
+            BlockingHelper.verifyNonBlocking();
             await();
         }
 
@@ -88,6 +90,7 @@ public T get() throws InterruptedException, ExecutionException {
     @Override
     public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
         if (getCount() != 0) {
+            BlockingHelper.verifyNonBlocking();
             if (!await(timeout, unit)) {
                 throw new TimeoutException();
             }
diff --git a/src/main/java/io/reactivex/internal/util/BlockingHelper.java b/src/main/java/io/reactivex/internal/util/BlockingHelper.java
@@ -16,6 +16,8 @@
 import java.util.concurrent.CountDownLatch;
 
 import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.schedulers.NonBlockingThread;
+import io.reactivex.plugins.RxJavaPlugins;
 
 /**
  * Utility methods for helping common blocking operations.
@@ -34,6 +36,7 @@ public static void awaitForComplete(CountDownLatch latch, Disposable subscriptio
         }
         // block until the subscription completes and then return
         try {
+            verifyNonBlocking();
             latch.await();
         } catch (InterruptedException e) {
             subscription.dispose();
@@ -45,5 +48,16 @@ public static void awaitForComplete(CountDownLatch latch, Disposable subscriptio
         }
     }
 
-
+    /**
+     * Checks if the {@code failOnNonBlockingScheduler} plugin setting is enabled and the current
+     * thread is a Scheduler sensitive to blocking operators.
+     * @throws IllegalStateException if the {@code failOnNonBlockingScheduler} and the current thread is sensitive to blocking
+     */
+    public static void verifyNonBlocking() {
+        if (RxJavaPlugins.isFailOnNonBlockingScheduler()
+                && (Thread.currentThread() instanceof NonBlockingThread
+                        || RxJavaPlugins.onBeforeBlocking())) {
+            throw new IllegalStateException("Attempt to block on a Scheduler " + Thread.currentThread().getName() + " that doesn't support blocking operators as they may lead to deadlock");
+        }
+    }
 }
diff --git a/src/main/java/io/reactivex/plugins/RxJavaPlugins.java b/src/main/java/io/reactivex/plugins/RxJavaPlugins.java
diff --git a/src/test/java/io/reactivex/XFlatMapTest.java b/src/test/java/io/reactivex/XFlatMapTest.java
diff --git a/src/test/java/io/reactivex/plugins/RxJavaPluginsTest.java b/src/test/java/io/reactivex/plugins/RxJavaPluginsTest.java
diff --git a/src/test/java/io/reactivex/schedulers/FailOnBlockingTest.java b/src/test/java/io/reactivex/schedulers/FailOnBlockingTest.java
diff --git a/src/test/java/io/reactivex/tck/DelaySubscriptionTckTest.java b/src/test/java/io/reactivex/tck/DelaySubscriptionTckTest.java

Original file line number	Diff line number	Diff line change
`@@ -57,6 +57,7 @@ public static <T> void subscribe(Publisher<? extends T> o, Subscriber<? super T>`
`57`	`57`	`if (bs.isCancelled()) {`
`58`	`58`	`break;`
`59`	`59`	`}`
	`60`	`+ BlockingHelper.verifyNonBlocking();`
`60`	`61`	`v = queue.take();`
`61`	`62`	`}`
`62`	`63`	`if (bs.isCancelled()) {`