forked from ReactiveX/RxJava
-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request ReactiveX#1048 from benjchristensen/executor-sched…
…uler Remove ExecutorScheduler - New ComputationScheduler
- Loading branch information
Showing
21 changed files
with
199 additions
and
449 deletions.
There are no files selected for viewing
41 changes: 0 additions & 41 deletions
41
...age-adaptors/rxjava-scala/src/main/scala/rx/lang/scala/schedulers/ExecutorScheduler.scala
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
109 changes: 109 additions & 0 deletions
109
rxjava-core/src/main/java/rx/schedulers/ComputationScheduler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| package rx.schedulers; | ||
|
|
||
| import java.util.concurrent.ThreadFactory; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicInteger; | ||
|
|
||
| import rx.Scheduler; | ||
| import rx.Subscription; | ||
| import rx.functions.Action0; | ||
| import rx.schedulers.NewThreadScheduler.OnActionComplete; | ||
| import rx.subscriptions.CompositeSubscription; | ||
| import rx.subscriptions.Subscriptions; | ||
|
|
||
| /* package */class ComputationScheduler extends Scheduler { | ||
|
|
||
| private static class ComputationSchedulerPool { | ||
| final int cores = Runtime.getRuntime().availableProcessors(); | ||
| final ThreadFactory factory = new ThreadFactory() { | ||
| final AtomicInteger counter = new AtomicInteger(); | ||
|
|
||
| @Override | ||
| public Thread newThread(Runnable r) { | ||
| Thread t = new Thread(r, "RxComputationThreadPool-" + counter.incrementAndGet()); | ||
| t.setDaemon(true); | ||
| return t; | ||
| } | ||
| }; | ||
|
|
||
| final EventLoopScheduler[] eventLoops; | ||
|
|
||
| ComputationSchedulerPool() { | ||
| // initialize event loops | ||
| eventLoops = new EventLoopScheduler[cores]; | ||
| for (int i = 0; i < cores; i++) { | ||
| eventLoops[i] = new EventLoopScheduler(factory); | ||
| } | ||
| } | ||
|
|
||
| private static ComputationSchedulerPool INSTANCE = new ComputationSchedulerPool(); | ||
|
|
||
| long n = 0; | ||
|
|
||
| public EventLoopScheduler getEventLoop() { | ||
| // round-robin selection (improvements to come) | ||
| return eventLoops[(int) (n++ % cores)]; | ||
| } | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public Inner createInner() { | ||
| return new EventLoop(); | ||
| } | ||
|
|
||
| private static class EventLoop extends Scheduler.Inner { | ||
| private final CompositeSubscription innerSubscription = new CompositeSubscription(); | ||
| private final EventLoopScheduler pooledEventLoop; | ||
| private final OnActionComplete onComplete; | ||
|
|
||
| EventLoop() { | ||
| pooledEventLoop = ComputationSchedulerPool.INSTANCE.getEventLoop(); | ||
| onComplete = new OnActionComplete() { | ||
|
|
||
| @Override | ||
| public void complete(Subscription s) { | ||
| innerSubscription.remove(s); | ||
| } | ||
|
|
||
| }; | ||
| } | ||
|
|
||
| @Override | ||
| public void unsubscribe() { | ||
| innerSubscription.unsubscribe(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isUnsubscribed() { | ||
| return innerSubscription.isUnsubscribed(); | ||
| } | ||
|
|
||
| @Override | ||
| public Subscription schedule(Action0 action) { | ||
| if (innerSubscription.isUnsubscribed()) { | ||
| // don't schedule, we are unsubscribed | ||
| return Subscriptions.empty(); | ||
| } | ||
| return pooledEventLoop.schedule(action, onComplete); | ||
| } | ||
|
|
||
| @Override | ||
| public Subscription schedule(Action0 action, long delayTime, TimeUnit unit) { | ||
| if (innerSubscription.isUnsubscribed()) { | ||
| // don't schedule, we are unsubscribed | ||
| return Subscriptions.empty(); | ||
| } | ||
|
|
||
| return pooledEventLoop.schedule(action, delayTime, unit, onComplete); | ||
| } | ||
|
|
||
| } | ||
|
|
||
| private static class EventLoopScheduler extends NewThreadScheduler.EventLoopScheduler { | ||
| EventLoopScheduler(ThreadFactory threadFactory) { | ||
| super(threadFactory); | ||
| } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.