CachedThreadScheduler should wait until the previous action (if any) … (ReactiveX#4231)

* CachedThreadScheduler should wait until the previous action (if any) completes before releasing a worker to the pool. Fixes ReactiveX#4230.

* Modified EventLoopWorker to implement Action0 and schedule itself to release. This saves an allocation.

Author: csabakos

src/main/java/rx/internal/schedulers/CachedThreadScheduler.java

@@ -174,7 +174,7 @@ public Worker createWorker() {
         return new EventLoopWorker(pool.get());
     }
 
-    static final class EventLoopWorker extends Scheduler.Worker {
+    static final class EventLoopWorker extends Scheduler.Worker implements Action0 {
         private final CompositeSubscription innerSubscription = new CompositeSubscription();
         private final CachedWorkerPool pool;
         private final ThreadWorker threadWorker;
@@ -190,11 +190,18 @@ static final class EventLoopWorker extends Scheduler.Worker {
         public void unsubscribe() {
             if (once.compareAndSet(false, true)) {
                 // unsubscribe should be idempotent, so only do this once
-                pool.release(threadWorker);
+
+                // Release the worker _after_ the previous action (if any) has completed
+                threadWorker.schedule(this);
             }
             innerSubscription.unsubscribe();
         }
 
+        @Override
+        public void call() {
+            pool.release(threadWorker);
+        }
+
         @Override
         public boolean isUnsubscribed() {
             return innerSubscription.isUnsubscribed();

src/test/java/rx/schedulers/IoSchedulerTest.java

@@ -18,6 +18,8 @@
 
 import static org.junit.Assert.assertTrue;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
 import org.junit.Test;
 
 import rx.*;
@@ -83,4 +85,71 @@ public void testCancelledTaskRetention() throws InterruptedException {
         }
     }
 
+    // Tests that an uninterruptible worker does not get reused
+    @Test(timeout = 10000)
+    public void testUninterruptibleActionDoesNotBlockOtherAction() throws InterruptedException {
+        final Worker uninterruptibleWorker = Schedulers.io().createWorker();
+        final AtomicBoolean running = new AtomicBoolean(false);
+        final AtomicBoolean shouldQuit = new AtomicBoolean(false);
+        try {
+            uninterruptibleWorker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    synchronized (running) {
+                        running.set(true);
+                        running.notifyAll();
+                    }
+                    synchronized (shouldQuit) {
+                        while (!shouldQuit.get()) {
+                            try {
+                                shouldQuit.wait();
+                            } catch (final InterruptedException ignored) {
+                            }
+                        }
+                    }
+                    synchronized (running) {
+                        running.set(false);
+                        running.notifyAll();
+                    }
+                }
+            });
+
+            // Wait for the action to start executing
+            synchronized (running) {
+                while (!running.get()) {
+                    running.wait();
+                }
+            }
+        } finally {
+            uninterruptibleWorker.unsubscribe();
+        }
+
+        final Worker otherWorker = Schedulers.io().createWorker();
+        final AtomicBoolean otherActionRan = new AtomicBoolean(false);
+        try {
+            otherWorker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    otherActionRan.set(true);
+                }
+            });
+            Thread.sleep(1000); // give the action a chance to run
+        } finally {
+            otherWorker.unsubscribe();
+        }
+
+        assertTrue(running.get()); // uninterruptible action keeps on running since InterruptedException is swallowed
+        assertTrue(otherActionRan.get());
+
+        // Wait for uninterruptibleWorker to exit (to clean up after ourselves)
+        synchronized (shouldQuit) {
+            shouldQuit.set(true);
+            shouldQuit.notifyAll();
+        }
+        synchronized (running) {
+            while (running.get()) {
+                running.wait();
+            }
+        }
+    }
 }