Items totally unbalanced between hot, synchronous consumers

[acrylic-origami]

In the case that hot consumers are synchronous, the first hot consumer will get all the values and the rest won't get any. This is because internal to Internal\Producer::advance, hot consumers implicitly race to get control from the Internal\Producer::$waiting promise. However, promises handle multiple subscribers via a queue, not randomly. So, when the first consumer gets control and handles the item synchronously, it will continue to receive control first in this race and consume all the elements, starving the rest of the consumers.

See for yourself:

<?php
use AmpReactor\InteractiveProducer;
use Amp\Promise;
use Amp\Coroutine;
use function Amp\Promise\all;
use function Amp\Promise\wait;
$source = InteractiveProducer::from_producerish(function($emitter) { for($i = 0; $i < 10; $i++) $emitter($i); if(false) yield null; });
wait(all([
  new Coroutine((function() use ($source) { while(yield $source->advance()) echo 'Consumer 1: '.$source->getCurrent()."\n"; })()),
  new Coroutine((function() use ($source) { while(yield $source->advance()) echo 'Consumer 2: '.$source->getCurrent()."\n"; })())
]));

[acrylic-origami]

While this is getting patched in the backend, consumers can improve balancing themselves. If one know the vague regime of items remaining and the number of consumers at a given point in time, one can yield AmpReactor\Util\defer() with any probability higher than num_consumers/num_items for num_consumers < num_items. This should improve balance substantially.

When consuming infinite streams, any probability of deferral above 0 will automatically balance consumers — the question is just how slowly.