Netflix/RxJava — pull request #1197 by spodila (2014-05-13 19:31:22+00:00)
Title: initial rev of two experimental event loop schedulers

I created two new variations of EventLoopsScheduler - BalancingEventLoopScheduler and ReroutingEventLoopScheduler. These are currently experimental since they are not fully implemented for performance, but are fully functional. There are a couple of known problems and still evolving solutions. However, the ideas are concrete enough to garner discussion and feedback. The objective of rebalancing workload when there is non-trivial disparity in loads across cores is constrained by costs to both maintain state as well as to run rebalancing algorithms. Especially while maintaining the contracts of event concurrency, etc. and minimizing synchronization for gathering state and rebalancing.
I also created a new JMH benchmark, MultipleObserversOnOneBusyObservable.java (pardon the naming), to help compare these two new variants with the existing scheduler.

Here's a description of the benchmark and the two new schedulers.

JMH Benchmark MultipleObserversOnOneBusyObservable.java

setup Input
theArray = Array of N numbers
Array of schedulers (schedlrIndex)
0: EventLoopsScheduler
1: BalancingEventLoopScheduler
2: ReroutingEventLoopScheduler
numObservers = NUM_CORES * 10
@GenerateMicroBenchmark
runAll(input)
for(i = 1 .. input.numObservers)
if(i%4 == 0)
theO = Observable.from(1,2,3,4,5)
else
theO = Observable.from(input.theArray)
theO.observeOn(input.scheduler[schedlrIndex]).subscribe(... input.blackhHole.consume(number)...)

Benchmark run on Ubuntu desktop: Intel Xeon W3550 @ 3.7 GHz, 8-Cores, 40GB memory.

gradlew benchmarks '-Pjmh=-f 1 -tu s -bm thrpt -wi 5 -i 10 -r 1 .*MultipleObserversOnOneBusyObservable.*'

Results:

N=50

Benchmark (schedulerIndex) Mode Samples Mean Mean error Units
r.s.MultipleObserversOnOneBusyObservable.runAll 0 thrpt 10 3460.543 50.047 ops/s
r.s.MultipleObserversOnOneBusyObservable.runAll 1 thrpt 10 430.402 1.783 ops/s
r.s.MultipleObserversOnOneBusyObservable.runAll 2 thrpt 10 3200.517 140.199 ops/s

N=500

Benchmark (schedulerIndex) Mode Samples Mean Mean error Units
r.s.MultipleObserversOnOneBusyObservable.runAll 0 thrpt 10 345.918 8.889 ops/s
r.s.MultipleObserversOnOneBusyObservable.runAll 1 thrpt 10 366.105 11.945 ops/s
r.s.MultipleObserversOnOneBusyObservable.runAll 2 thrpt 10 361.587 10.627 ops/s

N=5000

Benchmark (schedulerIndex) Mode Samples Mean Mean error Units
r.s.MultipleObserversOnOneBusyObservable.runAll 0 thrpt 10 36.383 1.698 ops/s
r.s.MultipleObserversOnOneBusyObservable.runAll 1 thrpt 10 88.317 4.917 ops/s
r.s.MultipleObserversOnOneBusyObservable.runAll 2 thrpt 10 37.533 0.951 ops/s

Note that the workload in this benchmark isn't such that there's a much rebalancing, if any, to do. However, we need to make sure that the variants do not degrade performance in that case.

Some observations:

BalancingEventLoopScheduler does poorly for small N, but does pretty well beyond a threshold.

RebalancingEventLoopScheduler

added logic to evaluate and maintain state (1. avgLatency per core, 2. hot queue per core) and still is able to do as good as the old EventLoopsScheduler. That is, there's near zero cost to maintaining that state.

Even if rebalancing proves costly and we evolve an algorithm later, the result so far itself might be a good improvement over the current one if we just "pause" the hot Q and randomly pick another core.

Below, I point out the problem area for each of the new schedulers. Ideas/feedback welcome.

BalancingEventLoopScheduler

N worker threads are created, 1 per core
Each of N worker threads are assigned 0 or more worker queues to process
One of the worker threads is assigned as the RebalancerThread for load balancing
Scheduler.createWorker() creates a queue and assigns to one of the worker threads
WorkerThread.run():
act on Q transfers (transferring queue in, or out)
grab updated list of Qs
totalLoad=0
for(Q q: theQlist)
q.drainTo(workList)
now = currentTime
for(Action a: workList)
a.call()
load = currentTime - now
totalLoad += load
maintain hotQueue: name and load; that has the highest load
update current worker thread's hot queue
if I am the RebalancerThread, call rebalanceQs()
if(totalLoad == 0)
did no work, sleep a bit to prevent busy cpu **PROBLEM**
would be nice to have a multi-queue poll() here to be called on all our Qs
this sleep kills performance when work comes in relatively slowly
rebalanceQs()
if did rebalance recently, return
get HotQueues from each core thread workers
if none of them had a hot q, return
sort in increasing order, hot Qs based on core thread worker's totalLoad and the qLoad
for loadedWorker: from end of sorted hot Q list
for availableWorker: from beginning of sorted hotQ list until loadedWorker
if(shouldRebalancePair(loadedWorker, availableWorker), then
move the hotQ from loadedWorker to availableWorker
That is, WorkerThread[loadedWorker].transferQueue(theHotQ, WorkerThread[availableWorker])
shouldRebalancePair(loadedWorker, availableWorker)
if( (loadedWorker.totalLoad - loadedWorker.hotQLoad) >= (availableWorker.totalLoad + loadedWorker.hotQLoad) )
return true
return false

ReroutingEventLoopsScheduler

N worker threads are created, 1 per core
Each of N worker threads has a single thread executor (ScheduledThreadPoolExecutor)
Scheduler.createWorker() creates a WorkerQueue and assigns to one of the worker threads (could possibly make it pick the least loaded one; that info is available based on the details below)
WorkerQueue.schedule() calls workerThread.schedule() on the currently assigned workerThread
WorkerThread routinely schedules a house keeping action, latencyEval, into its schedulerService to note time it takes for it to get executed (this measures latency)
WorkerThread.schedule(callerQId, action)
wrap each runnable with
increment counter to track #items
increment per queue counter to track per queueId throughput (queueCounts)
executor.schedule(wrapped runnable)
WorkerThread essentially maintains two numbers: avgLatency and throughput (overall + perWorker)
WorkerThread.latencyEval
note latency (currentTime - latencyEvalLaunchTime)
if(queueCounts.size > 1)
maxQ = queue from queueCounts with max number, where number > EPS (threshold)
destination = shouldReroute(myWorkerIndex)
if(desination >= 0)
WorkerQueue[maxQ].rerouteTo(destination)
shouldReroute(sourceWorker)
ensure nobody else is rerouting right now or just did
get avgLatencies for all workers (as many as #Cores)
pick a worker with that can take sourceWorker's load based on latencies (currently always return -1, effectively actual load balancing is turned off)
return picked worker
WorkerQueue.schedule(action)
if am being rerouted
initialize reroute
put an ending marker action into current worker that notifies us of the last action handled
we can use the new worker only after the last one sent to previous one completes
if (ending marker indicated)
switch worker to the new one
move all actions in transfer queue to new worker
else
park action into a transfer queue
Don't yet know how to manage subscriptions when parking actions **PROBLEM**
It is functional though since actual rebalancing isn't happening.
Perhaps, don't park actions, just pause until ending marker comes through. Subscriptions are managed just like before then. The caller is then blocked until rebalancing finishes. Which make be OK?
else
worker.schedule(action)