/*
 * Copyright (c) 2011-2017 Pivotal Software Inc, All Rights Reserved.
 *
 * 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
 *
 *       https://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 reactor.core.scheduler;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

import reactor.core.Disposable;
import reactor.util.annotation.Nullable;

A runnable task for Scheduler Workers that are time-capable (implementing a relevant schedule(delay) and schedulePeriodically(period) methods). Unlike the one in DelegateServiceScheduler, this runnable doesn't expose the ability to cancel inner task when interrupted.
Author:Simon Baslé, David Karnok
/** * A runnable task for {@link Scheduler} Workers that are time-capable (implementing a * relevant schedule(delay) and schedulePeriodically(period) methods). * * Unlike the one in {@link DelegateServiceScheduler}, this runnable doesn't expose the * ability to cancel inner task when interrupted. * * @author Simon Baslé * @author David Karnok */
final class WorkerTask implements Runnable, Disposable, Callable<Void> { final Runnable task;
marker that the Worker was disposed and the parent got notified
/** marker that the Worker was disposed and the parent got notified */
static final Composite DISPOSED = new EmptyCompositeDisposable();
marker that the Worker has completed, for the PARENT field
/** marker that the Worker has completed, for the PARENT field */
static final Composite DONE = new EmptyCompositeDisposable();
marker that the Worker has completed, for the FUTURE field
/** marker that the Worker has completed, for the FUTURE field */
static final Future<Void> FINISHED = new FutureTask<>(() -> null);
marker that the Worker was cancelled from the same thread (ie. within call()/run()), which means setFuture might race: we avoid interrupting the Future in this case.
/** * marker that the Worker was cancelled from the same thread (ie. within call()/run()), * which means setFuture might race: we avoid interrupting the Future in this case. */
static final Future<Void> SYNC_CANCELLED = new FutureTask<>(() -> null);
marker that the Worker was cancelled from another thread, making it safe to interrupt the Future task.
/** * marker that the Worker was cancelled from another thread, making it safe to * interrupt the Future task. */
//see https://github.com/reactor/reactor-core/issues/1107 static final Future<Void> ASYNC_CANCELLED = new FutureTask<>(() -> null); volatile Future<?> future; static final AtomicReferenceFieldUpdater<WorkerTask, Future> FUTURE = AtomicReferenceFieldUpdater.newUpdater(WorkerTask.class, Future.class, "future"); volatile Composite parent; static final AtomicReferenceFieldUpdater<WorkerTask, Composite> PARENT = AtomicReferenceFieldUpdater.newUpdater(WorkerTask.class, Composite.class, "parent"); volatile Thread thread; static final AtomicReferenceFieldUpdater<WorkerTask, Thread> THREAD = AtomicReferenceFieldUpdater.newUpdater(WorkerTask.class, Thread.class, "thread"); WorkerTask(Runnable task, Composite parent) { this.task = task; PARENT.lazySet(this, parent); } @Override @Nullable public Void call() { THREAD.lazySet(this, Thread.currentThread()); try { try { task.run(); } catch (Throwable ex) { Schedulers.handleError(ex); } } finally { THREAD.lazySet(this, null); Composite o = parent; //note: the o != null check must happen after the compareAndSet for it to always mark task as DONE if (o != DISPOSED && PARENT.compareAndSet(this, o, DONE) && o != null) { o.remove(this); } Future f; for (;;) { f = future; if (f == SYNC_CANCELLED || f == ASYNC_CANCELLED || FUTURE.compareAndSet(this, f, FINISHED)) { break; } } } return null; } @Override public void run() { call(); } void setFuture(Future<?> f) { for (;;) { Future o = future; if (o == FINISHED) { return; } if (o == SYNC_CANCELLED) { f.cancel(false); return; } if (o == ASYNC_CANCELLED) { f.cancel(true); return; } if (FUTURE.compareAndSet(this, o, f)) { return; } } } @Override public boolean isDisposed() { Composite o = PARENT.get(this); return o == DISPOSED || o == DONE; } @Override public void dispose() { for (;;) { Future f = future; if (f == FINISHED || f == SYNC_CANCELLED || f == ASYNC_CANCELLED) { break; } boolean async = thread != Thread.currentThread(); if (FUTURE.compareAndSet(this, f, async ? ASYNC_CANCELLED : SYNC_CANCELLED)) { if (f != null) { f.cancel(async); } break; } } for (;;) { Composite o = parent; if (o == DONE || o == DISPOSED || o == null) { return; } if (PARENT.compareAndSet(this, o, DISPOSED)) { o.remove(this); return; } } } }