Merge pull request ReactiveX#3149 from akarnokd/SchedulerShutdownV3

Scheduler shutdown capability

Author: akarnokd

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

@@ -16,13 +16,14 @@
 package rx.internal.schedulers;
 
 import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 
 import rx.*;
 import rx.functions.Action0;
 import rx.internal.util.*;
 import rx.subscriptions.*;
 
-public class EventLoopsScheduler extends Scheduler {
+public class EventLoopsScheduler extends Scheduler implements SchedulerLifecycle {
     /** Manages a fixed number of workers. */
     private static final String THREAD_NAME_PREFIX = "RxComputationThreadPool-";
     private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);
@@ -44,40 +45,82 @@ public class EventLoopsScheduler extends Scheduler {
         }
         MAX_THREADS = max;
     }
+    
+    static final PoolWorker SHUTDOWN_WORKER;
+    static {
+        SHUTDOWN_WORKER = new PoolWorker(new RxThreadFactory("RxComputationShutdown-"));
+        SHUTDOWN_WORKER.unsubscribe();
+    }
+    
     static final class FixedSchedulerPool {
         final int cores;
 
         final PoolWorker[] eventLoops;
         long n;
 
-        FixedSchedulerPool() {
+        FixedSchedulerPool(int maxThreads) {
             // initialize event loops
-            this.cores = MAX_THREADS;
-            this.eventLoops = new PoolWorker[cores];
-            for (int i = 0; i < cores; i++) {
+            this.cores = maxThreads;
+            this.eventLoops = new PoolWorker[maxThreads];
+            for (int i = 0; i < maxThreads; i++) {
                 this.eventLoops[i] = new PoolWorker(THREAD_FACTORY);
             }
         }
 
         public PoolWorker getEventLoop() {
+            int c = cores;
+            if (c == 0) {
+                return SHUTDOWN_WORKER;
+            }
             // simple round robin, improvements to come
-            return eventLoops[(int)(n++ % cores)];
+            return eventLoops[(int)(n++ % c)];
+        }
+        
+        public void shutdown() {
+            for (PoolWorker w : eventLoops) {
+                w.unsubscribe();
+            }
         }
     }
+    /** This will indicate no pool is active. */
+    static final FixedSchedulerPool NONE = new FixedSchedulerPool(0);
 
-    final FixedSchedulerPool pool;
+    final AtomicReference<FixedSchedulerPool> pool;
 
     /**
      * Create a scheduler with pool size equal to the available processor
      * count and using least-recent worker selection policy.
      */
     public EventLoopsScheduler() {
-        pool = new FixedSchedulerPool();
+        this.pool = new AtomicReference<FixedSchedulerPool>(NONE);
+        start();
     }
 
     @Override
     public Worker createWorker() {
-        return new EventLoopWorker(pool.getEventLoop());
+        return new EventLoopWorker(pool.get().getEventLoop());
+    }
+    
+    @Override
+    public void start() {
+        FixedSchedulerPool update = new FixedSchedulerPool(MAX_THREADS);
+        if (!pool.compareAndSet(NONE, update)) {
+            update.shutdown();
+        }
+    }
+    
+    @Override
+    public void shutdown() {
+        for (;;) {
+            FixedSchedulerPool curr = pool.get();
+            if (curr == NONE) {
+                return;
+            }
+            if (pool.compareAndSet(curr, NONE)) {
+                curr.shutdown();
+                return;
+            }
+        }
     }
 
     /**
@@ -87,7 +130,7 @@ public Worker createWorker() {
      * @return the subscription
      */
     public Subscription scheduleDirect(Action0 action) {
-       PoolWorker pw = pool.getEventLoop();
+       PoolWorker pw = pool.get().getEventLoop();
        return pw.scheduleActual(action, -1, TimeUnit.NANOSECONDS);
     }
 
@@ -137,4 +180,4 @@ private static final class PoolWorker extends NewThreadWorker {
             super(threadFactory);
         }
     }
-}
+}

src/main/java/rx/schedulers/GenericScheduledExecutorService.java renamed to src/main/java/rx/internal/schedulers/GenericScheduledExecutorService.java

@@ -13,13 +13,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package rx.schedulers;
+package rx.internal.schedulers;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Scheduler;
-import rx.internal.schedulers.NewThreadWorker;
 import rx.internal.util.RxThreadFactory;
-
-import java.util.concurrent.*;
+import rx.schedulers.*;
 
 /**
  * A default {@link ScheduledExecutorService} that can be used for scheduling actions when a {@link Scheduler} implementation doesn't have that ability.
@@ -30,15 +31,29 @@
  * the work asynchronously on the appropriate {@link Scheduler} implementation. This means for example that you would not use this approach
  * along with {@link TrampolineScheduler} or {@link ImmediateScheduler}.
  */
-/* package */final class GenericScheduledExecutorService {
+public final class GenericScheduledExecutorService implements SchedulerLifecycle{
 
     private static final String THREAD_NAME_PREFIX = "RxScheduledExecutorPool-";
     private static final RxThreadFactory THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX);
 
-    private final static GenericScheduledExecutorService INSTANCE = new GenericScheduledExecutorService();
-    private final ScheduledExecutorService executor;
-
+    /* Schedulers needs acces to this in order to work with the lifecycle. */
+    public final static GenericScheduledExecutorService INSTANCE = new GenericScheduledExecutorService();
+    
+    private final AtomicReference<ScheduledExecutorService> executor;
+    
+    static final ScheduledExecutorService NONE;
+    static {
+        NONE = Executors.newScheduledThreadPool(0);
+        NONE.shutdownNow();
+    }
+    
     private GenericScheduledExecutorService() {
+        executor = new AtomicReference<ScheduledExecutorService>(NONE);
+        start();
+    }
+
+    @Override
+    public void start() {
         int count = Runtime.getRuntime().availableProcessors();
         if (count > 4) {
             count = count / 2;
@@ -47,21 +62,41 @@ private GenericScheduledExecutorService() {
         if (count > 8) {
             count = 8;
         }
+        
         ScheduledExecutorService exec = Executors.newScheduledThreadPool(count, THREAD_FACTORY);
-        if (!NewThreadWorker.tryEnableCancelPolicy(exec)) {
-            if (exec instanceof ScheduledThreadPoolExecutor) {
-                NewThreadWorker.registerExecutor((ScheduledThreadPoolExecutor)exec);
+        if (executor.compareAndSet(NONE, exec)) {
+            if (!NewThreadWorker.tryEnableCancelPolicy(exec)) {
+                if (exec instanceof ScheduledThreadPoolExecutor) {
+                    NewThreadWorker.registerExecutor((ScheduledThreadPoolExecutor)exec);
+                }
             }
+            return;
+        } else {
+            exec.shutdownNow();
         }
-        executor = exec;
     }
-
+    
+    @Override
+    public void shutdown() {
+        for (;;) {
+            ScheduledExecutorService exec = executor.get();
+            if (exec == NONE) {
+                return;
+            }
+            if (executor.compareAndSet(exec, NONE)) {
+                NewThreadWorker.deregisterExecutor(exec);
+                exec.shutdownNow();
+                return;
+            }
+        }
+    }
+    
     /**
      * See class Javadoc for information on what this is for and how to use.
      * 
      * @return {@link ScheduledExecutorService} for generic use.
      */
     public static ScheduledExecutorService getInstance() {
-        return INSTANCE.executor;
+        return INSTANCE.executor.get();
     }
-}
+}

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

@@ -82,6 +82,8 @@ public void run() {
                 }, PURGE_FREQUENCY, PURGE_FREQUENCY, TimeUnit.MILLISECONDS);
 
                 break;
+            } else {
+                exec.shutdownNow();
             }
         } while (true);
 

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

@@ -0,0 +1,20 @@
+package rx.internal.schedulers;
+
+/**
+ * Represents the capability of a Scheduler to be start or shut down its maintained
+ * threads.
+ */
+public interface SchedulerLifecycle {
+    /** 
+     * Allows the Scheduler instance to start threads 
+     * and accept tasks on them.
+     * <p>Implementations should make sure the call is idempotent and threadsafe. 
+     */
+    void start();
+    /** 
+     * Instructs the Scheduler instance to stop threads 
+     * and stop accepting tasks on any outstanding Workers. 
+     * <p>Implementations should make sure the call is idempotent and threadsafe. 
+     */
+    void shutdown();
+}

src/main/java/rx/internal/util/ObjectPool.java

@@ -18,20 +18,22 @@
 package rx.internal.util;
 
 import java.util.Queue;
-import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
 
-import rx.Scheduler;
+import rx.Scheduler.Worker;
 import rx.functions.Action0;
-import rx.internal.util.unsafe.MpmcArrayQueue;
-import rx.internal.util.unsafe.UnsafeAccess;
+import rx.internal.schedulers.SchedulerLifecycle;
+import rx.internal.util.unsafe.*;
 import rx.schedulers.Schedulers;
 
-public abstract class ObjectPool<T> {
+public abstract class ObjectPool<T> implements SchedulerLifecycle {
     private Queue<T> pool;
+    private final int minSize;
     private final int maxSize;
+    private final long validationInterval;
 
-    private Scheduler.Worker schedulerWorker;
+    private final AtomicReference<Worker> schedulerWorker;
 
     public ObjectPool() {
         this(0, 0, 67);
@@ -50,31 +52,14 @@ public ObjectPool() {
      *            When the number of objects is greater than maxIdle, too many instances will be removed.
      */
     private ObjectPool(final int min, final int max, final long validationInterval) {
+        this.minSize = min;
         this.maxSize = max;
+        this.validationInterval = validationInterval;
+        this.schedulerWorker = new AtomicReference<Worker>();
         // initialize pool
         initialize(min);
 
-        schedulerWorker = Schedulers.computation().createWorker();
-        schedulerWorker.schedulePeriodically(new Action0() {
-
-            @Override
-            public void call() {
-                int size = pool.size();
-                if (size < min) {
-                    int sizeToBeAdded = max - size;
-                    for (int i = 0; i < sizeToBeAdded; i++) {
-                        pool.add(createObject());
-                    }
-                } else if (size > max) {
-                    int sizeToBeRemoved = size - max;
-                    for (int i = 0; i < sizeToBeRemoved; i++) {
-                        //                        pool.pollLast();
-                        pool.poll();
-                    }
-                }
-            }
-
-        }, validationInterval, validationInterval, TimeUnit.SECONDS);
+        start();
     }
 
     /**
@@ -109,10 +94,43 @@ public void returnObject(T object) {
     /**
      * Shutdown this pool.
      */
+    @Override
     public void shutdown() {
-        schedulerWorker.unsubscribe();
+        Worker w = schedulerWorker.getAndSet(null);
+        if (w != null) {
+            w.unsubscribe();
+        }
     }
 
+    @Override
+    public void start() {
+        Worker w = Schedulers.computation().createWorker();
+        if (schedulerWorker.compareAndSet(null, w)) {
+            w.schedulePeriodically(new Action0() {
+    
+                @Override
+                public void call() {
+                    int size = pool.size();
+                    if (size < minSize) {
+                        int sizeToBeAdded = maxSize - size;
+                        for (int i = 0; i < sizeToBeAdded; i++) {
+                            pool.add(createObject());
+                        }
+                    } else if (size > maxSize) {
+                        int sizeToBeRemoved = size - maxSize;
+                        for (int i = 0; i < sizeToBeRemoved; i++) {
+                            //                        pool.pollLast();
+                            pool.poll();
+                        }
+                    }
+                }
+    
+            }, validationInterval, validationInterval, TimeUnit.SECONDS);
+        } else {
+            w.unsubscribe();
+        }
+    }
+    
     /**
      * Creates a new object.
      *

src/main/java/rx/internal/util/RxRingBuffer.java

@@ -276,7 +276,8 @@ public static RxRingBuffer getSpmcInstance() {
     }
     public static final int SIZE = _size;
 
-    private static ObjectPool<Queue<Object>> SPSC_POOL = new ObjectPool<Queue<Object>>() {
+    /* Public so Schedulers can manage the lifecycle of the inner worker. */
+    public static ObjectPool<Queue<Object>> SPSC_POOL = new ObjectPool<Queue<Object>>() {
 
         @Override
         protected SpscArrayQueue<Object> createObject() {
@@ -285,7 +286,8 @@ protected SpscArrayQueue<Object> createObject() {
 
     };
 
-    private static ObjectPool<Queue<Object>> SPMC_POOL = new ObjectPool<Queue<Object>>() {
+    /* Public so Schedulers can manage the lifecycle of the inner worker. */
+    public static ObjectPool<Queue<Object>> SPMC_POOL = new ObjectPool<Queue<Object>>() {
 
         @Override
         protected SpmcArrayQueue<Object> createObject() {
@@ -452,4 +454,4 @@ public boolean isUnsubscribed() {
         return queue == null;
     }
 
-}
+}