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
<pre>
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)...)
</pre>

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:
<pre>
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
</pre>

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:
<ul>
  <li>BalancingEventLoopScheduler does poorly for small N, but does pretty well beyond a threshold.</li>
  <li>RebalancingEventLoopScheduler
    <ul>
        <li>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.</li>
<li>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. </li>
    </ul></li>
</ul>

Below, I point out the problem area for each of the new schedulers. Ideas/feedback welcome.

<h2>BalancingEventLoopScheduler</h2>
<pre>
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
</pre>

<h2>ReroutingEventLoopsScheduler</h2>
<pre>
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)
</pre>