From 980d51daca41702e2b99a50b6cdfd9be8859e75c Mon Sep 17 00:00:00 2001
From: Sergio Garcia <sergio@argent.xyz>
Date: Mon, 25 Jan 2021 11:39:01 +0000
Subject: [PATCH 1/3] Support for scheduled release of threads in Io Scheduler

---
 .../internal/schedulers/IoScheduler.java      | 21 ++++++++++++++++---
 .../rxjava3/schedulers/Schedulers.java        |  6 ++++++
 2 files changed, 24 insertions(+), 3 deletions(-)

diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java
index 6d63715770..0e0939cc58 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java
@@ -48,6 +48,10 @@ public final class IoScheduler extends Scheduler {
     /** The name of the system property for setting the thread priority for this Scheduler. */
     private static final String KEY_IO_PRIORITY = "rx3.io-priority";
 
+    /** The name of the system property for setting the release behaviour for this Scheduler. */
+    private static final String KEY_SCHEDULED_RELEASE = "rx3.io-scheduled-release";
+    private static final boolean USE_SCHEDULED_RELEASE;
+
     static final CachedWorkerPool NONE;
 
     static {
@@ -63,6 +67,8 @@ public final class IoScheduler extends Scheduler {
 
         EVICTOR_THREAD_FACTORY = new RxThreadFactory(EVICTOR_THREAD_NAME_PREFIX, priority);
 
+        USE_SCHEDULED_RELEASE = Boolean.getBoolean(KEY_SCHEDULED_RELEASE);
+
         NONE = new CachedWorkerPool(0, null, WORKER_THREAD_FACTORY);
         NONE.shutdown();
     }
@@ -194,7 +200,7 @@ public int size() {
         return pool.get().allWorkers.size();
     }
 
-    static final class EventLoopWorker extends Scheduler.Worker {
+    static final class EventLoopWorker extends Scheduler.Worker implements Runnable {
         private final CompositeDisposable tasks;
         private final CachedWorkerPool pool;
         private final ThreadWorker threadWorker;
@@ -212,11 +218,20 @@ public void dispose() {
             if (once.compareAndSet(false, true)) {
                 tasks.dispose();
 
-                // releasing the pool should be the last action
-                pool.release(threadWorker);
+                if (USE_SCHEDULED_RELEASE) {
+                    threadWorker.scheduleActual(this, 0, TimeUnit.NANOSECONDS, null);
+                } else {
+                    // releasing the pool should be the last action
+                    pool.release(threadWorker);
+                }
             }
         }
 
+        @Override
+        public void run() {
+            pool.release(threadWorker);
+        }
+
         @Override
         public boolean isDisposed() {
             return once.get();
diff --git a/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java b/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java
index ad7ff69bc7..90689654b9 100644
--- a/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java
+++ b/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java
@@ -32,6 +32,9 @@
  * <ul>
  * <li>{@code rx3.io-keep-alive-time} (long): sets the keep-alive time of the {@link #io()} Scheduler workers, default is {@link IoScheduler#KEEP_ALIVE_TIME_DEFAULT}</li>
  * <li>{@code rx3.io-priority} (int): sets the thread priority of the {@link #io()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
+ * <li>{@code rx3.io-scheduled-release} (boolean): Changes release behaviour mode for {@link #io()} Scheduler.
+ *   Enabling this guarantees that workers are not released before they finish all processing, therefore preventing deadlocks.
+ *   The drawback is that it can lead to more threads being created, since threads are released later. Default is false</li>
  * <li>{@code rx3.computation-threads} (int): sets the number of threads in the {@link #computation()} Scheduler, default is the number of available CPUs</li>
  * <li>{@code rx3.computation-priority} (int): sets the thread priority of the {@link #computation()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
  * <li>{@code rx3.newthread-priority} (int): sets the thread priority of the {@link #newThread()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
@@ -159,6 +162,9 @@ public static Scheduler computation() {
      * <ul>
      * <li>{@code rx3.io-keep-alive-time} (long): sets the keep-alive time of the {@code io()} Scheduler workers, default is {@link IoScheduler#KEEP_ALIVE_TIME_DEFAULT}</li>
      * <li>{@code rx3.io-priority} (int): sets the thread priority of the {@code io()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
+     * <li>{@code rx3.io-scheduled-release} (boolean): Changes release behaviour mode for {@code #io()} Scheduler.
+     *    Enabling this guarantees that workers are not released before they finish all processing, therefore preventing deadlocks.
+     *    The drawback is that it can lead to more threads being created, since threads are released later. Default is false</li>
      * </ul>
      * <p>
      * The default value of this scheduler can be overridden at initialization time via the

From 49a138bd249c8bd71b1deed4c939131e1e922e21 Mon Sep 17 00:00:00 2001
From: Sergio Garcia <sergio@argent.xyz>
Date: Mon, 25 Jan 2021 11:39:30 +0000
Subject: [PATCH 2/3] Test for rx3.io-scheduled-release

---
 .../schedulers/IoScheduledReleaseTest.java    | 45 +++++++++++++++++++
 1 file changed, 45 insertions(+)
 create mode 100644 src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java

diff --git a/src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java b/src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java
new file mode 100644
index 0000000000..9840cdb0eb
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java
@@ -0,0 +1,45 @@
+/**
+ * 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.rxjava3.schedulers;
+
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.RxJavaTest;
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+public class IoScheduledReleaseTest extends RxJavaTest {
+
+    /* This test will be stuck in a deadlock if rx3.io-scheduled-release is not set */
+    @Test
+    public void scheduledRelease() {
+        System.setProperty("rx3.io-scheduled-release", Boolean.TRUE.toString());
+        Flowable.just("item")
+                .observeOn(Schedulers.io())
+                .firstOrError()
+                .map(item -> {
+                    for (int i = 0; i < 50; i++) {
+                        Completable.complete()
+                                .observeOn(Schedulers.io())
+                                .blockingAwait();
+                    }
+                    return "Done";
+                })
+                .ignoreElement()
+                .test()
+                .awaitDone(5, TimeUnit.SECONDS)
+                .assertComplete();
+    }
+}

From 526d6006d3a03c526abfc138d6458b58bf0a6e38 Mon Sep 17 00:00:00 2001
From: Sergio Garcia <sergio@argent.xyz>
Date: Mon, 25 Jan 2021 13:06:46 +0000
Subject: [PATCH 3/3] Fix tests, addressed Javadoc feedback

---
 .../internal/schedulers/IoScheduler.java      |  2 +-
 .../rxjava3/schedulers/Schedulers.java        | 25 ++++++---
 .../schedulers/IoScheduledReleaseTest.java    | 51 +++++++++++++++++++
 .../schedulers/IoScheduledReleaseTest.java    | 45 ----------------
 4 files changed, 71 insertions(+), 52 deletions(-)
 create mode 100644 src/test/java/io/reactivex/rxjava3/internal/schedulers/IoScheduledReleaseTest.java
 delete mode 100644 src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java

diff --git a/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java b/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java
index 0e0939cc58..13c62fcb10 100644
--- a/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java
+++ b/src/main/java/io/reactivex/rxjava3/internal/schedulers/IoScheduler.java
@@ -50,7 +50,7 @@ public final class IoScheduler extends Scheduler {
 
     /** The name of the system property for setting the release behaviour for this Scheduler. */
     private static final String KEY_SCHEDULED_RELEASE = "rx3.io-scheduled-release";
-    private static final boolean USE_SCHEDULED_RELEASE;
+    static boolean USE_SCHEDULED_RELEASE;
 
     static final CachedWorkerPool NONE;
 
diff --git a/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java b/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java
index 90689654b9..6711be2645 100644
--- a/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java
+++ b/src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java
@@ -32,9 +32,8 @@
  * <ul>
  * <li>{@code rx3.io-keep-alive-time} (long): sets the keep-alive time of the {@link #io()} Scheduler workers, default is {@link IoScheduler#KEEP_ALIVE_TIME_DEFAULT}</li>
  * <li>{@code rx3.io-priority} (int): sets the thread priority of the {@link #io()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
- * <li>{@code rx3.io-scheduled-release} (boolean): Changes release behaviour mode for {@link #io()} Scheduler.
- *   Enabling this guarantees that workers are not released before they finish all processing, therefore preventing deadlocks.
- *   The drawback is that it can lead to more threads being created, since threads are released later. Default is false</li>
+ * <li>{@code rx3.io-scheduled-release} (boolean): {@code true} sets the worker release mode of the
+ * {@link #io()} Scheduler to <em>scheduled</em>, default is {@code false} for <em>eager</em> mode.</li>
  * <li>{@code rx3.computation-threads} (int): sets the number of threads in the {@link #computation()} Scheduler, default is the number of available CPUs</li>
  * <li>{@code rx3.computation-priority} (int): sets the thread priority of the {@link #computation()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
  * <li>{@code rx3.newthread-priority} (int): sets the thread priority of the {@link #newThread()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
@@ -162,9 +161,8 @@ public static Scheduler computation() {
      * <ul>
      * <li>{@code rx3.io-keep-alive-time} (long): sets the keep-alive time of the {@code io()} Scheduler workers, default is {@link IoScheduler#KEEP_ALIVE_TIME_DEFAULT}</li>
      * <li>{@code rx3.io-priority} (int): sets the thread priority of the {@code io()} Scheduler, default is {@link Thread#NORM_PRIORITY}</li>
-     * <li>{@code rx3.io-scheduled-release} (boolean): Changes release behaviour mode for {@code #io()} Scheduler.
-     *    Enabling this guarantees that workers are not released before they finish all processing, therefore preventing deadlocks.
-     *    The drawback is that it can lead to more threads being created, since threads are released later. Default is false</li>
+     * <li>{@code rx3.io-scheduled-release} (boolean): {@code true} sets the worker release mode of the
+     * {@code #io()} Scheduler to <em>scheduled</em>, default is {@code false} for <em>eager</em> mode.</li>
      * </ul>
      * <p>
      * The default value of this scheduler can be overridden at initialization time via the
@@ -181,6 +179,21 @@ public static Scheduler computation() {
      * <p>Operators on the base reactive classes that use this scheduler are marked with the
      * &#64;{@link io.reactivex.rxjava3.annotations.SchedulerSupport SchedulerSupport}({@link io.reactivex.rxjava3.annotations.SchedulerSupport#IO IO})
      * annotation.
+     * <p>
+     * When the {@link Scheduler.Worker} is disposed, the underlying worker can be released to the cached worker pool in two modes:
+     * <ul>
+     * <li>In <em>eager</em> mode (default), the underlying worker is returned immediately to the cached worker pool
+     *  and can be reused much quicker by operators. The drawback is that if the currently running task doesn't
+     * respond to interruption in time or at all, this may lead to delays or deadlock with the reuse use of the
+     * underlying worker.
+     * </li>
+     * <li>In <em>scheduled</em> mode (enabled via the system parameter {@code rx3.io-scheduled-release}
+     * set to {@code true}), the underlying worker is returned to the cached worker pool only after the currently running task
+     * has finished. This can help prevent premature reuse of the underlying worker and likely won't lead to delays or
+     * deadlock with such reuses. The drawback is that the delay in release may lead to an excess amount of underlying
+     * workers being created.
+     * </li>
+     * </ul>
      * @return a {@link Scheduler} meant for IO-bound work
      */
     @NonNull
diff --git a/src/test/java/io/reactivex/rxjava3/internal/schedulers/IoScheduledReleaseTest.java b/src/test/java/io/reactivex/rxjava3/internal/schedulers/IoScheduledReleaseTest.java
new file mode 100644
index 0000000000..b845c08ae2
--- /dev/null
+++ b/src/test/java/io/reactivex/rxjava3/internal/schedulers/IoScheduledReleaseTest.java
@@ -0,0 +1,51 @@
+/**
+ * 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.rxjava3.internal.schedulers;
+
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.RxJavaTest;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+public class IoScheduledReleaseTest extends RxJavaTest {
+
+    /* This test will be stuck in a deadlock if IoScheduler.USE_SCHEDULED_RELEASE is not set */
+    @Test
+    public void scheduledRelease() {
+        boolean savedScheduledRelease = IoScheduler.USE_SCHEDULED_RELEASE;
+        IoScheduler.USE_SCHEDULED_RELEASE = true;
+        try {
+            Flowable.just("item")
+                    .observeOn(Schedulers.io())
+                    .firstOrError()
+                    .map(item -> {
+                        for (int i = 0; i < 50; i++) {
+                            Completable.complete()
+                                    .observeOn(Schedulers.io())
+                                    .blockingAwait();
+                        }
+                        return "Done";
+                    })
+                    .ignoreElement()
+                    .test()
+                    .awaitDone(5, TimeUnit.SECONDS)
+                    .assertComplete();
+        } finally {
+            IoScheduler.USE_SCHEDULED_RELEASE = savedScheduledRelease;
+        }
+    }
+}
diff --git a/src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java b/src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java
deleted file mode 100644
index 9840cdb0eb..0000000000
--- a/src/test/java/io/reactivex/rxjava3/schedulers/IoScheduledReleaseTest.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/**
- * 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.rxjava3.schedulers;
-
-import io.reactivex.rxjava3.core.Completable;
-import io.reactivex.rxjava3.core.Flowable;
-import io.reactivex.rxjava3.core.RxJavaTest;
-import org.junit.Test;
-
-import java.util.concurrent.TimeUnit;
-
-public class IoScheduledReleaseTest extends RxJavaTest {
-
-    /* This test will be stuck in a deadlock if rx3.io-scheduled-release is not set */
-    @Test
-    public void scheduledRelease() {
-        System.setProperty("rx3.io-scheduled-release", Boolean.TRUE.toString());
-        Flowable.just("item")
-                .observeOn(Schedulers.io())
-                .firstOrError()
-                .map(item -> {
-                    for (int i = 0; i < 50; i++) {
-                        Completable.complete()
-                                .observeOn(Schedulers.io())
-                                .blockingAwait();
-                    }
-                    return "Done";
-                })
-                .ignoreElement()
-                .test()
-                .awaitDone(5, TimeUnit.SECONDS)
-                .assertComplete();
-    }
-}