@myronmarston first let me clarify that, although the producer_consumer sends demand as soon as handle_events is done, those events are not consumed until demand is received from downstream. This is a form of pre-fetching to ensure we always have data in flux.

That said, the issue with your second design is that you no longer have back-pressure all the way. It won’t be a problem if the ShardWorker is the slowest layer in your pipeline but it also means you are not gaining anything by having two layers of GenStage.

For example, instead of a second pipeline, you could directly start the shard worker children directly in a supervisor:

          /[consumer]\
[producer]-[consumer]-[supervisor]
          \[consumer]/

It would work similarly to what you have designed: the shard worker will start multiple children in the supervisor and wait for those children to reply back.

However, my preferred solution would be to simply not do any asynchronous work in the stages. The question is: does the stale shard id producer provides enough events to make all shard workers busy, using 100% of your machine resources without the need to start subworkers?

Imagine the stale shard id producer can provide events faster than they can be consumed. In this case, you have enough work on each shard worker to use all cores without needing to break each shard worker in a subtask worker. In this scenario, if you have 4 cores and 4 shard workers, that will be enough. Breaking it into smaller tasks won’t buy you anything because all of the tasks will still be working towards the same 4 shard workers.

However, this may not be the case. You may process events faster than the producer can emit them. Or maybe the subtask workers are IO bound. In this case, you can also keep with a single stale shard id and multiple subworkers except you start multiple tasks (Task.async or Task.Supervisor.async) and await for them inside handle_events/3 for each subworker. This way you keep the synchronicity and still can break the work apart.

I would say that GenStage, as a library, only provides a unidirectional flow, but ultimately, each stage is a process, and can of course do normal Elixir/Erlang stuff like send messages. While it’s perhaps a bit non-standard to send results back, there are certainly valid reasons for doing so. (As I’ll explain below).

While I can see why you would think this, the ShardWorker stage does in fact wait on the sub task workers. In my prototype, it did so by calling this, as I showed in my answer to @josevalim above:

ShardProcess.perform_build(stale_shard_id)

ShardProcess here is a gen server, so this did the following:

• Started the gen server
• Waited for the gen server to completely finish building the shard and exit

The ShardProcess was responsible for sending sub tasks into the worker pool, keeping track of pending tasks, and receiving results until it got all results back–at which point it would persist the shard and exit.

Such a design would probably work, and might be conceptually simpler, but I don’t think it’s the one we will go with. The shard data structure we are building can get quite large (in the worst cases, multiple GB) and as such, we really only want it to exist in one process. Including it in a message to another process would involve copying the entire data structure, which would be quite slow.

In addition, we have designed the shard so that individual pieces (such as individual values in a map within the shard data structure) can be compressed individually. We get a pretty good compression ratio (about 20:1) so by having each sub task send its results back to the ShardProcess, it allows the ShardProcess to compress the results, and put it in the shard as sub tasks complete, which works out nicely to keep things from using more memory than needed. I don’t think we could get the same kind of benefits out of the design you have proposed.