2.x: internal API to get distinct Workers from some Schedulers #5741

akarnokd · 2017-11-24T14:54:28Z

This PR adds an internal interface SchedulerMultiWorkerSupport that allows retrieving multiple workers from a Scheduler that implements this interface.

The standard Scheduler.getWorker() can be invoked as many times as necessary, but specific implementations such as the computation() Scheduler and the ParallelScheduler is not guaranteed to return workers that are backed by distinct single-threaded thread pools.

This does not effect other scheduler types because:

they are single threaded (single()) or don't use threads at all (trampoline()) and
already hand out distinct workers (io(), newThread()).

Such worker reuse can happen when in a highly concurrent application, typical tasks are mixed with parallel tasks and both pull out workers from these Schedulers. If this is happens, it is possible there will be duplicate threads used by the parallel operations and thus not utilize the originally intended parallelism level.

By implementing this suggested interface, a batch-retrieval can be supported by the Schedulers and they can make sure the caller gets as many distinct thread-pool as possible. If more workers are requested than the Scheduler's parallelism, the workers are handed out in round-robin fashion similar to the standard createWorker().

Why a callback instead of returning an array?

even if both require an allocation to set up, there is no need to have all workers visible at once,
simply less memory usage, and
avoids looping twice: once for filling in the array and once for using the elements of the array.

codecov · 2017-11-24T15:09:35Z

Codecov Report

Merging #5741 into 2.x will decrease coverage by 0.11%.
The diff coverage is 100%.

@@             Coverage Diff              @@
##                2.x    #5741      +/-   ##
============================================
- Coverage     96.29%   96.18%   -0.12%     
  Complexity     5833     5833              
============================================
  Files           634      634              
  Lines         41615    41639      +24     
  Branches       5761     5766       +5     
============================================
- Hits          40073    40050      -23     
- Misses          611      640      +29     
- Partials        931      949      +18

Impacted Files	Coverage Δ	Complexity Δ
...vex/internal/operators/parallel/ParallelRunOn.java	`97.1% <100%> (-0.37%)`	`8 <2> (+2)`
...ivex/internal/schedulers/ComputationScheduler.java	`97.53% <100%> (+0.51%)`	`14 <1> (+1)`	⬆️
.../operators/observable/ObservableFlatMapSingle.java	`88.8% <0%> (-5.98%)`	`2% <0%> (ø)`
...nternal/operators/observable/ObservableCreate.java	`92.3% <0%> (-5.13%)`	`2% <0%> (ø)`
...ava/io/reactivex/processors/BehaviorProcessor.java	`89.38% <0%> (-4.87%)`	`61% <0%> (ø)`
...tivex/internal/observers/FutureSingleObserver.java	`94.33% <0%> (-3.78%)`	`24% <0%> (-1%)`
...rnal/operators/observable/ObservableObserveOn.java	`97.95% <0%> (-2.05%)`	`3% <0%> (ø)`
...ernal/operators/flowable/FlowableFlatMapMaybe.java	`93.23% <0%> (-1.94%)`	`2% <0%> (ø)`
...main/java/io/reactivex/subjects/ReplaySubject.java	`96.25% <0%> (-1.77%)`	`50% <0%> (-1%)`
...rnal/operators/flowable/FlowableFlatMapSingle.java	`91.3% <0%> (-1.64%)`	`2% <0%> (ø)`
... and 22 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update e25be7c...a787346. Read the comment docs.

artem-zinnatullin

Rename PR to "Internal API to …"?

artem-zinnatullin · 2017-11-27T20:39:37Z

src/main/java/io/reactivex/internal/schedulers/ComputationScheduler.java

+
+        @Override
+        public void createWorkers(int number, WorkerCallback callback) {
+            int c = cores;


cores is final in this context, why copy it to local var?

With potential volatiles around, this won't re-read the field every time it is needed in the loop below.

artem-zinnatullin · 2017-11-27T20:41:30Z

src/main/java/io/reactivex/internal/schedulers/SchedulerMultiWorkerSupport.java

+     * The callback interface for the {@link SchedulerMultiWorkerSupport#createWorkers(int, WorkerCallback)}
+     * method.
+     */
+    interface WorkerCallback {


WorkerCreatedCallback?

While still it is an internal API, I don't think we need a more complicated naming.

artem-zinnatullin · 2017-11-27T20:41:32Z

src/main/java/io/reactivex/internal/schedulers/SchedulerMultiWorkerSupport.java

+         * @param index the worker index, zero-based
+         * @param worker the worker instance
+         */
+        void onWorker(int index, @NonNull Scheduler.Worker worker);


onCreated?

What's the use case for the index?

In ParallelRunOn, it let's index into the parent and subscribers arrays without the need for additional state, such as counting how many times onWorker was invoked.

ok, potentially reduces allocations for the user, thanks

artem-zinnatullin · 2017-11-27T21:35:38Z

👍

davidmoten

Good idea, will be useful.

davidmoten · 2017-11-27T22:17:29Z

src/main/java/io/reactivex/internal/schedulers/SchedulerMultiWorkerSupport.java

+         * @param index the worker index, zero-based
+         * @param worker the worker instance
+         */
+        void onWorker(int index, @NonNull Scheduler.Worker worker);


What's the use case for the index?

davidmoten · 2017-11-27T23:01:26Z

The PR has already been merged but for the record I approve too

akarnokd · 2017-11-27T23:04:07Z

Thanks for the reviews @artem-zinnatullin @vanniktech & @davidmoten !

2.x: API to get distinct Workers from some Schedulers

a787346

akarnokd added 2.x Enhancement labels Nov 24, 2017

akarnokd added this to the 2.2 milestone Nov 24, 2017

artem-zinnatullin approved these changes Nov 27, 2017

View reviewed changes

akarnokd changed the title ~~2.x: API to get distinct Workers from some Schedulers~~ 2.x: internal API to get distinct Workers from some Schedulers Nov 27, 2017

vanniktech approved these changes Nov 27, 2017

View reviewed changes

akarnokd merged commit d2fe631 into ReactiveX:2.x Nov 27, 2017

akarnokd deleted the SchedulerMultiWorker branch November 27, 2017 22:12

davidmoten reviewed Nov 27, 2017

View reviewed changes

akarnokd mentioned this pull request Dec 18, 2017

2.1.8 release preparations #5769

Closed

jared-bot mentioned this pull request Dec 27, 2017

Dependency update: rxjava xmartlabs/bigbang-template#45

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

2.x: internal API to get distinct Workers from some Schedulers #5741

2.x: internal API to get distinct Workers from some Schedulers #5741

akarnokd commented Nov 24, 2017

codecov bot commented Nov 24, 2017

artem-zinnatullin left a comment

artem-zinnatullin Nov 27, 2017

akarnokd Nov 27, 2017

artem-zinnatullin Nov 27, 2017

akarnokd Nov 27, 2017 •

edited

Loading

artem-zinnatullin Nov 27, 2017

davidmoten Nov 27, 2017

akarnokd Nov 27, 2017

davidmoten Nov 27, 2017

artem-zinnatullin commented Nov 27, 2017

davidmoten left a comment

davidmoten Nov 27, 2017

davidmoten commented Nov 27, 2017

akarnokd commented Nov 27, 2017

2.x: internal API to get distinct Workers from some Schedulers #5741

2.x: internal API to get distinct Workers from some Schedulers #5741

Conversation

akarnokd commented Nov 24, 2017

codecov bot commented Nov 24, 2017

Codecov Report

artem-zinnatullin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akarnokd Nov 27, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artem-zinnatullin commented Nov 27, 2017

davidmoten left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

davidmoten commented Nov 27, 2017

akarnokd commented Nov 27, 2017

akarnokd Nov 27, 2017 •

edited

Loading