How To Interleave Streams (with Backpressure)
Solution 1:
Here's a crazy chunk of code that might help.
It turns the input streams into a single stream of 'value' events, then merges them with 'send' events (and 'end' events for bookkeeping). Then, using a state machine, it builds up queues out of the 'value' events, and dispatches values on 'send' events.
Originally I wrote a roundRobinThrottle, but I've moved it to a gist.
Here is a roundRobinPromiseMap that is very similar. The code in the gist is tested, but this is not.
# roundRobinPromiseMap :: (a -> Promise b) -> [EventStream] -> EventStream
roundRobinPromiseMap = (promiser, streams) ->
# A bus to trigger new sends based on promise fulfillment
promiseFulfilled = new Bacon.Bus()
# Merge the input streams into a single, keyed stream
theStream = Bacon.mergeAll(streams.map((s, idx) ->
s.map((val) -> {
type: 'value'index: idx
value: val
})
))
# Merge in 'end' events
.merge(Bacon.mergeAll(streams.map((s) ->
s.mapEnd(-> {
type: 'end'
})
)))
# Merge in 'send' events that fire when the promise is fulfilled.
.merge(promiseFulfilled.map({ type: 'send' }))
# Feed into a state machine that keeps queues and only creates# output events on 'send' input events.
.withStateMachine(
{
queues: streams.map(-> [])
toPush: 0
ended: 0
}
handleState
)
# Feed this output to the promiser
theStream.onValue((value) ->
Bacon.fromPromise(promiser(value)).onValue(->
promiseFulfilled.push()
))
handleState = (state, baconEvent) ->
outEvents = []
if baconEvent.hasValue()
# Handle a round robin event of 'value', 'send', or 'end'
outEvents = handleRoundRobinEvent(state, baconEvent.value())
else
outEvents = [baconEvent]
[state, outEvents]
handleRoundRobinEvent = (state, rrEvent) ->
outEvents = []
# 'value' : push onto queueif rrEvent.type == 'value'
state.queues[rrEvent.index].push(rrEvent.value)
# 'send' : send the next value by round-robin selectionelseif rrEvent.type == 'send'# Here's a sentinel for empty queues
noValue = {}
nextValue = noValue
triedQueues = 0while nextValue == noValue && triedQueues < state.queues.length
if state.queues[state.toPush].length > 0
nextValue = state.queues[state.toPush].shift()
state.toPush = (state.toPush + 1) % state.queues.length
triedQueues++
if nextValue != noValue
outEvents.push(new Bacon.Next(nextValue))
# 'end': Keep track of ended streamselseif rrEvent.type == 'end'
state.ended++
# End the round-robin stream if all inputs have endedif roundRobinEnded(state)
outEvents.push(new Bacon.End())
outEvents
roundRobinEnded = (state) ->
emptyQueues = allEmpty(state.queues)
emptyQueues && state.ended == state.queues.length
allEmpty = (arrays) ->
for a in arrays
return false if a.length > 0return true
Solution 2:
The core challenge here was to understand, how to formalise fairness. In the question I already mentioned worker analogy. Turned out that the obvious fairness criteria is to pick a stream that generated less events than others, or taken even further: whom generated streams waited for less time.
After that it was quite trivial to formalise the desired output using denotational semantics: code is on GitHub
I didn't had time to develop the denotational combinators to include withStateMachine
from Bacon.js, so the next step was to reimplement it in JavaScript with Bacon.js directly. The whole runnable solution is available as a gist.
The idea is to make a state machine with
- per stream costs and queues as a state
- streams and additional feedback stream as inputs
As output of the whole system is feeded back, we can dequeue the next event when the previous flatMapped stream is ended.
For that I had to make a bit ugly rec
combinator
functionrec(f) {
var bus = newBacon.Bus();
var result = f(bus);
bus.plug(result);
return result;
}
It's type is (EventStream a -> EventStream a) -> EventStream a
- the type resembles other recursion combinators, e.g. fix
.
It can be made with better system-wide behaviour, as Bus breaks unsubscription propagation. We have to work on that.
The Second helper function is stateMachine
, which takes an array of streams and turns them into single state machine. Essentially it's .withStateMachine ∘ mergeAll ∘ zipWithIndex
.
functionstateMachine(inputs, initState, f) {
var mapped = inputs.map(function (input, i) {
return input.map(function (x) {
return [i, x];
})
});
returnBacon.mergeAll(mapped).withStateMachine(initState, function (state, p) {
if (p.hasValue()) {
p = p.value();
returnf(state, p[0], p[1]);
} else {
return [state, p];
}
});
}
Using this two helpers we can write a not-so-complex fair scheduler:
functionfairScheduler(streams, fn) {
var streamsCount = streams.length;
returnrec(function (res) {
returnstateMachine(append(streams, res), initialFairState(streamsCount), function (state, i, x) {
// console.log("FAIR: " + JSON.stringify(state), i, x);// END eventif (i == streamsCount && x.end) {
var additionalCost = newDate().getTime() - x.started;
// add cost to input stream cost centervar updatedState = _.extend({}, state, {
costs: updateArray(
state.costs,
x.idx, function (cost) { return cost + additionalCost; }),
});
if (state.queues.every(function (q) { return q.length === 0; })) {
// if queues are empty, set running: false and don't emit any eventsreturn [_.extend({}, updatedState, { running: false }), []];
} else {
// otherwise pick a stream with// - non-empty queue// - minimal costvar minQueueIdx = _.chain(state.queues)
.map(function (q, i) {
return [q, i];
})
.filter(function (p) {
return p[0].length !== 0;
})
.sortBy(function (p) {
return state.costs[p[1]];
})
.value()[0][1];
// emit an event from that streamreturn [
_.extend({}, updatedState, {
queues: updateArray(state.queues, minQueueIdx, function (q) { return q.slice(1); }),
running: true,
}),
[newBacon.Next({
value: state.queues[minQueueIdx][0],
idx: minQueueIdx,
})],
];
}
} elseif (i < streamsCount) {
// event from input streamif (state.running) {
// if worker is running, just enquee the eventreturn [
_.extend({}, state, {
queues: updateArray(state.queues, i, function (q) { return q .concat([x]); }),
}),
[],
];
} else {
// if worker isn't running, start it right awayreturn [
_.extend({}, state, {
running: true,
}),
[newBacon.Next({ value: x, idx: i})],
]
}
} else {
return [state, []];
}
})
.flatMapConcat(function (x) {
// map passed thru events,// and append special "end" eventreturnfn(x).concat(Bacon.once({
end: true,
idx: x.idx,
started: newDate().getTime(),
}));
});
})
.filter(function (x) {
// filter out END eventsreturn !x.end;
})
.map(".value"); // and return only value field
}
Rest of the code in the gist is quite straight-forward.
Post a Comment for "How To Interleave Streams (with Backpressure)"