worker_runtime: Add a producer-consumer structure.

The runtime used to tell the worker that there is no more inputs in all of the
cases.  In the new model, the runtime will tell the worker that there is more
inputs in case there is more inputs to come.

At the moment, the runtime has all of the inputs when the task starts but this
will change with the concurrent stages.