4.x: +shared, +parallel, +blocking Scheduler (#8154)

* 4.x: +shared, +parallel, +blocking Scheduler

* Improve new scheduler tests, fix access issues

* BlockingScheduler clarification

* Fix documentation style error

* Fix unit mistakes in the config record

Author: akarnokd

src/main/java/io/reactivex/rxjava4/core/Flowable.java

@@ -52,7 +52,7 @@
  * <em>Reactive Streams</em> implementations.
  * <p>
  * The {@code Flowable} hosts the default buffer size of 128 elements for operators, accessible via {@link #bufferSize()},
- * that can be overridden globally via the system parameter {@code rx4.buffer-size}. Most operators, however, have
+ * that can be overridden globally via the system parameter {@code rxjava4.buffer-size}. Most operators, however, have
  * overloads that allow setting their internal buffer size explicitly.
  * <p>
  * The documentation for this class makes use of marble diagrams. The following legend explains these diagrams:
@@ -162,7 +162,7 @@ public abstract non-sealed class Flowable<@NonNull T> implements Publisher<T>,
     /** The default buffer size. */
     static final int BUFFER_SIZE;
     static {
-        BUFFER_SIZE = Math.max(1, Integer.getInteger("rx4.buffer-size", 128));
+        BUFFER_SIZE = Math.max(1, Integer.getInteger("rxjava4.buffer-size", 128));
     }
 
     /**
@@ -251,7 +251,7 @@ public abstract non-sealed class Flowable<@NonNull T> implements Publisher<T>,
 
     /**
      * Returns the default internal buffer size used by most async operators.
-     * <p>The value can be overridden via system parameter {@code rx4.buffer-size}
+     * <p>The value can be overridden via system parameter {@code rxjava4.buffer-size}
      * <em>before</em> the {@code Flowable} class is loaded.
      * @return the default internal buffer size.
      */

src/main/java/io/reactivex/rxjava4/core/Observable.java

@@ -47,7 +47,7 @@
  * for such non-backpressured flows, which {@code Observable} itself implements as well.
  * <p>
  * The {@code Observable}'s operators, by default, run with a buffer size of 128 elements (see {@link Flowable#bufferSize()}),
- * that can be overridden globally via the system parameter {@code rx4.buffer-size}. Most operators, however, have
+ * that can be overridden globally via the system parameter {@code rxjava4.buffer-size}. Most operators, however, have
  * overloads that allow setting their internal buffer size explicitly.
  * <p>
  * The documentation for this class makes use of marble diagrams. The following legend explains these diagrams:
@@ -181,7 +181,7 @@ public abstract class Observable<@NonNull T> implements ObservableSource<T> {
     /**
      * Returns the default 'island' size or capacity-increment hint for unbounded buffers.
      * <p>Delegates to {@link Flowable#bufferSize} but is public for convenience.
-     * <p>The value can be overridden via system parameter {@code rx4.buffer-size}
+     * <p>The value can be overridden via system parameter {@code rxjava4.buffer-size}
      * <em>before</em> the {@link Flowable} class is loaded.
      * @return the default 'island' size or capacity-increment hint
      */

src/main/java/io/reactivex/rxjava4/core/Scheduler.java

@@ -61,7 +61,7 @@
  * can detect the earlier hook and not apply a new one over again.
  * <p>
  * The default implementation of {@link #now(TimeUnit)} and {@link Worker#now(TimeUnit)} methods to return current {@link System#currentTimeMillis()}
- * value in the desired time unit, unless {@code rx4.scheduler.use-nanotime} (boolean) is set. When the property is set to
+ * value in the desired time unit, unless {@code rxjava4.scheduler.use-nanotime} (boolean) is set. When the property is set to
  * {@code true}, the method uses {@link System#nanoTime()} as its basis instead. Custom {@code Scheduler} implementations can override this
  * to provide specialized time accounting (such as virtual time to be advanced programmatically).
  * Note that operators requiring a {@code Scheduler} may rely on either of the {@code now()} calls provided by
@@ -74,7 +74,7 @@
  * based on the relative time between it and {@link Worker#now(TimeUnit)}. However, drifts or changes in the
  * system clock could affect this calculation either by scheduling subsequent runs too frequently or too far apart.
  * Therefore, the default implementation uses the {@link #clockDriftTolerance()} value (set via
- * {@code rx4.scheduler.drift-tolerance} and {@code rx4.scheduler.drift-tolerance-unit}) to detect a
+ * {@code rxjava4.scheduler.drift-tolerance} and {@code rxjava4.scheduler.drift-tolerance-unit}) to detect a
  * drift in {@link Worker#now(TimeUnit)} and re-adjust the absolute/relative time calculation accordingly.
  * <p>
  * The default implementations of {@link #start()} and {@link #shutdown()} do nothing and should be overridden if the
@@ -95,10 +95,10 @@ public abstract class Scheduler {
      * <p>
      * Associated system parameter:
      * <ul>
-     *   <li>{@code rx4.scheduler.use-nanotime}, boolean, default {@code false}
+     *   <li>{@code rxjava4.scheduler.use-nanotime}, boolean, default {@code false}
      * </ul>
      */
-    static boolean IS_DRIFT_USE_NANOTIME = Boolean.getBoolean("rx4.scheduler.use-nanotime");
+    static boolean IS_DRIFT_USE_NANOTIME = Boolean.getBoolean("rxjava4.scheduler.use-nanotime");
 
     /**
      * Returns the current clock time depending on state of {@link Scheduler#IS_DRIFT_USE_NANOTIME} in given {@code unit}
@@ -122,15 +122,15 @@ static long computeNow(TimeUnit unit) {
      * <p>
      * Associated system parameters:
      * <ul>
-     * <li>{@code rx4.scheduler.drift-tolerance}, long, default {@code 15}</li>
-     * <li>{@code rx4.scheduler.drift-tolerance-unit}, string, default {@code minutes},
+     * <li>{@code rxjava4.scheduler.drift-tolerance}, long, default {@code 15}</li>
+     * <li>{@code rxjava4.scheduler.drift-tolerance-unit}, string, default {@code minutes},
      *     supports {@code seconds} and {@code milliseconds}.
      * </ul>
      */
     static final long CLOCK_DRIFT_TOLERANCE_NANOSECONDS =
             computeClockDrift(
-                    Long.getLong("rx4.scheduler.drift-tolerance", 15),
-                    System.getProperty("rx4.scheduler.drift-tolerance-unit", "minutes")
+                    Long.getLong("rxjava4.scheduler.drift-tolerance", 15),
+                    System.getProperty("rxjava4.scheduler.drift-tolerance-unit", "minutes")
             );
 
     /**
@@ -152,8 +152,8 @@ static long computeClockDrift(long time, String timeUnit) {
      * Returns the clock drift tolerance in nanoseconds.
      * <p>Related system properties:
      * <ul>
-     * <li>{@code rx4.scheduler.drift-tolerance}, long, default {@code 15}</li>
-     * <li>{@code rx4.scheduler.drift-tolerance-unit}, string, default {@code minutes},
+     * <li>{@code rxjava4.scheduler.drift-tolerance}, long, default {@code 15}</li>
+     * <li>{@code rxjava4.scheduler.drift-tolerance-unit}, string, default {@code minutes},
      *     supports {@code seconds} and {@code milliseconds}.
      * </ul>
      * @return the tolerance in nanoseconds
@@ -320,6 +320,24 @@ public ExecutorService toExecutorService(boolean useWorker) {
         return new SchedulerToExecutorService(this, new AtomicReference<>(useWorker ? createWorker() : null));
     }
 
+    /**
+     * Creates a new {@code Scheduler} that uses one of the Workers from this Scheduler
+     * and shares the access to it through its own Workers.
+     * <p>
+     * Disposing a worker doesn't dispose the underlying shared worker so other
+     * workers of this class can continue their work; use {@link #shutdown()} to release
+     * the underlying shared worker.
+     * <p>
+     * This scheduler doesn't support {@link #start()} (it's a no-op) and once {@link #shutdown()}
+     * it can't be revived.
+     * @return the shared Scheduler instance
+     * @since 4.0.0
+     */
+    @NonNull
+    public Scheduler share() {
+        return createWorker().share();
+    }
+
     /**
      * Represents an isolated, sequential worker of a parent Scheduler for executing {@code Runnable} tasks on
      * an underlying task-execution scheme (such as custom Threads, event loop, {@link java.util.concurrent.Executor Executor} or Actor system).
@@ -332,7 +350,7 @@ public ExecutorService toExecutorService(boolean useWorker) {
      * {@link #dispose()} can prevent their execution or potentially interrupt them if they are currently running.
      * <p>
      * The default implementation of the {@link #now(TimeUnit)} method returns current {@link System#currentTimeMillis()}
-     * value in the desired time unit, unless {@code rx4.scheduler.use-nanotime} (boolean) is set. When the property is set to
+     * value in the desired time unit, unless {@code rxjava4.scheduler.use-nanotime} (boolean) is set. When the property is set to
      * {@code true}, the method uses {@link System#nanoTime()} as its basis instead. Custom {@code Worker} implementations can override this
      * to provide specialized time accounting (such as virtual time to be advanced programmatically).
      * Note that operators requiring a scheduler may rely on either of the {@code now()} calls provided by
@@ -345,7 +363,7 @@ public ExecutorService toExecutorService(boolean useWorker) {
      * based on the relative time between it and {@link #now(TimeUnit)}. However, drifts or changes in the
      * system clock would affect this calculation either by scheduling subsequent runs too frequently or too far apart.
      * Therefore, the default implementation uses the {@link #clockDriftTolerance()} value (set via
-     * {@code rx4.scheduler.drift-tolerance} and {@code rx4.scheduler.drift-tolerance-unit}) to detect a drift in {@link #now(TimeUnit)} and
+     * {@code rxjava4.scheduler.drift-tolerance} and {@code rxjava4.scheduler.drift-tolerance-unit}) to detect a drift in {@link #now(TimeUnit)} and
      * re-adjust the absolute/relative time calculation accordingly.
      * <p>
      * If the {@code Worker} is disposed, the {@code schedule} methods
@@ -455,6 +473,24 @@ public long now(@NonNull TimeUnit unit) {
             return computeNow(unit);
         }
 
+        /**
+         * Creates a new {@code Scheduler} that uses this Worker
+         * and shares the access to it through its own Workers.
+         * <p>
+         * Disposing a worker doesn't dispose the underlying shared worker so other
+         * workers of this class can continue their work; use {@link #shutdown()} to release
+         * the underlying shared worker.
+         * <p>
+         * This scheduler doesn't support {@link #start()} (it's a no-op) and once {@link #shutdown()}
+         * it can't be revived.
+         * @return the shared Scheduler instance
+         * @since 4.0.0
+         */
+        @NonNull
+        public Scheduler share() {
+            return new SharedScheduler(this);
+        }
+
         /**
          * Holds state and logic to calculate when the next delayed invocation
          * of this task has to happen (accounting for clock drifts).

src/main/java/io/reactivex/rxjava4/core/config/ParallelSchedulerConfig.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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 io.reactivex.rxjava4.core.config;
+
+import java.util.concurrent.ThreadFactory;
+
+import io.reactivex.rxjava4.annotations.*;
+import io.reactivex.rxjava4.schedulers.Schedulers;
+
+/**
+ * Configuration record for {@link Schedulers#createParallel(ParallelSchedulerConfig)}.
+ * @param parallelism the number of concurrent threads, default the number of CPUs.
+ * @param tracking if true, tasks submitted to it will be tracked and can be en-masse disposed
+ * @param priority the thread priority of the created platform threads.
+ * @param threadNamePrefix the prefix to name the scheduler's threads
+ * @param factory the customizable factory for the underlying Executor, if non-null, the priority and threadNamePrefix
+ *                are ignored
+ */
+public record ParallelSchedulerConfig(
+        int parallelism,
+        boolean tracking,
+        int priority,
+        @NonNull String threadNamePrefix,
+        @Nullable ThreadFactory factory) {
+
+    /**
+     * Creates a default config with available CPUs parallelism,
+     * normal priority, tracking and RxParallelScheduler thread name prefix.
+     */
+    public ParallelSchedulerConfig() {
+        this(Runtime.getRuntime().availableProcessors(), true, Thread.NORM_PRIORITY, "RxParallelScheduler", null);
+    }
+
+    /**
+     * Creates a default config with the given parallelism,
+     * normal priority, tracking and RxParallelScheduler thread name prefix.
+     * @param parallelism the number of threads to work with in the scheduler
+     */
+    public ParallelSchedulerConfig(int parallelism) {
+        this(parallelism, true, Thread.NORM_PRIORITY, "RxParallelScheduler", null);
+    }
+
+    /**
+     * Creates a default config with the given parallelism,
+     * normal priority, optionally tracking and RxParallelScheduler thread name prefix.
+     * @param parallelism the number of threads to work with in the scheduler
+     * @param tracking if true, tasks submitted to it will be tracked and can be en-masse disposed
+     */
+    public ParallelSchedulerConfig(int parallelism, boolean tracking) {
+        this(parallelism, tracking, Thread.NORM_PRIORITY, "RxParallelScheduler", null);
+    }
+
+    /**
+     * Creates a default config with the given parallelism,
+     * normal priority, optionally tracking and RxParallelScheduler thread name prefix.
+     * @param parallelism the number of threads to work with in the scheduler
+     * @param threadNamePrefix the prefix to name the scheduler's threads
+     */
+    public ParallelSchedulerConfig(int parallelism, @NonNull String threadNamePrefix) {
+        this(parallelism, true, Thread.NORM_PRIORITY, threadNamePrefix, null);
+    }
+
+    /**
+     * Creates a default config with the given parallelism,
+     * normal priority, optionally tracking and RxParallelScheduler thread name prefix.
+     * @param parallelism the number of threads to work with in the scheduler
+     * @param tracking if true, tasks submitted to it will be tracked and can be en-masse disposed
+     * @param threadNamePrefix the prefix to name the scheduler's threads
+     */
+    public ParallelSchedulerConfig(int parallelism, boolean tracking, @NonNull String threadNamePrefix) {
+        this(parallelism, tracking, Thread.NORM_PRIORITY, threadNamePrefix, null);
+    }
+
+    /**
+     * Creates a default config with the given parallelism,
+     * normal priority, optionally tracking and RxParallelScheduler thread name prefix.
+     * @param parallelism the number of threads to work with in the scheduler
+     * @param tracking if true, tasks submitted to it will be tracked and can be en-masse disposed
+     * @param factory the customizable factory for the underlying Executor
+     */
+    public ParallelSchedulerConfig(int parallelism, boolean tracking, @NonNull ThreadFactory factory) {
+        this(parallelism, tracking, Thread.NORM_PRIORITY, "", factory);
+    }
+
+    /**
+     * Creates a fully configurable ParallelSchedulerConfig object.
+     * @param parallelism the number of threads to work with in the scheduler
+     * @param tracking if true, tasks submitted to it will be tracked and can be en-masse disposed
+     * @param priority
+     * @param threadNamePrefix the prefix to name the scheduler's threads
+     * @param factory the customizable factory for the underlying Executor, if non-null, the priority
+     *                and threadNamePrefix are ignored
+     */
+    public ParallelSchedulerConfig(
+            int parallelism,
+            boolean tracking,
+            int priority,
+            @NonNull String threadNamePrefix,
+            @Nullable ThreadFactory factory) {
+        this.parallelism = parallelism;
+        this.tracking = tracking;
+        this.priority = priority;
+        this.threadNamePrefix = threadNamePrefix;
+        this.factory = factory;
+    }
+}