();
- this.lock = new ReentrantLock();
- this.condition = this.lock.newCondition();
- this.running = new AtomicBoolean();
- this.shutdown = new AtomicBoolean();
- this.wip = new AtomicLong();
- this.timedHelper = Schedulers.single();
- }
-
- /**
- * Begin executing the blocking event loop without any initial action.
- *
- * This method will block until the {@link #shutdown()} is invoked.
- * @see #execute(Action)
- */
- public void execute() {
- execute(Functions.EMPTY_ACTION);
- }
-
- /**
- * Begin executing the blocking event loop with the given initial action
- * (usually contain the rest of the 'main' method).
- *
- * This method will block until the {@link #shutdown()} is invoked.
- * @param action the action to execute
- */
- public void execute(Action action) {
- ObjectHelper.requireNonNull(action, "action is null");
- if (!running.get() && running.compareAndSet(false, true)) {
- thread = Thread.currentThread();
- queue.offer(action);
- wip.getAndIncrement();
- drainLoop();
- }
- }
-
- void drainLoop() {
- final AtomicBoolean stop = shutdown;
- final AtomicLong wip = this.wip;
-
- for (;;) {
- if (stop.get()) {
- cancelAll();
- return;
- }
- do {
- Action a = queue.poll();
- if (a == SHUTDOWN) {
- cancelAll();
- return;
- }
- try {
- a.run();
- } catch (Throwable ex) {
- RxJavaPlugins.onError(ex);
- }
- } while (wip.decrementAndGet() != 0);
-
- if (wip.get() == 0 && !stop.get()) {
- lock.lock();
- try {
- while (wip.get() == 0 && !stop.get()) {
- condition.await();
- }
- } catch (InterruptedException ex) {
- // deliberately ignored
- } finally {
- lock.unlock();
- }
- }
- }
- }
-
- void cancelAll() {
- final ConcurrentLinkedQueue q = queue;
-
- Action a;
-
- while ((a = q.poll()) != null) {
- if (a instanceof Disposable) {
- ((Disposable)a).dispose();
- }
- }
- }
-
- @Override
- public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
- ObjectHelper.requireNonNull(run, "run is null");
- ObjectHelper.requireNonNull(unit, "unit is null");
- if (shutdown.get()) {
- return Disposables.disposed();
- }
-
- final BlockingDirectTask task = new BlockingDirectTask(run);
-
- if (delay == 0L) {
- enqueue(task);
- return task;
- }
-
- SequentialDisposable inner = new SequentialDisposable();
- final SequentialDisposable outer = new SequentialDisposable(inner);
-
- Disposable d = timedHelper.scheduleDirect(new Runnable() {
- @Override
- public void run() {
- outer.replace(task);
- enqueue(task);
- }
- }, delay, unit);
-
- if (d == Disposables.disposed()) {
- return d;
- }
-
- inner.replace(d);
-
- return outer;
- }
-
- @Override
- public void shutdown() {
- if (shutdown.compareAndSet(false, true)) {
- enqueue(SHUTDOWN);
- }
- }
-
- void enqueue(Action action) {
- queue.offer(action);
- if (wip.getAndIncrement() == 0L) {
- lock.lock();
- try {
- condition.signal();
- } finally {
- lock.unlock();
- }
- }
- }
-
- @Override
- public Worker createWorker() {
- return new BlockingWorker();
- }
-
- static final int READY = 0;
- static final int RUNNING = 1;
- static final int INTERRUPTING = 2;
- static final int INTERRUPTED = 3;
- static final int FINISHED = 4;
- static final int CANCELLED = 5;
-
- final class BlockingDirectTask
- extends AtomicInteger
- implements Action, Disposable {
-
- private static final long serialVersionUID = -9165914884456950194L;
- final Runnable task;
-
- BlockingDirectTask(Runnable task) {
- this.task = task;
- }
-
- @Override
- public void run() throws Exception {
- try {
- if (compareAndSet(READY, RUNNING)) {
- try {
- task.run();
- } finally {
- compareAndSet(RUNNING, FINISHED);
- }
- }
- } finally {
- while (get() == INTERRUPTING) { }
-
- if (get() == INTERRUPTED) {
- Thread.interrupted();
- }
- }
- }
-
- @Override
- public void dispose() {
- for (;;) {
- int s = get();
- if (s >= INTERRUPTING) {
- break;
- }
-
- if (s == READY && compareAndSet(READY, CANCELLED)) {
- break;
- }
- if (compareAndSet(RUNNING, INTERRUPTING)) {
- Thread t = thread;
- if (t != null) {
- t.interrupt();
- }
- set(INTERRUPTED);
- break;
- }
- }
- }
-
- @Override
- public boolean isDisposed() {
- return get() >= INTERRUPTING;
- }
- }
-
- final class BlockingWorker extends Worker {
-
- final CompositeDisposable tasks;
-
- BlockingWorker() {
- this.tasks = new CompositeDisposable();
- }
-
- @Override
- public void dispose() {
- tasks.dispose();
- }
-
- @Override
- public boolean isDisposed() {
- return tasks.isDisposed();
- }
-
- @Override
- public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
- ObjectHelper.requireNonNull(run, "run is null");
- ObjectHelper.requireNonNull(unit, "unit is null");
-
- if (shutdown.get() || isDisposed()) {
- return Disposables.disposed();
- }
-
- final BlockingTask task = new BlockingTask(run);
- tasks.add(task);
-
- if (delay == 0L) {
- enqueue(task);
- return task;
- }
-
- SequentialDisposable inner = new SequentialDisposable();
- final SequentialDisposable outer = new SequentialDisposable(inner);
-
- Disposable d = timedHelper.scheduleDirect(new Runnable() {
- @Override
- public void run() {
- outer.replace(task);
- enqueue(task);
- }
- }, delay, unit);
-
- if (d == Disposables.disposed()) {
- return d;
- }
-
- inner.replace(d);
-
- return outer;
- }
-
- final class BlockingTask
- extends AtomicInteger
- implements Action, Disposable {
-
- private static final long serialVersionUID = -9165914884456950194L;
-
- final Runnable task;
-
- BlockingTask(Runnable task) {
- this.task = task;
- }
-
- @Override
- public void run() throws Exception {
- try {
- if (compareAndSet(READY, RUNNING)) {
- try {
- task.run();
- } finally {
- compareAndSet(RUNNING, FINISHED);
- tasks.remove(this);
- }
- }
- } finally {
- while (get() == INTERRUPTING) { }
-
- if (get() == INTERRUPTED) {
- Thread.interrupted();
- }
- }
- }
-
- @Override
- public void dispose() {
- for (;;) {
- int s = get();
- if (s >= INTERRUPTING) {
- return;
- }
-
- if (s == READY && compareAndSet(READY, CANCELLED)) {
- break;
- }
- if (compareAndSet(RUNNING, INTERRUPTING)) {
- Thread t = thread;
- if (t != null) {
- t.interrupt();
- }
- set(INTERRUPTED);
- break;
- }
- }
- tasks.remove(this);
- }
-
- @Override
- public boolean isDisposed() {
- return get() >= INTERRUPTING;
- }
- }
- }
-}
diff --git a/src/main/java/hu/akarnokd/rxjava2/schedulers/ParallelScheduler.java b/src/main/java/hu/akarnokd/rxjava2/schedulers/ParallelScheduler.java
deleted file mode 100644
index 9b345bbe..00000000
--- a/src/main/java/hu/akarnokd/rxjava2/schedulers/ParallelScheduler.java
+++ /dev/null
@@ -1,431 +0,0 @@
-/*
- * Copyright 2016-2019 David Karnok
- *
- * 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 hu.akarnokd.rxjava2.schedulers;
-
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicReference;
-
-import io.reactivex.Scheduler;
-import io.reactivex.disposables.*;
-import io.reactivex.exceptions.Exceptions;
-import io.reactivex.internal.disposables.*;
-import io.reactivex.internal.functions.Functions;
-import io.reactivex.internal.schedulers.RxThreadFactory;
-import io.reactivex.plugins.RxJavaPlugins;
-
-/**
- * Scheduler with a configurable fixed amount of thread-pools.
- */
-public final class ParallelScheduler extends Scheduler {
-
- static final ScheduledExecutorService[] SHUTDOWN;
-
- static final ScheduledExecutorService REJECTING;
-
- final ThreadFactory factory;
-
- final int parallelism;
-
- final boolean tracking;
-
- final AtomicReference pool;
-
- int n;
-
- static {
- SHUTDOWN = new ScheduledExecutorService[0];
-
- REJECTING = Executors.newSingleThreadScheduledExecutor();
- REJECTING.shutdownNow();
- }
-
- public ParallelScheduler() {
- this(Runtime.getRuntime().availableProcessors());
- }
-
- public ParallelScheduler(String threadNamePrefix) {
- this(Runtime.getRuntime().availableProcessors(), new RxThreadFactory(threadNamePrefix));
- }
-
- public ParallelScheduler(int parallelism) {
- this(parallelism, true);
- }
-
- public ParallelScheduler(int parallelism, boolean tracking) {
- this(parallelism, tracking, Thread.NORM_PRIORITY);
- }
-
- public ParallelScheduler(int parallelism, boolean tracking, int priority) {
- this(parallelism, tracking, priority, "RxParallelScheduler");
- }
-
- public ParallelScheduler(int parallelism, boolean tracking, int priority, String threadNamePrefix) {
- this(parallelism, new RxThreadFactory(threadNamePrefix, checkPriority(priority)), tracking);
- }
-
- static int checkPriority(int priority) {
- if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {
- throw new IllegalArgumentException("priority out of range");
- }
- return priority;
- }
-
- public ParallelScheduler(int parallelism, ThreadFactory factory) {
- this(parallelism, factory, true);
- }
-
- public ParallelScheduler(int parallelism, ThreadFactory factory, boolean tracking) {
- if (parallelism <= 0) {
- throw new IllegalArgumentException("parallelism > 0 required but it was " + parallelism);
- }
- this.parallelism = parallelism;
- this.factory = factory;
- this.tracking = tracking;
- this.pool = new AtomicReference(SHUTDOWN);
- start();
- }
-
- @Override
- public void start() {
- ScheduledExecutorService[] next = null;
- for (;;) {
- ScheduledExecutorService[] current = pool.get();
- if (current != SHUTDOWN) {
- if (next != null) {
- for (ScheduledExecutorService exec : next) {
- exec.shutdownNow();
- }
- }
- return;
- }
- if (next == null) {
- next = new ScheduledExecutorService[parallelism];
- for (int i = 0; i < next.length; i++) {
- next[i] = Executors.newSingleThreadScheduledExecutor(factory);
- }
- }
-
- if (pool.compareAndSet(current, next)) {
- return;
- }
- }
- }
-
- @Override
- public void shutdown() {
- for (;;) {
- ScheduledExecutorService[] current = pool.get();
- if (current == SHUTDOWN) {
- return;
- }
- if (pool.compareAndSet(current, SHUTDOWN)) {
- for (ScheduledExecutorService exec : current) {
- exec.shutdownNow();
- }
- }
- }
- }
-
- ScheduledExecutorService pick() {
- ScheduledExecutorService[] current = pool.get();
- if (current.length == 0) {
- return REJECTING;
- }
- int idx = this.n;
- if (idx >= parallelism) {
- idx = 0;
- }
- this.n = idx + 1; // may race, we don't care
- return current[idx];
- }
-
- @Override
- public Worker createWorker() {
- if (tracking) {
- return new TrackingParallelWorker(pick());
- }
- return new NonTrackingParallelWorker(pick());
- }
-
- @Override
- public Disposable scheduleDirect(Runnable run) {
- ScheduledExecutorService exec = pick();
- if (exec == REJECTING) {
- return Disposables.disposed();
- }
- try {
- return Disposables.fromFuture(exec.submit(RxJavaPlugins.onSchedule(run)));
- } catch (RejectedExecutionException ex) {
- return Disposables.disposed();
- }
- }
-
- @Override
- public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
- ScheduledExecutorService exec = pick();
- if (exec == REJECTING) {
- return Disposables.disposed();
- }
- try {
- return Disposables.fromFuture(exec.schedule(RxJavaPlugins.onSchedule(run), delay, unit));
- } catch (RejectedExecutionException ex) {
- return Disposables.disposed();
- }
- }
-
- @Override
- public Disposable schedulePeriodicallyDirect(Runnable run, long initialDelay, long period, TimeUnit unit) {
- ScheduledExecutorService exec = pick();
- if (exec == REJECTING) {
- return Disposables.disposed();
- }
- try {
- return Disposables.fromFuture(exec.scheduleAtFixedRate(RxJavaPlugins.onSchedule(run), initialDelay, period, unit));
- } catch (RejectedExecutionException ex) {
- return Disposables.disposed();
- }
- }
-
- static final class NonTrackingParallelWorker extends Worker {
-
- final ScheduledExecutorService exec;
-
- volatile boolean shutdown;
-
- NonTrackingParallelWorker(ScheduledExecutorService exec) {
- this.exec = exec;
- }
-
- @Override
- public void dispose() {
- shutdown = true;
- }
-
- @Override
- public boolean isDisposed() {
- return shutdown;
- }
-
- @Override
- public Disposable schedule(Runnable run) {
- if (!shutdown) {
- try {
- NonTrackingTask ntt = new NonTrackingTask(RxJavaPlugins.onSchedule(run));
- exec.submit(ntt);
- return ntt;
- } catch (RejectedExecutionException ex) {
- // just let it fall through
- }
- }
- return Disposables.disposed();
- }
-
- @Override
- public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
- if (!shutdown) {
- try {
- NonTrackingTask ntt = new NonTrackingTask(RxJavaPlugins.onSchedule(run));
- exec.schedule(ntt, delay, unit);
- return ntt;
- } catch (RejectedExecutionException ex) {
- // just let it fall through
- }
- }
- return Disposables.disposed();
- }
-
- // Not implementing a custom schedulePeriodically as it would require tracking the Future.
-
- final class NonTrackingTask implements Callable
