References

module DataFlowTasks

Create Tasks which can keep track of how data flows through it.

@enum AccessMode READ WRITE READWRITE

Describe how a DataFlowTask access its data.

struct CleanupLog

Logs the removal of a finished DataFlowTask from its DAG by the dag_cleaner.

Fields:

• tag : task id in DAG
• time_start : time the removal began (the task waits to be removed since it finished, see TaskLog)
• time_finish : time the removal finished

All times are in nanoseconds.

struct DAG{T}

Representation of a directed acyclic graph containing nodes of type T. The list of nodes with edges coming into a node i can be retrieved using inneighbors(dag,i); similarly, the list of nodes with edges leaving from i can be retrieved using outneighbors(dag,i).

DAG is a buffered structure with a buffer of size sz_max: calling addnode! on it will block if the DAG already holds sz_max nodes.

The DAG is thread-safe: mutating operations acquire dag.lock, and the conditions cond_push and cond_empty are associated with that same lock.

DAG{T}(sz)

Create a buffered DAG holding a maximum of sz nodes of type T.

DataFlowTask(func,data,mode)

Create a task-like object similar to Task(func) which accesses data with AccessMode mode.

When a DataFlowTask is created, the elements in its data field will be checked against all other active DataFlowTask to determine if a dependency is present based on a data-flow analysis. The resulting Task will then wait on those dependencies.

A DataFlowTask behaves much like a Julia Task: you can call wait(t), schedule(t) and fetch(t) on it.

See also: @dtask, @dspawn.

struct FinishedChannel{T} <: AbstractChannel{T}

Used to store tasks which have been completed, but not yet removed from the underlying DAG. Taking from an empty FinishedChannel will block.

struct InsertionLog

Logs the execution trace of a DataFlowTask insertion.

Fields:

• time_request : time the insertion was requested, before waiting for the DAG to be available (i.e. not locked, and not at full capacity)
• time_start : time the insertion began, i.e. the data-flow analysis
• time_finish : time the insertion finished
• gc_time : time spent in garbage collection during the insertion
• taskid : the task it is inserting
• tid : the thread on which the insertion is happening

All times are in nanoseconds.

mutable struct LogInfo

Contains information on the program's progress. Events are appended to the logs under lock, so that a LogInfo can be safely shared by all threads.

Besides the logs of the tasks, of their insertion and of their removal from the DAG, a LogInfo records the logging window (time_start and time_finish, in nanoseconds), the number of threads available to run the tasks (nworkers), and the time spent in garbage collection and compilation during the window (gc_time and compile_time, in seconds).

Use DataFlowTasks.describe or DataFlowTasks.summarize to analyze a LogInfo, and GraphViz.Graph or Makie.plot to visualize it.

struct LogSummary

Summary of the events recorded in a LogInfo, as computed by DataFlowTasks.summarize. All times are in seconds.

Fields:

• elapsed : wall-clock time of the logging window
• workers : number of threads available to run the tasks (the threads of the :default threadpool)
• running : total time the tasks spent running
• blocked : total time the tasks spent descheduled during their body (sleeping, waiting on I/O or on a lock...)
• insertion : total time spent inserting the tasks in the DAG (on the threads calling @dspawn)
• longest_path_length : length (in wall time) of the longest chain of dependent tasks
• gc, compile : time spent in garbage collection and in compilation during the logging window
• categories, running_per_category : the categories of tasks, and the running time of the tasks in each of them (the last entry gathers the tasks matching no category)
• longest_path : tags of the tasks on the longest path, from the last one to the first one
• ntasks, nfailed, nmigrated : number of tasks, of tasks which threw an error, and of tasks which finished on another thread than the one they started on

The following quantities are derived from these fields:

• capacity(s) = s.workers * s.elapsed: the time the workers were available;
• idle(s) = capacity(s) - s.running: the time the workers did not run any task;
• nowait(s) = s.running / s.workers: the elapsed time if the running time were perfectly shared between the workers, without ever waiting.

Both longest_path_length and nowait are lower bounds of elapsed.

struct Stop

Singleton type used to safely interrupt a task reading from an AbstractChannel.

struct TaskGraph

A directed acyclic graph used to represent the dependencies between DataFlowTasks.

TaskGraph(sz) creates a task graph that can hold at most sz elements at any given time. In particular, trying to add a new DataFlowTask will block if the TaskGraph is already full.

See also: get_active_taskgraph, set_active_taskgraph!

struct TaskLog

Logs the execution trace of a DataFlowTask.

Fields:

• tag : task id in DAG
• time_start : time the task body started running (after waiting for its dependencies)
• time_finish : time the task body finished running
• running_time : time the task body spent actually running on a thread, i.e. excluding the time it was descheduled (sleeping, waiting on I/O or locks...)
• tid : thread on which the task body started
• tid_finish : thread on which the task body finished (a task may migrate between threads when it is descheduled)
• inneighbors : vector of incoming neighbors in DAG
• label : a string used for displaying and/or postprocessing tasks
• failed : whether the task threw an error

All times are in nanoseconds.

struct TaskTimeline

Lifecycle of a logged DataFlowTask, as computed by DataFlowTasks.timeline. Times are in seconds since the start of the logging window.

Fields:

• tag : task id in DAG
• label : label of the task
• requested : the insertion of the task in the DAG was requested
• insertion_start : the data-flow analysis began, once the DAG was available
• inserted : the task was inserted in the DAG
• ready : the task was inserted and all its predecessors finished
• started : the task body started running
• finished : the task body finished running
• removed : the task was removed from the DAG, or nothing if it was not removed during the logging window
• running : time the task body spent running on a thread (see TaskLog)
• predecessors : tags of the logged tasks the task depends on
• ready_by : tag of the predecessor which finished last, if it made the task ready (i.e. after its insertion), and 0 otherwise
• lane : lane of the task in a trace of the tasks, such that the tasks of a lane do not overlap in time. Lanes are not threads: a task may migrate between threads, and a thread runs other tasks while a task is descheduled.
• migrated : whether the task finished on another thread than the one it started on
• failed : whether the task threw an error

empty!(tg::TaskGraph)

Remove all the tasks from tg, so that the tasks spawned afterwards do not depend on them.

This function is useful when tasks in tg are blocked (e.g. waiting for a resource which is never released), to avoid having to restart the REPL. The removed tasks are not interrupted: those which are running or waiting to run keep doing so.

resize!(tg::TaskGraph, sz)

Change the buffer size of tg to sz.

schedule(t::DataFlowTask)

Schedule t to run on the :default threadpool (as @dspawn does), and return its Julia Task.

wait(tg::TaskGraph)

Wait for all nodes in tg to be finished before continuing.

To wait on the active TaskGraph, use wait(get_active_taskgraph()).

access_mode(t::DataFlowTask[,i])

How t accesses its data.

See: AccessMode

addedge!(dag,i,j)

Add (directed) edge connecting node i to node j in the dag.

addnode!(dag,(k,v)::Pair[, check=false, logger=nothing, time_request=0])
addnode!(dag,k[, check=false, logger=nothing, time_request=0])

Add a node to the dag. If passed only a key k, the value v is initialized as empty (no edges added). The check flag is used to indicate if a data flow analysis should be performed to update the dependencies of the newly inserted node.

If a logger is given, the insertion is recorded in it as an InsertionLog, requested at time_request (by default, when the insertion begins).

capacity(dag)

The maximum number of nodes that dag can contain.

concurrency(l::LogInfo) -> (; time, dag, ready, running)

Number of logged tasks over time, as step functions: on [time[k], time[k+1]),

• dag[k] tasks are in the DAG (inserted and not yet removed),
• ready[k] tasks are ready but not started (see TaskTimeline),
• running[k] tasks are running, i.e. started and not finished (a running task may be descheduled, and leave its thread to another task).

Times are in seconds since the start of the logging window, from 0 to the duration of the window. Tasks which were not removed from the DAG during the logging window are counted in the DAG until its end.

data(t::DataFlowTask[,i])

Data accessed by t.

data_dependency(t1::DataFlowTask,t2::DataFlowTask)

Determines if there is a data dependency between t1 and t2 based on the data they read from and write to.

describe(loginfo::LogInfo; categories = String[])
describe(io, loginfo::LogInfo; categories = String[])

Analyses the information contained in loginfo and displays a summary on io (stdout by default). See LogSummary for the definitions of the reported quantities.

Passing a categories argument allows grouping tasks by category. The categories can be a vector of Strings or a vector of String => Regex pairs, which will be matched against the tasks' labels.

enable_debug(mode = true)

If mode is true (the default), enable debug mode: errors inside tasks will be shown.

enable_log(mode = true)

If mode is true (the default), logging is enabled through the @log macro. Calling enable_log(false) will de-activate logging at compile time to avoid any possible overhead.

Note that changing the log mode at runtime may invalidate code, possibly triggering recompilation.

See also: @log, with_logging

force_linear_dag(mode=false)

If mode is true, nodes are added to the DAG in a linear fashion, i.e. the DAG connects node i to node i+1. This is useful for debugging purposes.

force_sequential(seq = true; static = false)

If seq is true, enable sequential mode: no tasks are created and scheduled, code is simply run as it appears in the sources. In effect, @dspawn expr evaluates expr and returns its value instead of a Task (fetch still works on it, but wait does not).

By default, sequential mode is disabled when the program starts.

The mode is checked each time a task is created. With static = true, it is instead fixed when the @dspawn macros are expanded (which avoids checking it, but requires reloading the code to change it).

See also: force_linear_dag.

get_active_taskgraph()

Return the active TaskGraph: the one of the innermost enclosing with_taskgraph block (a task created inside such a block is also considered inside it), or the default one (see set_active_taskgraph!).

inneighbors(dag,i)

List of predecessors of i in dag.

loggertodot(logger)  --> dagstring

Return a string in the DOT format representing the underlying graph in logger.

If GraphViz is installed, you can use GraphViz.Graph(logger) to produce an image.

memory_overlap(di,dj)

Determine if data di and dj have overlapping memory in the sense that mutating di can change dj (or vice versa). This function is used to build the dependency graph between DataFlowTasks.

A generic version is implemented: it returns false if di or dj is a bits value (which cannot be mutated), and compares Tasks by identity. Otherwise, it returns true and prints a warning. Users should overload this function for the specific data types used in the arguments to allow for appropriate inference of data dependencies.

memory_overlap(di::AbstractArray,dj::AbstractArray)

Try to determine if the arrays di and dj have overlapping memory.

When both di and dj are Arrays of bitstype, simply compare their addresses. Otherwise, compare their parents by default.

When both di and dj are SubArrays we compare the actual indices of the SubArrays when their parents are the same (to avoid too many false positives).

nodes(dag::DAG)

Return an iterator over the nodes of dag.

num_edges(dag::DAG)

Number of edges in the DAG.

num_nodes(dag::DAG)

Number of nodes in the DAG.

outneighbors(dag,i)

List of successors of i in dag.

realized_critical_path(l::LogInfo) -> Vector{Int64}

Tags of the chain of logged tasks which determined when the last task finished, in execution order.

Starting from the last task to finish, the chain goes back to the predecessor which made the task ready (see ready_by in TaskTimeline), until a task made ready by its insertion. Along the chain, the time is spent either waiting for the first task to be inserted, or between the moment a task is ready and the moment it starts (scheduling latency), or running the tasks: these segments cover the time from the start of the logging window to the end of the last task.

This is the path that was actually followed during the execution, which differs from the longest path of DataFlowTasks.summarize: the longest chain of dependent tasks, a lower bound of the elapsed time.

remove_node!(dag::DAG,i[, logger=nothing])

Remove node i and all of its edges from dag. If a logger is given, the removal is recorded in it as a CleanupLog.

DataFlowTasks.savedag(filepath, graph)

Save graph as an SVG image at filepath. This requires GraphViz to be available.

set_active_taskgraph!(tg)

Set the default TaskGraph to tg. The default TaskGraph is used outside of with_taskgraph blocks.

start_dag_cleaner(tg)

Start a task associated with tg which takes nodes from its finished queue and removes them from the dag. The task blocks if finished is empty.

summarize(loginfo::LogInfo; categories = String[]) -> LogSummary

Compute a LogSummary of the events recorded in loginfo.

Passing a categories argument allows grouping tasks by category. The categories can be a vector of Strings or a vector of String => Regex pairs, which will be matched against the tasks' labels.

timeline(l::LogInfo) -> Vector{TaskTimeline}

Lifecycle of the tasks logged in l, sorted by tag. See TaskTimeline.

The ready time of a task is derived from the logs: it is the latest of its insertion and of the end of its predecessors (in the complete dependency graph of the logged tasks).

traceplot(loginfo::LogInfo; categories, highlight, realized_path)
traceplot!(ax, loginfo::LogInfo; ...)

Plot the trace of the tasks logged in loginfo, packed in lanes. This is the top panel of Makie.plot(loginfo).

This function requires Makie; see its documentation in the Makie extension for more details.

traceplot(loginfo::LogInfo; categories, highlight, realized_path)
traceplot!(ax, loginfo::LogInfo; ...)

Plot the trace of the tasks logged in loginfo, packed in lanes. This is the top panel of Makie.plot(loginfo).

This function requires Makie; see its documentation in the Makie extension for more details.

update_edges!(dag::DAG, i, linear = _linear_dag())

Perform the data-flow analysis to add the edges coming into node i from the nodes inserted before it. If linear is true, i only depends on the previous node (see force_linear_dag).

with_logging!(f,l::LogInfo)

Similar to with_logging, but append events to l.

When l is used for several calls, its logging window spans from the start of the first call to the end of the last one: the time between the calls is counted in the elapsed time, during which the workers are reported as idle.

with_logging(f) --> f(),loginfo

Execute f() and log DataFlowTasks into the loginfo object.

Examples:

using DataFlowTasks

A,B = zeros(2), ones(2);

out,loginfo = DataFlowTasks.with_logging() do
    @dspawn fill!(@W(A),1)
    @dspawn fill!(@W(B),1)
    res = @dspawn sum(@R(A)) + sum(@R(B))
    fetch(res)
end

#

out

See also: LogInfo

with_taskgraph(f,tg::TaskGraph)

Run f, but push the DataFlowTasks created by f to tg. This includes the DataFlowTasks created by the tasks that f creates, but other tasks running concurrently are not affected.

@dspawn expr [kwargs...]

Spawn a Julia Task to execute the code given by expr, and schedule it to run on the :default threadpool.

Annotate the code in expr with @R, @W and/or @RW (or equivalently @←, @→ and @↔) to indicate how it accesses data (see examples below). This information is used to automatically infer task dependencies.

Additionally, the following keyword arguments can be provided:

• label: provide a label to identify the task. This is useful when logging scheduling information;
• priority: an integer in 0:65535 (0 by default) setting the priority of the underlying Julia Task. As in Julia, 0 is the most urgent: when several tasks are waiting for a thread, the ones with a smaller priority value tend to be started first. Priorities do not interrupt running tasks, are only approximately respected, and have no effect when the :default threadpool has a single thread.

Semantics

• The task is inserted in the active TaskGraph (see get_active_taskgraph) before @dspawn returns. Its dependencies are inferred from the tasks already in the TaskGraph, so that they follow the order of the @dspawn calls (see nested tasks for the case of tasks spawned concurrently).
• If the TaskGraph is full, @dspawn blocks until a task finishes (see resize!).
• The task runs on the :default threadpool, once all the tasks it depends on have finished. If one of them failed, the task fails as well.
• @dspawn returns the Julia Task running the DataFlowTask, which can be waited on or fetched. To access the DataFlowTask itself, create it with @dtask and run it with schedule.

Examples:

Below are 3 equivalent ways to create the same Task, which expresses a Read-Write dependency on C and Read dependencies on A and B

using LinearAlgebra
using DataFlowTasks
A = ones(5, 5)
B = ones(5, 5)
C = zeros(5, 5)
α, β = (1, 0)

# Option 1: annotate arguments in a function call
@dspawn mul!(@RW(C), @R(A), @R(B), α, β)

# Option 2: specify data access modes in the code block
@dspawn begin
   @RW C
   @R  A B
   mul!(C, A, B, α, β)
end

# Option 3: specify data access modes after the code block
# (i.e. alongside keyword arguments)
res = @dspawn mul!(C, A, B, α, β) @RW(C) @R(A,B)

fetch(res) # a 5×5 matrix of 5.0

Here is a more complete example, demonstrating a full computation involving 2 different tasks.

using DataFlowTasks

A = rand(5)

# create a task with WRITE access mode to A
# and label "writer"
t1 = @dspawn begin
    @W A
    sleep(1)
    fill!(A,0)
    println("finished writing")
end  label="writer"

# create a task with READ access mode to A
t2 = @dspawn begin
    @R A
    println("I automatically wait for `t1` to finish")
    sum(A)
end  priority=1

fetch(t2) # 0

# output

finished writing
I automatically wait for `t1` to finish
0.0

Note that in the example above t2 waited for t1 because it read a data field that t1 accessed in a writable manner.

@dtask expr [kwargs...]

Create a DataFlowTask to execute expr, where data have been tagged to specify how they are accessed. The task is inserted in the active TaskGraph, but not scheduled: schedule(t) runs it on the :default threadpool, as @dspawn does. Unlike @dspawn, which returns the Julia Task, this gives access to the DataFlowTask itself (e.g. its label or priority).

See @dspawn for information on how to annotate expr to specify data dependencies, and a list of supported keyword arguments.

See also: @dspawn

DataFlowTasks.@log expr --> LogInfo

Execute expr and return a LogInfo instance with the recorded events. The DataFlowTasks created by expr (directly, or by the tasks it creates) are logged, and @log waits for them to be completed before returning. A task created by expr but scheduled after @log returns (see @dtask) is not logged.

Tasks which are not logged (e.g. spawned before @log, or by other code running concurrently) may still run during the logging window: the time the workers spend on them is reported as idle. For clean measurements, wait for such tasks before logging, e.g. with wait(get_active_taskgraph()).

See also: with_logging, with_logging!

GraphViz.Graph(log_info::LogInfo)

Produce a GraphViz.Graph representing the DAG of tasks collected in log_info.

See also: DataFlowTasks.@log

TracePlot is the plot type associated with plotting function traceplot. Check the docstring for traceplot for further information.

traceplot(loginfo; categories, highlight, realized_path)

Plot the trace of the tasks logged in loginfo (see DataFlowTasks.timeline): each task is a bar, from the moment its body started to the moment it finished. Bars are packed in lanes such that the bars of a lane do not overlap. A lane is not a thread, since a task may migrate between threads.

• Bars are colored by category (see DataFlowTasks.summarize), and filled up to the fraction of their duration the tasks were running, like a gauge: a task which only sleeps is an empty (light) bar. Failed tasks are outlined in red.
• The insertions in the DAG are shown in lane 0: the time spent waiting for the DAG (in light red), in the garbage collector (in yellow), and analysing the data dependencies (in red).
• If realized_path is true, the tasks of the realized critical path (see DataFlowTasks.realized_critical_path) are outlined in black.
• The task with tag highlight (if any) is outlined in purple, its predecessors in blue and its successors in orange, and the predecessor which made it ready is linked to it by a dashed line.

Plot type

The plot type alias for the traceplot function is TracePlot.

Attributes

categories = String[] — Categories of tasks, given as in DataFlowTasks.summarize.

highlight = 0 — Tag of the task to highlight with its dependencies, or 0 for none.

realized_path = false — Whether to outline the realized critical path.

traceplot! is the mutating variant of plotting function traceplot. Check the docstring for traceplot for further information.

plot(log_info; categories)

Plot a dashboard of log_info: the trace of the tasks (see DataFlowTasks.traceplot), the time spent by activity, the elapsed time and its bounds, and the running time by category.

Entries in categories define how to group tasks in categories for plotting. Each entry can be:

• a String: in this case, all tasks having labels in which the string occurs are grouped together. The string is also used as a label for the category itself.
• a String => Regex pair: in this case, all tasks having labels matching the regex are grouped together. The string is used as a label for the category itself.

See the documentation for more information on how to profile and visualize DataFlowTasks.