A wait_for_all function should be implemented for df operators

I think a central control is contrary to the whole dataflow model, which is one of asynchronous distributed synchronous event processing by operators. It is possible to build a central control over the dataflow model and if there is a requirement for this then it can be programmed for that application: it should not be a property of the GPars framework.

Russell, I understand your concern that this seems a bit contrary to the dataflow model so more thought must be put in before adding this to the GPars framework. Let me provide some context for the original motivation. Consider the following situation where the program is not structured as an asychronous distributed system but a sequential system with some portion that would benefit from parallelism:

statement1;
statement2;

// Some expensive computation that is best expressed using dataflow
// Create the dataflow network
graph.operator(...)
graph.operator(...)
// Pass in some data and let it start computing until everything is done

// Continue execution only when the dataflow network is done
graph.waitForAll();
statement3;
statement4;

One could envision this as a form of forking a dataflow network and then joining when the graph is done. Here a waitForAll construct on the graph would be very convenient (where the keyword here is convenient). It is entirely possible to solve this problem without the waitForAll construct by using some other form of a synchronization point like waiting on a DataflowVariable, Latch, Barrier or Phaser. However, being able to just rely on the graph to terminate when there is nothing else left to do (e.g, no more messages and all processors are non-active) seems like a nice feature.

Are there other nice ways to solve this fork/join problem without having the user explicitly code the terminate/wait condition? Or is this use case not a fit for the dataflow model within GPars?

I think you have actually supplied the answer to your problem in the above, your architecture is of a dataflow computation embedded in a fork/join driver. Java (and hence GPars has, arguably, the best fork/join framework of any language, so I would embed your dataflow operator evaluations in the above fragment as forks, then the "wait for all" is integral to the driver code. An application doesn't have to be actor or dataflow or fork/join, it can be any sensible mixture of all of them.

Unless I am missing something of your problem...

I suspect the code fragment is confusing. Let me try to motivate it better

It is not a fork/join problem with the usual recursive decomposition. The graph.operator(...) does not exists in GPars currently. It's an API modeled after TBB's flow graph. It does not "fork" things. It creates new DataflowOperators that are part of the dataflow network.

I'm dealing with a dataflow problem like a pipeline. Here is a more complete example.

final DataflowChannel channel1 = new DataflowQueue()
final DataflowChannel channel2 = new DataflowQueue()
final DataflowChannel channel3 = new DataflowQueue()
final List<Integer> outputs = new ArrayList<Integer>()

FlowGraph fGraph = new FlowGraph()

// Set up dataflow network

fGraph.operator([channel1], [channel2]) { input ->
    bindOutput input * input
}

fGraph.operator([channel2], [channel3]) { input ->
    bindOutput input + 1
}

fGraph.operator([channel3], []) { input ->
    outputs.add(input)
}

// Pump data in.

channel1 << 1
channel1 << 2
channel1 << 3
channel1 << 4

fGraph.waitForAll()

// Proceed when fGraph is done

In such a case, how would one wait for the pipeline to finish using the current API in GPars? From my understanding, it would require setting up some synchronization point (wait on the last operator) or using an explicit PoisonPill or CountingPoisonPill. However, these solutions are very specific to the particular dataflow network: one needs to determine what is the last operator or one would need to determine how many counts for the CountingPoisonPill. This seems brittle to me because anytime your dataflow network changes (e.g., need to add another stage), you need to explicitly remember to change your definition of last operator or counts for the CountingPoisonPill.

Thoughts?

I feel our Poison message may serve well the purpose, if we combine it with a few new concepts. Here's an idea:

1. We need lifecycle events fired from operators. Detecting shutdown is then just a matter of listening for "onStop" event on all operators and counting the number of alive ones.
2. Producers each send Poison once they are done sending useful data.
3. If multiple producers share an output channel, that channel needs a ProducerCoordination cap to count registered producers and propagate Poison only after all the registered producers unregister/close. I'm reluctant to add a producer coordination mechanism to the standard channel API.
4. We need a flag on channels indicating whether or not to filter "control" messages, such as Poison. Most likely we want to stop these messages on the graph boundaries.
5. Operators shutdown as soon as they see Poison on one of their inputs.
6. Selectors shutdown as soon as they gradually collect Poison messages from all their inputs.
7. Selectors need to detect loops in the network. If some of the operator's input is coming from behind the selector, the selector must not expect a Poison message to ever come from that channel. An automated detection of such loops will most likely require another type of "control" message.

Vaclav, I just remembered that using {{PoisonPill} from Java with parameterized channels has a potential limitation:

DataflowQueue<Integer> channel = new DataflowQueue<Integer>();
channel.bind(1); // No problem
channel.bind(PoisonPill.getInstance()); // this is a type error

So to use this from Java you would need to forgo the convenience of the type parameter.

Nick, thanks for your comments. I'll take a look at your implementation (busy times over here, so I might need some time, unfortunately). I'm sure it will help me clarify my thoughts about the matter.
In fact, dataflow operators is the last remaining big piece of functionality that we have to finish before we can start polishing our way towards the 1.0 release.

Now, with operator lifecycle events in place I'm experimenting with a more generic way to monitor the state of operators and their input channels. Since all the monitoring would be taken care of by plain listeners, we could use this graceful shutdown mechanism to all networks, even if the operators have been created by the usual factory mehods.

http://gpars.org/SNAPSHOT/guide/guide/dataflow.html#dataflow_operators_shutdown