diff --git a/src/main/java/rx/internal/operators/OperatorObserveOn.java b/src/main/java/rx/internal/operators/OperatorObserveOn.java index 1720e1dfe2..f09a424020 100644 --- a/src/main/java/rx/internal/operators/OperatorObserveOn.java +++ b/src/main/java/rx/internal/operators/OperatorObserveOn.java @@ -22,11 +22,11 @@ import rx.Observable.Operator; import rx.exceptions.MissingBackpressureException; import rx.functions.Action0; +import rx.internal.schedulers.*; import rx.internal.util.*; import rx.internal.util.atomic.SpscAtomicArrayQueue; import rx.internal.util.unsafe.*; import rx.plugins.RxJavaPlugins; -import rx.schedulers.*; /** * Delivers events on the specified {@code Scheduler} asynchronously via an unbounded buffer. diff --git a/src/main/java/rx/schedulers/ExecutorScheduler.java b/src/main/java/rx/internal/schedulers/ExecutorScheduler.java similarity index 98% rename from src/main/java/rx/schedulers/ExecutorScheduler.java rename to src/main/java/rx/internal/schedulers/ExecutorScheduler.java index 11d10214f9..b40c0b4900 100644 --- a/src/main/java/rx/schedulers/ExecutorScheduler.java +++ b/src/main/java/rx/internal/schedulers/ExecutorScheduler.java @@ -13,14 +13,13 @@ * 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.AtomicInteger; import rx.*; import rx.functions.Action0; -import rx.internal.schedulers.*; import rx.plugins.RxJavaPlugins; import rx.subscriptions.*; @@ -30,7 +29,7 @@ * Note that thread-hopping is unavoidable with this kind of Scheduler as we don't know about the underlying * threading behavior of the executor. */ -/* public */final class ExecutorScheduler extends Scheduler { +public final class ExecutorScheduler extends Scheduler { final Executor executor; public ExecutorScheduler(Executor executor) { this.executor = executor; diff --git a/src/main/java/rx/internal/schedulers/GenericScheduledExecutorService.java b/src/main/java/rx/internal/schedulers/GenericScheduledExecutorService.java index e322945068..3bc60f076b 100644 --- a/src/main/java/rx/internal/schedulers/GenericScheduledExecutorService.java +++ b/src/main/java/rx/internal/schedulers/GenericScheduledExecutorService.java @@ -20,7 +20,6 @@ import rx.Scheduler; import rx.internal.util.RxThreadFactory; -import rx.schedulers.*; /** * A default {@link ScheduledExecutorService} that can be used for scheduling actions when a {@link Scheduler} implementation doesn't have that ability. diff --git a/src/main/java/rx/internal/schedulers/ImmediateScheduler.java b/src/main/java/rx/internal/schedulers/ImmediateScheduler.java new file mode 100644 index 0000000000..c3dd7a6f85 --- /dev/null +++ b/src/main/java/rx/internal/schedulers/ImmediateScheduler.java @@ -0,0 +1,73 @@ +/** + * 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.internal.schedulers; + +import java.util.concurrent.TimeUnit; + +import rx.Scheduler; +import rx.Subscription; +import rx.functions.Action0; +import rx.subscriptions.BooleanSubscription; +import rx.subscriptions.Subscriptions; + +/** + * Executes work immediately on the current thread. + */ +public final class ImmediateScheduler extends Scheduler { + public static final ImmediateScheduler INSTANCE = new ImmediateScheduler(); + + private ImmediateScheduler() { + } + + @Override + public Worker createWorker() { + return new InnerImmediateScheduler(); + } + + private class InnerImmediateScheduler extends Scheduler.Worker implements Subscription { + + final BooleanSubscription innerSubscription = new BooleanSubscription(); + + InnerImmediateScheduler() { + } + + @Override + public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { + // since we are executing immediately on this thread we must cause this thread to sleep + long execTime = ImmediateScheduler.this.now() + unit.toMillis(delayTime); + + return schedule(new SleepingAction(action, this, execTime)); + } + + @Override + public Subscription schedule(Action0 action) { + action.call(); + return Subscriptions.unsubscribed(); + } + + @Override + public void unsubscribe() { + innerSubscription.unsubscribe(); + } + + @Override + public boolean isUnsubscribed() { + return innerSubscription.isUnsubscribed(); + } + + } + +} diff --git a/src/main/java/rx/schedulers/SleepingAction.java b/src/main/java/rx/internal/schedulers/SleepingAction.java similarity index 98% rename from src/main/java/rx/schedulers/SleepingAction.java rename to src/main/java/rx/internal/schedulers/SleepingAction.java index bb13734475..c41c01caf0 100644 --- a/src/main/java/rx/schedulers/SleepingAction.java +++ b/src/main/java/rx/internal/schedulers/SleepingAction.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package rx.schedulers; +package rx.internal.schedulers; import rx.Scheduler; import rx.functions.Action0; diff --git a/src/main/java/rx/internal/schedulers/TrampolineScheduler.java b/src/main/java/rx/internal/schedulers/TrampolineScheduler.java new file mode 100644 index 0000000000..94fea1964a --- /dev/null +++ b/src/main/java/rx/internal/schedulers/TrampolineScheduler.java @@ -0,0 +1,131 @@ +/** + * 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.internal.schedulers; + +import java.util.concurrent.PriorityBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import rx.Scheduler; +import rx.Subscription; +import rx.functions.Action0; +import rx.subscriptions.BooleanSubscription; +import rx.subscriptions.Subscriptions; + +/** + * Schedules work on the current thread but does not execute immediately. Work is put in a queue and executed + * after the current unit of work is completed. + */ +public final class TrampolineScheduler extends Scheduler { + public static final TrampolineScheduler INSTANCE = new TrampolineScheduler(); + + @Override + public Worker createWorker() { + return new InnerCurrentThreadScheduler(); + } + + private TrampolineScheduler() { + } + + private static class InnerCurrentThreadScheduler extends Scheduler.Worker implements Subscription { + + final AtomicInteger counter = new AtomicInteger(); + final PriorityBlockingQueue queue = new PriorityBlockingQueue(); + private final BooleanSubscription innerSubscription = new BooleanSubscription(); + private final AtomicInteger wip = new AtomicInteger(); + + InnerCurrentThreadScheduler() { + } + + @Override + public Subscription schedule(Action0 action) { + return enqueue(action, now()); + } + + @Override + public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { + long execTime = now() + unit.toMillis(delayTime); + + return enqueue(new SleepingAction(action, this, execTime), execTime); + } + + private Subscription enqueue(Action0 action, long execTime) { + if (innerSubscription.isUnsubscribed()) { + return Subscriptions.unsubscribed(); + } + final TimedAction timedAction = new TimedAction(action, execTime, counter.incrementAndGet()); + queue.add(timedAction); + + if (wip.getAndIncrement() == 0) { + do { + final TimedAction polled = queue.poll(); + if (polled != null) { + polled.action.call(); + } + } while (wip.decrementAndGet() > 0); + return Subscriptions.unsubscribed(); + } else { + // queue wasn't empty, a parent is already processing so we just add to the end of the queue + return Subscriptions.create(new Action0() { + + @Override + public void call() { + queue.remove(timedAction); + } + + }); + } + } + + @Override + public void unsubscribe() { + innerSubscription.unsubscribe(); + } + + @Override + public boolean isUnsubscribed() { + return innerSubscription.isUnsubscribed(); + } + + } + + private static final class TimedAction implements Comparable { + final Action0 action; + final Long execTime; + final int count; // In case if time between enqueueing took less than 1ms + + TimedAction(Action0 action, Long execTime, int count) { + this.action = action; + this.execTime = execTime; + this.count = count; + } + + @Override + public int compareTo(TimedAction that) { + int result = execTime.compareTo(that.execTime); + if (result == 0) { + return compare(count, that.count); + } + return result; + } + } + + // because I can't use Integer.compare from Java 7 + static int compare(int x, int y) { + return (x < y) ? -1 : ((x == y) ? 0 : 1); + } + +} diff --git a/src/main/java/rx/schedulers/ImmediateScheduler.java b/src/main/java/rx/schedulers/ImmediateScheduler.java index e480754a58..3be8edda4a 100644 --- a/src/main/java/rx/schedulers/ImmediateScheduler.java +++ b/src/main/java/rx/schedulers/ImmediateScheduler.java @@ -15,63 +15,20 @@ */ package rx.schedulers; -import java.util.concurrent.TimeUnit; - import rx.Scheduler; -import rx.Subscription; -import rx.functions.Action0; -import rx.subscriptions.BooleanSubscription; -import rx.subscriptions.Subscriptions; /** - * Executes work immediately on the current thread. + * @deprecated This type was never publicly instantiable. Use {@link Schedulers#immediate()}. */ +@Deprecated +@SuppressWarnings("unused") // Class was part of public API. public final class ImmediateScheduler extends Scheduler { - private static final ImmediateScheduler INSTANCE = new ImmediateScheduler(); - - /* package */static ImmediateScheduler instance() { - return INSTANCE; - } - - /* package accessible for unit tests */ImmediateScheduler() { + private ImmediateScheduler() { + throw new AssertionError(); } @Override public Worker createWorker() { - return new InnerImmediateScheduler(); + return null; } - - private class InnerImmediateScheduler extends Scheduler.Worker implements Subscription { - - final BooleanSubscription innerSubscription = new BooleanSubscription(); - - InnerImmediateScheduler() { - } - - @Override - public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { - // since we are executing immediately on this thread we must cause this thread to sleep - long execTime = ImmediateScheduler.this.now() + unit.toMillis(delayTime); - - return schedule(new SleepingAction(action, this, execTime)); - } - - @Override - public Subscription schedule(Action0 action) { - action.call(); - return Subscriptions.unsubscribed(); - } - - @Override - public void unsubscribe() { - innerSubscription.unsubscribe(); - } - - @Override - public boolean isUnsubscribed() { - return innerSubscription.isUnsubscribed(); - } - - } - } diff --git a/src/main/java/rx/schedulers/NewThreadScheduler.java b/src/main/java/rx/schedulers/NewThreadScheduler.java index 5dc6046268..192f8b29a8 100644 --- a/src/main/java/rx/schedulers/NewThreadScheduler.java +++ b/src/main/java/rx/schedulers/NewThreadScheduler.java @@ -21,6 +21,7 @@ * @deprecated This type was never publicly instantiable. Use {@link Schedulers#newThread()}. */ @Deprecated +@SuppressWarnings("unused") // Class was part of public API. public final class NewThreadScheduler extends Scheduler { private NewThreadScheduler() { throw new AssertionError(); diff --git a/src/main/java/rx/schedulers/Schedulers.java b/src/main/java/rx/schedulers/Schedulers.java index f81235347c..eae594ef08 100644 --- a/src/main/java/rx/schedulers/Schedulers.java +++ b/src/main/java/rx/schedulers/Schedulers.java @@ -61,21 +61,21 @@ private Schedulers() { /** * Creates and returns a {@link Scheduler} that executes work immediately on the current thread. - * - * @return an {@link ImmediateScheduler} instance + * + * @return a {@link Scheduler} that executes work immediately */ public static Scheduler immediate() { - return ImmediateScheduler.instance(); + return rx.internal.schedulers.ImmediateScheduler.INSTANCE; } /** * Creates and returns a {@link Scheduler} that queues work on the current thread to be executed after the * current work completes. - * - * @return a {@link TrampolineScheduler} instance + * + * @return a {@link Scheduler} that queues work on the current thread */ public static Scheduler trampoline() { - return TrampolineScheduler.instance(); + return rx.internal.schedulers.TrampolineScheduler.INSTANCE; } /** @@ -83,7 +83,7 @@ public static Scheduler trampoline() { *

* Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}. * - * @return a {@link NewThreadScheduler} instance + * @return a {@link Scheduler} that creates new threads */ public static Scheduler newThread() { return INSTANCE.newThreadScheduler; diff --git a/src/main/java/rx/schedulers/TrampolineScheduler.java b/src/main/java/rx/schedulers/TrampolineScheduler.java index 45bb18546c..5f708cdc23 100644 --- a/src/main/java/rx/schedulers/TrampolineScheduler.java +++ b/src/main/java/rx/schedulers/TrampolineScheduler.java @@ -15,121 +15,20 @@ */ package rx.schedulers; -import java.util.concurrent.PriorityBlockingQueue; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - import rx.Scheduler; -import rx.Subscription; -import rx.functions.Action0; -import rx.subscriptions.BooleanSubscription; -import rx.subscriptions.Subscriptions; /** - * Schedules work on the current thread but does not execute immediately. Work is put in a queue and executed - * after the current unit of work is completed. + * @deprecated This type was never publicly instantiable. Use {@link Schedulers#trampoline()}. */ +@Deprecated +@SuppressWarnings("unused") // Class was part of public API. public final class TrampolineScheduler extends Scheduler { - private static final TrampolineScheduler INSTANCE = new TrampolineScheduler(); - - /* package */static TrampolineScheduler instance() { - return INSTANCE; + private TrampolineScheduler() { + throw new AssertionError(); } @Override public Worker createWorker() { - return new InnerCurrentThreadScheduler(); - } - - /* package accessible for unit tests */TrampolineScheduler() { + return null; } - - private static class InnerCurrentThreadScheduler extends Scheduler.Worker implements Subscription { - - final AtomicInteger counter = new AtomicInteger(); - final PriorityBlockingQueue queue = new PriorityBlockingQueue(); - private final BooleanSubscription innerSubscription = new BooleanSubscription(); - private final AtomicInteger wip = new AtomicInteger(); - - InnerCurrentThreadScheduler() { - } - - @Override - public Subscription schedule(Action0 action) { - return enqueue(action, now()); - } - - @Override - public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { - long execTime = now() + unit.toMillis(delayTime); - - return enqueue(new SleepingAction(action, this, execTime), execTime); - } - - private Subscription enqueue(Action0 action, long execTime) { - if (innerSubscription.isUnsubscribed()) { - return Subscriptions.unsubscribed(); - } - final TimedAction timedAction = new TimedAction(action, execTime, counter.incrementAndGet()); - queue.add(timedAction); - - if (wip.getAndIncrement() == 0) { - do { - final TimedAction polled = queue.poll(); - if (polled != null) { - polled.action.call(); - } - } while (wip.decrementAndGet() > 0); - return Subscriptions.unsubscribed(); - } else { - // queue wasn't empty, a parent is already processing so we just add to the end of the queue - return Subscriptions.create(new Action0() { - - @Override - public void call() { - queue.remove(timedAction); - } - - }); - } - } - - @Override - public void unsubscribe() { - innerSubscription.unsubscribe(); - } - - @Override - public boolean isUnsubscribed() { - return innerSubscription.isUnsubscribed(); - } - - } - - private static final class TimedAction implements Comparable { - final Action0 action; - final Long execTime; - final int count; // In case if time between enqueueing took less than 1ms - - TimedAction(Action0 action, Long execTime, int count) { - this.action = action; - this.execTime = execTime; - this.count = count; - } - - @Override - public int compareTo(TimedAction that) { - int result = execTime.compareTo(that.execTime); - if (result == 0) { - return compare(count, that.count); - } - return result; - } - } - - // because I can't use Integer.compare from Java 7 - static int compare(int x, int y) { - return (x < y) ? -1 : ((x == y) ? 0 : 1); - } - } diff --git a/src/test/java/rx/schedulers/ExecutorSchedulerTest.java b/src/test/java/rx/internal/schedulers/ExecutorSchedulerTest.java similarity index 70% rename from src/test/java/rx/schedulers/ExecutorSchedulerTest.java rename to src/test/java/rx/internal/schedulers/ExecutorSchedulerTest.java index ed4e03213d..fad2435ce1 100644 --- a/src/test/java/rx/schedulers/ExecutorSchedulerTest.java +++ b/src/test/java/rx/internal/schedulers/ExecutorSchedulerTest.java @@ -13,22 +13,21 @@ * License for the specific language governing permissions and limitations under * the License. */ -package rx.schedulers; +package rx.internal.schedulers; import static org.junit.Assert.*; -import java.lang.management.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; - import org.junit.Test; - import rx.*; import rx.Scheduler.Worker; import rx.functions.*; -import rx.internal.schedulers.NewThreadWorker; +import rx.internal.schedulers.ExecutorScheduler.ExecutorSchedulerWorker; import rx.internal.util.RxThreadFactory; -import rx.schedulers.ExecutorScheduler.ExecutorSchedulerWorker; +import rx.schedulers.AbstractSchedulerConcurrencyTests; +import rx.schedulers.SchedulerTests; +import rx.schedulers.Schedulers; public class ExecutorSchedulerTest extends AbstractSchedulerConcurrencyTests { @@ -49,72 +48,6 @@ public final void testHandledErrorIsNotDeliveredToThreadHandler() throws Interru SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(getScheduler()); } - public static void testCancelledRetention(Scheduler.Worker w, boolean periodic) throws InterruptedException { - System.out.println("Wait before GC"); - Thread.sleep(1000); - - System.out.println("GC"); - System.gc(); - - Thread.sleep(1000); - - - MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean(); - MemoryUsage memHeap = memoryMXBean.getHeapMemoryUsage(); - long initial = memHeap.getUsed(); - - System.out.printf("Starting: %.3f MB%n", initial / 1024.0 / 1024.0); - - int n = 500 * 1000; - if (periodic) { - final CountDownLatch cdl = new CountDownLatch(n); - final Action0 action = new Action0() { - @Override - public void call() { - cdl.countDown(); - } - }; - for (int i = 0; i < n; i++) { - if (i % 50000 == 0) { - System.out.println(" -> still scheduling: " + i); - } - w.schedulePeriodically(action, 0, 1, TimeUnit.DAYS); - } - - System.out.println("Waiting for the first round to finish..."); - cdl.await(); - } else { - for (int i = 0; i < n; i++) { - if (i % 50000 == 0) { - System.out.println(" -> still scheduling: " + i); - } - w.schedule(Actions.empty(), 1, TimeUnit.DAYS); - } - } - - memHeap = memoryMXBean.getHeapMemoryUsage(); - long after = memHeap.getUsed(); - System.out.printf("Peak: %.3f MB%n", after / 1024.0 / 1024.0); - - w.unsubscribe(); - - System.out.println("Wait before second GC"); - Thread.sleep(NewThreadWorker.PURGE_FREQUENCY + 2000); - - System.out.println("Second GC"); - System.gc(); - - Thread.sleep(1000); - - memHeap = memoryMXBean.getHeapMemoryUsage(); - long finish = memHeap.getUsed(); - System.out.printf("After: %.3f MB%n", finish / 1024.0 / 1024.0); - - if (finish > initial * 5) { - fail(String.format("Tasks retained: %.3f -> %.3f -> %.3f", initial / 1024 / 1024.0, after / 1024 / 1024.0, finish / 1024 / 1024d)); - } - } - @Test(timeout = 30000) public void testCancelledTaskRetention() throws InterruptedException { ExecutorService exec = Executors.newSingleThreadExecutor(); @@ -122,14 +55,14 @@ public void testCancelledTaskRetention() throws InterruptedException { try { Scheduler.Worker w = s.createWorker(); try { - testCancelledRetention(w, false); + SchedulerTests.testCancelledRetention(w, false); } finally { w.unsubscribe(); } w = s.createWorker(); try { - testCancelledRetention(w, true); + SchedulerTests.testCancelledRetention(w, true); } finally { w.unsubscribe(); } diff --git a/src/test/java/rx/schedulers/ComputationSchedulerTests.java b/src/test/java/rx/schedulers/ComputationSchedulerTests.java index 7191f60015..7e9163e7c0 100644 --- a/src/test/java/rx/schedulers/ComputationSchedulerTests.java +++ b/src/test/java/rx/schedulers/ComputationSchedulerTests.java @@ -157,13 +157,13 @@ public final void testHandledErrorIsNotDeliveredToThreadHandler() throws Interru public void testCancelledTaskRetention() throws InterruptedException { Worker w = Schedulers.computation().createWorker(); try { - ExecutorSchedulerTest.testCancelledRetention(w, false); + SchedulerTests.testCancelledRetention(w, false); } finally { w.unsubscribe(); } w = Schedulers.computation().createWorker(); try { - ExecutorSchedulerTest.testCancelledRetention(w, true); + SchedulerTests.testCancelledRetention(w, true); } finally { w.unsubscribe(); } diff --git a/src/test/java/rx/schedulers/CachedThreadSchedulerTest.java b/src/test/java/rx/schedulers/IoSchedulerTest.java similarity index 91% rename from src/test/java/rx/schedulers/CachedThreadSchedulerTest.java rename to src/test/java/rx/schedulers/IoSchedulerTest.java index 9abb52b7ec..a16a19a61c 100644 --- a/src/test/java/rx/schedulers/CachedThreadSchedulerTest.java +++ b/src/test/java/rx/schedulers/IoSchedulerTest.java @@ -24,7 +24,7 @@ import rx.Scheduler.Worker; import rx.functions.*; -public class CachedThreadSchedulerTest extends AbstractSchedulerConcurrencyTests { +public class IoSchedulerTest extends AbstractSchedulerConcurrencyTests { @Override protected Scheduler getScheduler() { @@ -71,16 +71,16 @@ public final void testHandledErrorIsNotDeliveredToThreadHandler() throws Interru public void testCancelledTaskRetention() throws InterruptedException { Worker w = Schedulers.io().createWorker(); try { - ExecutorSchedulerTest.testCancelledRetention(w, false); + SchedulerTests.testCancelledRetention(w, false); } finally { w.unsubscribe(); } w = Schedulers.io().createWorker(); try { - ExecutorSchedulerTest.testCancelledRetention(w, true); + SchedulerTests.testCancelledRetention(w, true); } finally { w.unsubscribe(); } } -} \ No newline at end of file +} diff --git a/src/test/java/rx/schedulers/SchedulerTests.java b/src/test/java/rx/schedulers/SchedulerTests.java index a9146fafde..cd1d0a1912 100644 --- a/src/test/java/rx/schedulers/SchedulerTests.java +++ b/src/test/java/rx/schedulers/SchedulerTests.java @@ -4,14 +4,20 @@ import rx.Observable; import rx.Observer; import rx.Scheduler; +import rx.functions.Action0; +import rx.functions.Actions; +import rx.internal.schedulers.NewThreadWorker; +import java.lang.management.ManagementFactory; +import java.lang.management.MemoryMXBean; +import java.lang.management.MemoryUsage; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.fail; -final class SchedulerTests { +public final class SchedulerTests { private SchedulerTests() { // No instances. } @@ -23,7 +29,7 @@ private SchedulerTests() { * Schedulers which execute on a separate thread from their calling thread should exhibit this behavior. Schedulers * which execute on their calling thread may not. */ - static void testUnhandledErrorIsDeliveredToThreadHandler(Scheduler scheduler) throws InterruptedException { + public static void testUnhandledErrorIsDeliveredToThreadHandler(Scheduler scheduler) throws InterruptedException { Thread.UncaughtExceptionHandler originalHandler = Thread.getDefaultUncaughtExceptionHandler(); try { CapturingUncaughtExceptionHandler handler = new CapturingUncaughtExceptionHandler(); @@ -57,7 +63,7 @@ static void testUnhandledErrorIsDeliveredToThreadHandler(Scheduler scheduler) th * This is a companion test to {@link #testUnhandledErrorIsDeliveredToThreadHandler}, and is needed only for the * same Schedulers. */ - static void testHandledErrorIsNotDeliveredToThreadHandler(Scheduler scheduler) throws InterruptedException { + public static void testHandledErrorIsNotDeliveredToThreadHandler(Scheduler scheduler) throws InterruptedException { Thread.UncaughtExceptionHandler originalHandler = Thread.getDefaultUncaughtExceptionHandler(); try { CapturingUncaughtExceptionHandler handler = new CapturingUncaughtExceptionHandler(); @@ -88,6 +94,72 @@ static void testHandledErrorIsNotDeliveredToThreadHandler(Scheduler scheduler) t } } + public static void testCancelledRetention(Scheduler.Worker w, boolean periodic) throws InterruptedException { + System.out.println("Wait before GC"); + Thread.sleep(1000); + + System.out.println("GC"); + System.gc(); + + Thread.sleep(1000); + + + MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean(); + MemoryUsage memHeap = memoryMXBean.getHeapMemoryUsage(); + long initial = memHeap.getUsed(); + + System.out.printf("Starting: %.3f MB%n", initial / 1024.0 / 1024.0); + + int n = 500 * 1000; + if (periodic) { + final CountDownLatch cdl = new CountDownLatch(n); + final Action0 action = new Action0() { + @Override + public void call() { + cdl.countDown(); + } + }; + for (int i = 0; i < n; i++) { + if (i % 50000 == 0) { + System.out.println(" -> still scheduling: " + i); + } + w.schedulePeriodically(action, 0, 1, TimeUnit.DAYS); + } + + System.out.println("Waiting for the first round to finish..."); + cdl.await(); + } else { + for (int i = 0; i < n; i++) { + if (i % 50000 == 0) { + System.out.println(" -> still scheduling: " + i); + } + w.schedule(Actions.empty(), 1, TimeUnit.DAYS); + } + } + + memHeap = memoryMXBean.getHeapMemoryUsage(); + long after = memHeap.getUsed(); + System.out.printf("Peak: %.3f MB%n", after / 1024.0 / 1024.0); + + w.unsubscribe(); + + System.out.println("Wait before second GC"); + Thread.sleep(NewThreadWorker.PURGE_FREQUENCY + 2000); + + System.out.println("Second GC"); + System.gc(); + + Thread.sleep(1000); + + memHeap = memoryMXBean.getHeapMemoryUsage(); + long finish = memHeap.getUsed(); + System.out.printf("After: %.3f MB%n", finish / 1024.0 / 1024.0); + + if (finish > initial * 5) { + fail(String.format("Tasks retained: %.3f -> %.3f -> %.3f", initial / 1024 / 1024.0, after / 1024 / 1024.0, finish / 1024 / 1024d)); + } + } + private static final class CapturingObserver implements Observer { CountDownLatch completed = new CountDownLatch(1); int errorCount = 0;