3.x: Add fair mode overload to Schedulers.from(Executor) (#6744)

src/main/java/io/reactivex/rxjava3/internal/schedulers/ExecutorScheduler.java

@@ -33,20 +33,23 @@ public final class ExecutorScheduler extends Scheduler {
 
  final boolean interruptibleWorker;
 
+ final boolean fair;
+
  @NonNull
  final Executor executor;
 
  static final Scheduler HELPER = Schedulers.single();
 
- public ExecutorScheduler(@NonNull Executor executor, boolean interruptibleWorker) {
+ public ExecutorScheduler(@NonNull Executor executor, boolean interruptibleWorker, boolean fair) {
  this.executor = executor;
  this.interruptibleWorker = interruptibleWorker;
+ this.fair = fair;
  }
 
  @NonNull
  @Override
  public Worker createWorker() {
- return new ExecutorWorker(executor, interruptibleWorker);
+ return new ExecutorWorker(executor, interruptibleWorker, fair);
  }
 
  @NonNull
@@ -123,6 +126,8 @@ public static final class ExecutorWorker extends Scheduler.Worker implements Run
 
  final boolean interruptibleWorker;
 
+ final boolean fair;
+
  final Executor executor;
 
  final MpscLinkedQueue<Runnable> queue;
@@ -133,10 +138,11 @@ public static final class ExecutorWorker extends Scheduler.Worker implements Run
 
  final CompositeDisposable tasks = new CompositeDisposable();
 
- public ExecutorWorker(Executor executor, boolean interruptibleWorker) {
+ public ExecutorWorker(Executor executor, boolean interruptibleWorker, boolean fair) {
  this.executor = executor;
  this.queue = new MpscLinkedQueue<Runnable>();
  this.interruptibleWorker = interruptibleWorker;
+ this.fair = fair;
  }
 
  @NonNull
@@ -236,6 +242,36 @@ public boolean isDisposed() {
 
  @Override
  public void run() {
+ if (fair) {
+ runFair();
+ } else {
+ runEager();
+ }
+ }
+
+ void runFair() {
+ final MpscLinkedQueue<Runnable> q = queue;
+ if (disposed) {
+ q.clear();
+ return;
+ }
+
+ Runnable run = q.poll();
+ if (run != null) {
+ run.run();
+ }
+
+ if (disposed) {
+ q.clear();
+ return;
+ }
+
+ if (wip.decrementAndGet() != 0) {
+ executor.execute(this);
+ }
+ }
+
+ void runEager() {
  int missed = 1;
  final MpscLinkedQueue<Runnable> q = queue;
  for (;;) {

src/main/java/io/reactivex/rxjava3/schedulers/Schedulers.java

@@ -321,6 +321,8 @@ public static Scheduler single() {
  * non-delayed tasks as it can, which may result in a longer than expected occupation of a
  * thread of the given backing Executor. In other terms, it does not allow per-Runnable fairness
  * in case the worker runs on a shared underlying thread of the Executor.
+ * See {@link #from(Executor, boolean, boolean)} to create a wrapper that uses the underlying Executor
+ * more fairly.
  * <p>
  * Starting, stopping and restarting this scheduler is not supported (no-op) and the provided
  * executor's lifecycle must be managed externally:
@@ -346,10 +348,11 @@ public static Scheduler single() {
  * @param executor
  * the executor to wrap
  * @return the new Scheduler wrapping the Executor
+ * @see #from(Executor, boolean, boolean)
  */
  @NonNull
  public static Scheduler from(@NonNull Executor executor) {
- return new ExecutorScheduler(executor, false);
+ return new ExecutorScheduler(executor, false, false);
  }
 
  /**
@@ -382,6 +385,8 @@ public static Scheduler from(@NonNull Executor executor) {
  * non-delayed tasks as it can, which may result in a longer than expected occupation of a
  * thread of the given backing Executor. In other terms, it does not allow per-Runnable fairness
  * in case the worker runs on a shared underlying thread of the Executor.
+ * See {@link #from(Executor, boolean, boolean)} to create a wrapper that uses the underlying Executor
+ * more fairly.
  * <p>
  * Starting, stopping and restarting this scheduler is not supported (no-op) and the provided
  * executor's lifecycle must be managed externally:
@@ -411,10 +416,82 @@ public static Scheduler from(@NonNull Executor executor) {
  * be interrupted when the task is disposed.
  * @return the new Scheduler wrapping the Executor
  * @since 3.0.0
+ * @see #from(Executor, boolean, boolean)
  */
  @NonNull
  public static Scheduler from(@NonNull Executor executor, boolean interruptibleWorker) {
- return new ExecutorScheduler(executor, interruptibleWorker);
+ return new ExecutorScheduler(executor, interruptibleWorker, false);
+ }
+
+ /**
+ * Wraps an {@link Executor} into a new Scheduler instance and delegates {@code schedule()}
+ * calls to it.
+ * <p>
+ * The tasks scheduled by the returned {@link Scheduler} and its {@link io.reactivex.rxjava3.core.Scheduler.Worker Scheduler.Worker}
+ * can be optionally interrupted.
+ * <p>
+ * If the provided executor doesn't support any of the more specific standard Java executor
+ * APIs, tasks scheduled with a time delay or periodically will use the
+ * {@link #single()} scheduler for the timed waiting
+ * before posting the actual task to the given executor.
+ * <p>
+ * If the provided executor supports the standard Java {@link ExecutorService} API,
+ * canceling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
+ * a time delay or periodically will use the {@link #single()} scheduler for the timed waiting
+ * before posting the actual task to the given executor.
+ * <p>
+ * If the provided executor supports the standard Java {@link ScheduledExecutorService} API,
+ * canceling tasks scheduled by this scheduler can be cancelled/interrupted by calling
+ * {@link io.reactivex.rxjava3.disposables.Disposable#dispose()}. In addition, tasks scheduled with
+ * a time delay or periodically will use the provided executor. Note, however, if the provided
+ * {@code ScheduledExecutorService} instance is not single threaded, tasks scheduled
+ * with a time delay close to each other may end up executing in different order than
+ * the original schedule() call was issued. This limitation may be lifted in a future patch.
+ * <p>
+ * The implementation of the Worker of this wrapper Scheduler can operate in both eager (non-fair) and
+ * fair modes depending on the specified parameter. In <em>eager</em> mode, it will execute as many
+ * non-delayed tasks as it can, which may result in a longer than expected occupation of a
+ * thread of the given backing Executor. In other terms, it does not allow per-Runnable fairness
+ * in case the worker runs on a shared underlying thread of the Executor. In <em>fair</em> mode,
+ * non-delayed tasks will still be executed in a FIFO and non-overlapping manner, but after each task,
+ * the execution for the next task is rescheduled with the same underlying Executor, allowing interleaving
+ * from both the same Scheduler or other external usages of the underlying Executor.
+ * <p>
+ * Starting, stopping and restarting this scheduler is not supported (no-op) and the provided
+ * executor's lifecycle must be managed externally:
+ * <pre><code>
+ * ExecutorService exec = Executors.newSingleThreadedExecutor();
+ * try {
+ * Scheduler scheduler = Schedulers.from(exec, true, true);
+ * Flowable.just(1)
+ * .subscribeOn(scheduler)
+ * .map(v -&gt; v + 1)
+ * .observeOn(scheduler)
+ * .blockingSubscribe(System.out::println);
+ * } finally {
+ * exec.shutdown();
+ * }
+ * </code></pre>
+ * <p>
+ * This type of scheduler is less sensitive to leaking {@link io.reactivex.rxjava3.core.Scheduler.Worker Scheduler.Worker} instances, although
+ * not disposing a worker that has timed/delayed tasks not cancelled by other means may leak resources and/or
+ * execute those tasks "unexpectedly".
+ * <p>
+ * Note that this method returns a new {@link Scheduler} instance, even for the same {@link Executor} instance.
+ * @param executor
+ * the executor to wrap
+ * @param interruptibleWorker if {@code true} the tasks submitted to the {@link io.reactivex.rxjava3.core.Scheduler.Worker Scheduler.Worker} will
+ * be interrupted when the task is disposed.
+ * @param fair if {@code true} tasks submitted to the will be executed by the underlying {@link Executor} one after the other, still
+ * in a FIFO and non-overlapping manner, but allows interleaving with other tasks submitted to the underlying {@code Executor}.
+ * If {@code false}, the underlying FIFO scheme will execute as many tasks as it can before giving up the underlying {@code Executor} thread.
+ * @return the new Scheduler wrapping the Executor
+ * @since 3.0.0
+ */
+ @NonNull
+ public static Scheduler from(@NonNull Executor executor, boolean interruptibleWorker, boolean fair) {
+ return new ExecutorScheduler(executor, interruptibleWorker, fair);
  }
 
  /**