Full provenance of tasks executions

[pditommaso]

Problem

The execution DAG created by Nextflow tracks the dependencies between processes, operators and any combination of them.

However, Nextflow is not able to track which is the task instance(s) that triggered a specific task execution.

This means that it's possible to know, for example, that a process A is connected with a process B e.g. A -> B, however, each of them can run n tasks and it's not possible to establish which task i-th of A trigger the task j-th of B.

This information is important to be able to precisely track the provenance of the task executions and solve other problems like the cleanup at runtime of the temporary files created by each task #452

Heuristic solution

A possible solution to this problem could consist in using the unique hash id associated with each task execution.

The task hash is used to allocate a scratch work directory where all output files are created. For example, a task having id 9a001d8539def2552dd604427ec9aee1 will create all output files into a directory having the path /some/work/dir/9a/001d8539def2552dd604427ec9aee1.

Therefore all downstream tasks receiving these outputs as input files could use the hash encoded in the file path to infer the task instance that created those file are re-create the tasks execution DAG.

However this approach does not provide a complete solution because Nextflow processes can produce arbitrary values are outputs other than file paths, that would make not possible to use this approach.

A similar problem can arise when a process execution is chained with an operator that can alter the file output path with an arbitrary path (e.g. collectFile) breaking the above assumption.

Object identity solution

This solution to some extent is similar to the previous approach, however, the main idea is to use the Java object identity associated to each input and output to track the relationship of the tasks.

This could be done by storing in the dictionary structure the pair < object id, task id>, where object id is taken using the identity hash code of the n-th output object, and task id is the nextflow TaskRun.id attribute. Let's call this structure prov-map

This information can be captured when the output of a task Tx is bound to the corresponding output channel, see here.

Correspondingly, when a new task execution Ty is triggered, for each value in the list of inputs should be looked up in the prov-map using the input object system identity as the access key.

If an entry is found a relationship between the two tasks can be established and a direct edge Tx -> Ty can be added to the provenance graph.

The problem still remains for processes interleaved with operator execution e.g. P1 -> map -> P2

A similar approach could be takin into consideration that all Nextflow operators are implemented as a DataflowProcessor class.

This class allows the use of a listener interface defined as show below:

public interface DataflowEventListener {

    /**
     * Invoked immediately after the listener has been attached to a dataflow processor.
     *
     * @param processor The reporting dataflow operator/selector
     */
    void registered(DataflowProcessor processor);

    /**
     * Invoked immediately after the operator starts by a pooled thread before the first message is obtained
     *
     * @param processor The reporting dataflow operator/selector
     */
    void afterStart(DataflowProcessor processor);

    /**
     * Invoked immediately after the operator terminates
     *
     * @param processor The reporting dataflow operator/selector
     */
    void afterStop(DataflowProcessor processor);

    /**
     * Invoked if an exception occurs.
     * If any of the listeners returns true, the operator will terminate.
     * Exceptions outside of the operator's body or listeners' messageSentOut() handlers will terminate the operator irrespective of the listeners' votes.
     * When using maxForks, the method may be invoked from threads running the forks.
     *
     * @param processor The reporting dataflow operator/selector
     * @param e         The thrown exception
     * @return True, if the operator should terminate in response to the exception, false otherwise.
     */
    boolean onException(DataflowProcessor processor, Throwable e);

    /**
     * Invoked when a message becomes available in an input channel.
     *
     * @param processor The reporting dataflow operator/selector
     * @param channel   The input channel holding the message
     * @param index     The index of the input channel within the operator
     * @param message   The incoming message
     * @return The original message or a message that should be used instead
     */
    Object messageArrived(DataflowProcessor processor, DataflowReadChannel<Object> channel, int index, Object message);

    /**
     * Invoked when a control message (instances of ControlMessage) becomes available in an input channel.
     *
     * @param processor The reporting dataflow operator/selector
     * @param channel   The input channel holding the message
     * @param index     The index of the input channel within the operator
     * @param message   The incoming message
     * @return The original message or a message that should be used instead
     */
    Object controlMessageArrived(DataflowProcessor processor, DataflowReadChannel<Object> channel, int index, Object message);

    /**
     * Invoked when a message is being bound to an output channel.
     * When using maxForks, the method may be invoked from threads running the forks.
     *
     * @param processor The reporting dataflow operator/selector
     * @param channel   The output channel to send the message to
     * @param index     The index of the output channel within the operator
     * @param message   The message to send
     * @return The original message or a message that should be used instead
     */
    Object messageSentOut(DataflowProcessor processor, DataflowWriteChannel<Object> channel, int index, Object message);

    /**
     * Invoked when all messages required to trigger the operator become available in the input channels.
     *
     * @param processor The reporting dataflow operator/selector
     * @param messages  The incoming messages
     * @return The original list of messages or a modified/new list of messages that should be used instead
     */
    List<Object> beforeRun(DataflowProcessor processor, List<Object> messages);

    /**
     * Invoked when the operator completes a single run
     * When using maxForks, the method may be invoked from threads running the forks.
     *
     * @param processor The reporting dataflow operator/selector
     * @param messages  The incoming messages that have been processed
     */
    void afterRun(DataflowProcessor processor, List<Object> messages);

    /**
     * Invoked when the fireCustomEvent() method is triggered manually on a dataflow operator/selector
     * When using maxForks, the method may be invoked from threads running the forks.
     *
     * @param processor The reporting dataflow operator/selector
     * @param data      The custom piece of data provided as part of the event
     * @return A value to return from the fireCustomEvent() method to the caller (event initiator)
     */
    Object customEvent(DataflowProcessor processor, Object data);
}

The methods beforeRun and afterRun can be used to propagate the task association the corresponding output value. For example let's consider the flow T1(x) -> map(x) -> T2(y)

1. The task T1 store the pair < x, T1 > in the prov-map.
2. The operator map receives the value x and invokes the beforeRun even listener
3. The even lister looks up the value T1 using the prov-map
4. the event listener afterRun is invoked and the pair < y, T1 > is stored into prov-map
5. The task T2 receives the value y and fetch the value T1 from the prov-map
6. The provenance edge T1 -> T2 is created