Re-submission of CachedThreadScheduler, implementing #1140

rxjava-core/src/main/java/rx/schedulers/CachedThreadScheduler.java

@@ -0,0 +1,169 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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 rx.schedulers;
+
+import rx.Scheduler;
+import rx.Subscription;
+import rx.functions.Action0;
+import rx.subscriptions.CompositeSubscription;
+import rx.subscriptions.Subscriptions;
+
+import java.util.Iterator;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+/* package */class CachedThreadScheduler extends Scheduler {
+    private static final String WORKER_THREAD_NAME_PREFIX = "RxCachedThreadScheduler-";
+    private static final NewThreadScheduler.RxThreadFactory WORKER_THREAD_FACTORY =
+            new NewThreadScheduler.RxThreadFactory(WORKER_THREAD_NAME_PREFIX);
+
+    private static final String EVICTOR_THREAD_NAME_PREFIX = "RxCachedWorkerPoolEvictor-";
+    private static final NewThreadScheduler.RxThreadFactory EVICTOR_THREAD_FACTORY =
+            new NewThreadScheduler.RxThreadFactory(EVICTOR_THREAD_NAME_PREFIX);
+
+    private static final class CachedWorkerPool {
+        private final long keepAliveTime;
+        private final ConcurrentLinkedQueue<ThreadWorker> expiringWorkerQueue;
+        private final ScheduledExecutorService evictExpiredWorkerExecutor;
+
+        CachedWorkerPool(long keepAliveTime, TimeUnit unit) {
+            this.keepAliveTime = unit.toNanos(keepAliveTime);
+            this.expiringWorkerQueue = new ConcurrentLinkedQueue<ThreadWorker>();
+
+            evictExpiredWorkerExecutor = Executors.newScheduledThreadPool(1, EVICTOR_THREAD_FACTORY);
+            evictExpiredWorkerExecutor.scheduleWithFixedDelay(
+                    new Runnable() {
+                        @Override
+                        public void run() {
+                            evictExpiredWorkers();
+                        }
+                    }, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS
+            );
+        }
+
+        private static CachedWorkerPool INSTANCE = new CachedWorkerPool(
+                60L, TimeUnit.SECONDS
+        );
+
+        ThreadWorker get() {
+            while (!expiringWorkerQueue.isEmpty()) {
+                ThreadWorker threadWorker = expiringWorkerQueue.poll();
+                if (threadWorker != null) {
+                    return threadWorker;
+                }
+            }
+
+            // No cached worker found, so create a new one.
+            return new ThreadWorker(WORKER_THREAD_FACTORY);
+        }
+
+        void release(ThreadWorker threadWorker) {
+            // Refresh expire time before putting worker back in pool
+            threadWorker.setExpirationTime(now() + keepAliveTime);
+
+            expiringWorkerQueue.offer(threadWorker);
+        }
+
+        void evictExpiredWorkers() {
+            if (!expiringWorkerQueue.isEmpty()) {
+                long currentTimestamp = now();
+
+                Iterator<ThreadWorker> threadWorkerIterator = expiringWorkerQueue.iterator();
+                while (threadWorkerIterator.hasNext()) {
+                    ThreadWorker threadWorker = threadWorkerIterator.next();
+                    if (threadWorker.getExpirationTime() <= currentTimestamp) {
+                        threadWorkerIterator.remove();
+                        threadWorker.unsubscribe();
+                    } else {
+                        // Queue is ordered with the worker that will expire first in the beginning, so when we
+                        // find a non-expired worker we can stop evicting.
+                        break;
+                    }
+                }
+            }
+        }
+
+        long now() {
+            return System.nanoTime();
+        }
+    }
+
+    @Override
+    public Worker createWorker() {
+        return new EventLoopWorker(CachedWorkerPool.INSTANCE.get());
+    }
+
+    private static class EventLoopWorker extends Scheduler.Worker {
+        private final CompositeSubscription innerSubscription = new CompositeSubscription();
+        private final ThreadWorker threadWorker;
+        volatile int once;
+        static final AtomicIntegerFieldUpdater<EventLoopWorker> ONCE_UPDATER
+                = AtomicIntegerFieldUpdater.newUpdater(EventLoopWorker.class, "once");
+
+        EventLoopWorker(ThreadWorker threadWorker) {
+            this.threadWorker = threadWorker;
+        }
+
+        @Override
+        public void unsubscribe() {
+            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+                // unsubscribe should be idempotent, so only do this once
+                CachedWorkerPool.INSTANCE.release(threadWorker);
+            }
+            innerSubscription.unsubscribe();
+        }
+
+        @Override
+        public boolean isUnsubscribed() {
+            return innerSubscription.isUnsubscribed();
+        }
+
+        @Override
+        public Subscription schedule(Action0 action) {
+            return schedule(action, 0, null);
+        }
+
+        @Override
+        public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) {
+            if (innerSubscription.isUnsubscribed()) {
+                // don't schedule, we are unsubscribed
+                return Subscriptions.empty();
+            }
+
+            NewThreadScheduler.NewThreadWorker.ScheduledAction s = threadWorker.scheduleActual(action, delayTime, unit);
+            innerSubscription.add(s);
+            s.addParent(innerSubscription);
+            return s;
+        }
+    }
+
+    private static final class ThreadWorker extends NewThreadScheduler.NewThreadWorker {
+        private long expirationTime;
+
+        ThreadWorker(ThreadFactory threadFactory) {
+            super(threadFactory);
+            this.expirationTime = 0L;
+        }
+
+        public long getExpirationTime() {
+            return expirationTime;
+        }
+
+        public void setExpirationTime(long expirationTime) {
+            this.expirationTime = expirationTime;
+        }
+    }
+}

rxjava-core/src/main/java/rx/schedulers/Schedulers.java

@@ -15,11 +15,11 @@
  */
 package rx.schedulers;
 
-import java.util.concurrent.Executor;
-
 import rx.Scheduler;
 import rx.plugins.RxJavaPlugins;
 
+import java.util.concurrent.Executor;
+
 /**
  * Static factory methods for creating Schedulers.
  */
@@ -43,7 +43,7 @@ private Schedulers() {
         if (io != null) {
             ioScheduler = io;
         } else {
-            ioScheduler = NewThreadScheduler.instance(); // defaults to new thread
+            ioScheduler = new CachedThreadScheduler();
         }
 
         Scheduler nt = RxJavaPlugins.getInstance().getDefaultSchedulers().getNewThreadScheduler();

rxjava-core/src/test/java/rx/schedulers/CachedThreadSchedulerTest.java

@@ -0,0 +1,60 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ *
+ * 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 rx.schedulers;
+
+import org.junit.Test;
+import rx.Observable;
+import rx.Scheduler;
+import rx.functions.Action1;
+import rx.functions.Func1;
+
+import static org.junit.Assert.assertTrue;
+
+public class CachedThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
+
+    @Override
+    protected Scheduler getScheduler() {
+        return Schedulers.io();
+    }
+
+    /**
+     * IO scheduler defaults to using CachedThreadScheduler
+     */
+    @Test
+    public final void testIOScheduler() {
+
+        Observable<Integer> o1 = Observable.from(1, 2, 3, 4, 5);
+        Observable<Integer> o2 = Observable.from(6, 7, 8, 9, 10);
+        Observable<String> o = Observable.merge(o1, o2).map(new Func1<Integer, String>() {
+
+            @Override
+            public String call(Integer t) {
+                assertTrue(Thread.currentThread().getName().startsWith("RxCachedThreadScheduler"));
+                return "Value_" + t + "_Thread_" + Thread.currentThread().getName();
+            }
+        });
+
+        o.subscribeOn(Schedulers.io()).toBlocking().forEach(new Action1<String>() {
+
+            @Override
+            public void call(String t) {
+                System.out.println("t: " + t);
+            }
+        });
+    }
+
+}

rxjava-core/src/test/java/rx/schedulers/NewThreadSchedulerTest.java

@@ -16,46 +16,12 @@
 
 package rx.schedulers;
 
-import static org.junit.Assert.assertTrue;
-
-import org.junit.Test;
-
-import rx.Observable;
 import rx.Scheduler;
-import rx.functions.Action1;
-import rx.functions.Func1;
 
 public class NewThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
 
     @Override
     protected Scheduler getScheduler() {
         return Schedulers.newThread();
     }
-
-    /**
-     * IO scheduler defaults to using NewThreadScheduler
-     */
-    @Test
-    public final void testIOScheduler() {
-
-        Observable<Integer> o1 = Observable.<Integer> from(1, 2, 3, 4, 5);
-        Observable<Integer> o2 = Observable.<Integer> from(6, 7, 8, 9, 10);
-        Observable<String> o = Observable.<Integer> merge(o1, o2).map(new Func1<Integer, String>() {
-
-            @Override
-            public String call(Integer t) {
-                assertTrue(Thread.currentThread().getName().startsWith("RxNewThreadScheduler"));
-                return "Value_" + t + "_Thread_" + Thread.currentThread().getName();
-            }
-        });
-
-        o.subscribeOn(Schedulers.io()).toBlocking().forEach(new Action1<String>() {
-
-            @Override
-            public void call(String t) {
-                System.out.println("t: " + t);
-            }
-        });
-    }
-
 }