ScheduledRunnable extends AtomicReferenceArray<Object> implements Runnable, Callable<Object>, Disposable { @Override public void dispose() { for (;;) { Object o = get(FUTURE_INDEX); if (o == DONE || o == SYNC_DISPOSED || o == ASYNC_DISPOSED) { break; } boolean async = get(THREAD_INDEX) != Thread.currentThread(); if (compareAndSet(FUTURE_INDEX, o, async ? ASYNC_DISPOSED : SYNC_DISPOSED)) { if (o != null) { ((Future<?>)o).cancel(async); } break; } } for (;;) { Object o = get(PARENT_INDEX); if (o == DONE || o == PARENT_DISPOSED || o == null) { return; } if (compareAndSet(PARENT_INDEX, o, PARENT_DISPOSED)) { ((DisposableContainer)o).delete(this); return; } } } ScheduledRunnable(Runnable actual, DisposableContainer parent); @Override Object call(); @Override void run(); void setFuture(Future<?> f); @Override void dispose(); @Override boolean isDisposed();  }
@Test public void dispose() { CompositeDisposable set = new CompositeDisposable(); ScheduledRunnable run = new ScheduledRunnable(Functions.EMPTY_RUNNABLE, set); set.add(run); assertFalse(run.isDisposed()); set.dispose(); assertTrue(run.isDisposed()); }
@Test public void disposeRace() { for (int i = 0; i < TestHelper.RACE_DEFAULT_LOOPS; i++) { CompositeDisposable set = new CompositeDisposable(); final ScheduledRunnable run = new ScheduledRunnable(Functions.EMPTY_RUNNABLE, set); set.add(run); Runnable r1 = new Runnable() { @Override public void run() { run.dispose(); } }; TestHelper.race(r1, r1); assertEquals(0, set.size()); } }