Scheduled action no interrupt

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

@@ -67,7 +67,7 @@ public ScheduledAction scheduleActual(final Action0 action, long delayTime, Time
         } else {
             f = executor.schedule(run, delayTime, unit);
         }
-        run.add(Subscriptions.from(f));
+        run.add(f);
 
         return run;
     }

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

@@ -15,7 +15,9 @@
  */
 package rx.internal.schedulers;
 
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Subscription;
 import rx.exceptions.OnErrorNotImplementedException;
@@ -27,12 +29,11 @@
  * A {@code Runnable} that executes an {@code Action0} and can be cancelled. The analog is the
  * {@code Subscriber} in respect of an {@code Observer}.
  */
-public final class ScheduledAction implements Runnable, Subscription {
+public final class ScheduledAction extends AtomicReference<Thread> implements Runnable, Subscription {
+    /** */
+    private static final long serialVersionUID = -3962399486978279857L;
     final CompositeSubscription cancel;
     final Action0 action;
-    volatile int once;
-    static final AtomicIntegerFieldUpdater<ScheduledAction> ONCE_UPDATER
-            = AtomicIntegerFieldUpdater.newUpdater(ScheduledAction.class, "once");
 
     public ScheduledAction(Action0 action) {
         this.action = action;
@@ -42,6 +43,7 @@ public ScheduledAction(Action0 action) {
     @Override
     public void run() {
         try {
+            lazySet(Thread.currentThread());
             action.call();
         } catch (Throwable e) {
             // nothing to do but print a System error as this is fatal and there is nowhere else to throw this
@@ -66,21 +68,30 @@ public boolean isUnsubscribed() {
 
     @Override
     public void unsubscribe() {
-        if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+        if (!cancel.isUnsubscribed()) {
             cancel.unsubscribe();
         }
     }
 
     /**
-     * @warn javadoc missing
+     * Adds a general Subscription to this {@code ScheduledAction} that will be unsubscribed
+     * if the underlying {@code action} completes or the this scheduled action is cancelled.
      *
-     * @param s
-     * @warn param "s" undescribed
+     * @param s the Subscription to add
      */
     public void add(Subscription s) {
         cancel.add(s);
     }
 
+    /**
+     * Adds the given Future to the unsubscription composite in order to support
+     * cancelling the underlying task in the executor framework.
+     * @param f the future to add
+     */
+    public void add(final Future<?> f) {
+        cancel.add(new FutureCompleter(f));
+    }
+
     /**
      * Adds a parent {@link CompositeSubscription} to this {@code ScheduledAction} so when the action is
      * cancelled or terminates, it can remove itself from this parent.
@@ -92,13 +103,39 @@ public void addParent(CompositeSubscription parent) {
         cancel.add(new Remover(this, parent));
     }
 
+    /**
+     * Cancels the captured future if the caller of the call method
+     * is not the same as the runner of the outer ScheduledAction to
+     * prevent unnecessary self-interrupting if the unsubscription
+     * happens from the same thread.
+     */
+    private final class FutureCompleter implements Subscription {
+        private final Future<?> f;
+
+        private FutureCompleter(Future<?> f) {
+            this.f = f;
+        }
+
+        @Override
+        public void unsubscribe() {
+            if (ScheduledAction.this.get() != Thread.currentThread()) {
+                f.cancel(true);
+            } else {
+                f.cancel(false);
+            }
+        }
+        @Override
+        public boolean isUnsubscribed() {
+            return f.isCancelled();
+        }
+    }
+
     /** Remove a child subscription from a composite when unsubscribing. */
-    private static final class Remover implements Subscription {
+    private static final class Remover extends AtomicBoolean implements Subscription {
+        /** */
+        private static final long serialVersionUID = 247232374289553518L;
         final Subscription s;
         final CompositeSubscription parent;
-        volatile int once;
-        static final AtomicIntegerFieldUpdater<Remover> ONCE_UPDATER
-                = AtomicIntegerFieldUpdater.newUpdater(Remover.class, "once");
 
         public Remover(Subscription s, CompositeSubscription parent) {
             this.s = s;
@@ -112,7 +149,7 @@ public boolean isUnsubscribed() {
 
         @Override
         public void unsubscribe() {
-            if (ONCE_UPDATER.compareAndSet(this, 0, 1)) {
+            if (compareAndSet(false, true)) {
                 parent.remove(s);
             }
         }

src/perf/java/rx/schedulers/ComputationSchedulerPerf.java

@@ -36,7 +36,7 @@ public class ComputationSchedulerPerf {
     @State(Scope.Thread)
     public static class Input extends InputWithIncrementingInteger {
 
-        @Param({ "1", "1000", "1000000" })
+        @Param({ "1", "10", "100", "1000", "2000", "3000", "4000", "10000", "100000", "1000000" })
         public int size;
 
         @Override

src/test/java/rx/schedulers/NewThreadSchedulerTest.java

@@ -16,8 +16,18 @@
 
 package rx.schedulers;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import junit.framework.Assert;
+
 import org.junit.Test;
+
 import rx.Scheduler;
+import rx.functions.Action0;
+import rx.internal.schedulers.ScheduledAction;
+import rx.subscriptions.Subscriptions;
 
 public class NewThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
 
@@ -35,4 +45,42 @@ public final void testUnhandledErrorIsDeliveredToThreadHandler() throws Interrup
     public final void testHandledErrorIsNotDeliveredToThreadHandler() throws InterruptedException {
         SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(getScheduler());
     }
+    @Test(timeout = 3000)
+    public void testNoSelfInterrupt() throws InterruptedException {
+        Scheduler.Worker worker = Schedulers.newThread().createWorker();
+        try {
+            final CountDownLatch run = new CountDownLatch(1);
+            final CountDownLatch done = new CountDownLatch(1);
+            final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
+            final AtomicBoolean interruptFlag = new AtomicBoolean();
+
+            ScheduledAction sa = (ScheduledAction)worker.schedule(new Action0() {
+                @Override
+                public void call() {
+                    try {
+                        run.await();
+                    } catch (InterruptedException ex) {
+                        exception.set(ex);
+                    }
+                }
+            });
+
+            sa.add(Subscriptions.create(new Action0() {
+                @Override
+                public void call() {
+                    interruptFlag.set(Thread.currentThread().isInterrupted());
+                    done.countDown();
+                }
+            }));
+
+            run.countDown();
+
+            done.await();
+
+            Assert.assertEquals(null, exception.get());
+            Assert.assertFalse("Interrupted?!", interruptFlag.get());
+        } finally {
+            worker.unsubscribe();
+        }
+    }
 }