codegram/futuroscope — pull request #23 by ggPeti (2014-09-29 14:58:40+00:00)
Title: Huge feature pack: Deadlock detection, prioritized futures, smart up/down spinning of workers, logging

Apologies for the giant PR, but one thing led to another, and it didn't make sense to implement one without another (well, logging could have been separated, but it was very useful for debugging while implementing the rest).

The features implemented are:

#### Deadlock detection
There are 2 kinds of deadlocks that we can detect.
##### 1. Circular dependency
When futures depend on each other in a circle, Futuroscope will detect that and send each thread involved a DeadlockError with a message describing the situation. For example:
```
2.1.0 :001 > f2 = nil
=> nil
2.1.0 :002 > f1 = future { f2 = future { f1.future_value }; f2.future_value }; f1.inspect
Futuroscope::DeadlockError: Cyclical dependency detected, the future was aborted.
(...stack trace...)
2.1.0 :004 > f2.inspect
Futuroscope::DeadlockError: Cyclical dependency detected, the future was aborted.
(...stack trace...)
```
##### 2. Pool size too low
When the pool is full, but all futures are waiting for another future that doesn't have a worker yet, Futuroscope will fail a future which does not depend on any other future (and the one with the least priority out of those). (Note: This selection might be optimized to selecting the root of the smallest unary branch of the dependency forest instead of the root of the smallest tree, but this is a good enough solution for a situation that rarely occurs.)
For example:
```
2.1.0 :001 > Futuroscope.default_pool = Futuroscope::Pool.new 1..1
=> #, @workers=#, @free=true, @thread=#>}>, @mutex=#, @futures={}>
2.1.0 :002 > future { future { 1 } + 1 }
Futuroscope::DeadlockError: Pool size is too low, the future was aborted.
(...stack trace...)
```

#### Prioritized futures
Instead of a queue, Futuroscope now uses something resembling a priority queue, where the priority is determined by how many threads are directly or indirectly blocking on the future's value. This can avoid deadlock situations, for example:
```
2.1.0 :001 > Futuroscope.default_pool = Futuroscope::Pool.new 2..2
=> #, @workers=#, @free=true, @thread=#>, #, @free=true, @thread=#>}>, @mutex=#, @futures={}>
2.1.0 :002 > f1, f2, f3, f4 = future { sleep 1; f4.future_value }, future { sleep 3 }, future { sleep 1; f4.future_value }, future { 4 }
=> [4, 3, 4, 4]
```
In the old implementation, this would raise a fatal error, because `f1` starts to block on f4 before it gets to the queue, then `f3` starts to block on `f4` as well after `f2` is done, so the pool gets full of blocking futures before `f4` has a chance to be evaluated.

#### Smart up/down spinning of workers
Previously, with the push of every new future, there was a 50% chance that a new worker gets spun up if the limit hasn't been reached yet. Now the workers keep track whether they are free or busy, and the pool only spins up a new worker if it has more futures without workers than free workers.

Also previously every thread immediately quit if it was over the minimum thread count. Now, if the pool has more workers than the minimum, the workers have a 2 second linger period in which they will pick up new work if it's available, and only if no new work came in will they die. This is to minimize the cost of spinning threads up and down when the work comes in close spikes.

This also leads to better parallelization in some cases. Old version:
```
2.1.2 :001 > require 'benchmark'
=> true
2.1.2 :002 > Futuroscope.default_pool = Futuroscope::Pool.new 1..10
=> #, @workers=#, @thread=#>}>, @mutex=#>
2.1.2 :004 > Benchmark.measure { (1..10).future_map { sleep 1 }.inspect }.real
=> 2.002055
```
New version:
```
2.1.0 :001 > require 'benchmark'
=> true
2.1.0 :002 > Futuroscope.default_pool = Futuroscope::Pool.new 1..10
=> #, @workers=#, @free=true, @thread=#>}>, @mutex=#, @futures={}>
2.1.0 :003 > Benchmark.measure { (1..10).future_map { sleep 1 }.inspect }.real
=> 1.002627
```

#### Logging
Futuroscope now supports logging. If you put loggers into `Futuroscope.loggers` (which is a simple array), they will get all log messages from inside. The loggers are expected to conform with Ruby's core `Logger`, in that they respond to these messages: `debug`, `info`, `warn`, `error`, `fatal`.

Example:
```
2.1.0 :001 > require 'logger'
=> true
2.1.0 :002 > Futuroscope.loggers << Logger.new(STDERR)
=> [#, @formatter=nil, @logdev=#>, @mutex=#>>>]
2.1.0 :003 > Futuroscope.default_pool = Futuroscope::Pool.new 1..1
I, [2014-09-29T16:50:17.214206 #85319] INFO -- : spun up worker with thread 2168719860
=> #, @workers=#, @free=true, @thread=#>}>, @mutex=#, @futures={}>
I, [2014-09-29T16:50:17.214845 #85319] INFO -- : POP: thread 2168719860 going to sleep until there's something to do...
2.1.0 :004 > f = future { f.future_value }
I, [2014-09-29T16:50:25.741545 #85319] INFO -- : PUSH: added future 2168704840
I, [2014-09-29T16:50:25.741620 #85319] INFO -- : sending signal to wake up a thread
D, [2014-09-29T16:50:25.741673 #85319] DEBUG -- : current priorities: {"future 2168704840"=>0}
I, [2014-09-29T16:50:25.741800 #85319] INFO -- : DEPEND: thread 2156059640 depends on future 2168704840
D, [2014-09-29T16:50:25.741861 #85319] DEBUG -- : current dependencies: {"thread 2156059640"=>"future 2168704840"}
I, [2014-09-29T16:50:25.741962 #85319] INFO -- : incrementing priority for future 2168704840
I, [2014-09-29T16:50:25.742025 #85319] INFO -- : POP: ... thread 2168719860 woke up
D, [2014-09-29T16:50:25.742069 #85319] DEBUG -- : current priorities: {"future 2168704840"=>1}
D, [2014-09-29T16:50:25.742110 #85319] DEBUG -- : current future workers: {"future 2168704840"=>nil}
I, [2014-09-29T16:50:25.742156 #85319] INFO -- : POP: thread 2168719860 will start working on future 2168704840
I, [2014-09-29T16:50:25.742203 #85319] INFO -- : DEPEND: thread 2168719860 depends on future 2168704840
D, [2014-09-29T16:50:25.742266 #85319] DEBUG -- : current dependencies: {"thread 2156059640"=>"future 2168704840", "thread 2168719860"=>"future 2168704840"}
E, [2014-09-29T16:50:25.742313 #85319] ERROR -- : deadlock! cyclical dependency, sending interrupt to all threads involved
I, [2014-09-29T16:50:25.742388 #85319] INFO -- : DONE: thread 2168719860 is done with future 2168704840
I, [2014-09-29T16:50:25.742423 #85319] INFO -- : deleting future 2168704840 from the task list
I, [2014-09-29T16:50:25.742462 #85319] INFO -- : deleting dependency from thread 2156059640 to future 2168704840
I, [2014-09-29T16:50:25.742498 #85319] INFO -- : deleting dependency from thread 2168719860 to future 2168704840
I, [2014-09-29T16:50:25.742548 #85319] INFO -- : POP: thread 2168719860 going to sleep until there's something to do...
Futuroscope::DeadlockError: Cyclical dependency detected, the future was aborted.
(...stack trace...)
```
---
I'd also like to note that this PR comes with multiple engineering decisions that are worth discussing. For example, I've moved away from using a `Queue` in both the pool and the future; instead, I'm now making use of `ConditionVariable`s (which `Queue` is using internally as well).

Another one is that the pool keeps track of the `__id__`s of the futures as hash keys. Why is that so, why am I using 2 hashes instead of one: one with `future.__id__`s pointing to priority values, and one with `future.__id__`s pointing to the actual futures? Because if you want to use a `Delegator` as a hash key, Ruby will call `#hash` on the object, which gets forwarded to the wrapped object, creating a deadlock. Not forwarding `#hash` is no solution: it will make the futures non-transparent, `hsh[:key]` will not be the same as `hsh[future { :key }]`.

Please discuss any parts that you disagree with, I'm happy to elaborate and of course it's always possible that I missed something.