go-taskflow

go-taskflow

go-taskflow is a general-purpose task-parallel programming framework for Go, inspired by taskflow-cpp. It leverages Go's native capabilities and simplicity, making it ideal for managing complex dependencies in concurrent tasks.

Features

• High Extensibility: Easily extend the framework to adapt to various specific use cases.

• Native Go Concurrency Model: Leverages Go's goroutines for efficient concurrent task execution.

• User-Friendly Programming Interface: Simplifies complex task dependency management in Go.

• Static, Subflow, Conditional, and Cyclic Tasking: Define static tasks, condition nodes, nested subflows, and cyclic flows to enhance modularity and programmability.

  Static	Subflow	Condition	Cyclic

• Priority Task Scheduling: Assign task priorities to ensure higher-priority tasks are executed first.

• Built-in Visualization, Profiling and Tracing: Generate visual representations of tasks, profile task execution with flamegraph, and capture Chrome Trace events for timeline analysis.

Use Cases

• Data Pipelines: Orchestrate data processing stages with complex dependencies.
• AI Agent Workflow Automation: Define and execute AI agent workflows with clear sequences and dependency structures.
• Parallel Graph Tasking: Execute graph-based tasks concurrently to maximize CPU utilization.

Installation

Import the latest version of go-taskflow using:

go get -u github.com/noneback/go-taskflow

Documentation

DeepWiki Page

Quick Start

A MapReduce word-count pipeline with parallel mappers, hash-partitioned shuffle, and reducers:

…

For more examples, visit the examples directory.

Benchmark

The following benchmarks provide a rough estimate of pure scheduling overhead using empty task functions. Note that most realistic workloads are I/O-bound, and their performance cannot be accurately reflected by these results. For CPU-intensive tasks, consider using taskflow-cpp.

Benchmark naming: N{task_count}-C{concurrency} where concurrency is goroutine pool size.

…

Understanding Conditional Tasks

Conditional nodes in go-taskflow behave similarly to those in taskflow-cpp. They participate in both conditional control and looping. To avoid common pitfalls, refer to the Conditional Tasking documentation.

Executor Options

NewExecutor accepts functional options to configure behavior:

executor := gtf.NewExecutor(1000,
    gtf.WithProfiler(), // enable flamegraph profiling
    gtf.WithTracer(),   // enable Chrome Trace recording
)

Error Handling in go-taskflow

In Go, errors are values, and it is the user's responsibility to handle them appropriately. Only unrecovered panic events are managed by the framework. If a panic occurs, the entire parent graph is canceled, leaving the remaining tasks incomplete. This behavior may evolve in the future. If you have suggestions, feel free to share them.

To prevent interruptions caused by panic, you can handle them manually when registering tasks:

tf.NewTask("not interrupt", func() {
    defer func() {
        if r := recover(); r != nil {
            // Handle the panic.
        }
    }()
    // User-defined logic.
})

Visualizing Taskflows

To generate a visual representation of a taskflow, use the Dump method:

if err := tf.Dump(os.Stdout); err != nil {
    log.Fatal(err)
}

The Dump method generates raw strings in DOT format. Use the dot tool to create a graph SVG.

Profiling Taskflows

To profile a taskflow, first enable the profiler with WithProfiler(), then call Profile:

executor := gtf.NewExecutor(1000, gtf.WithProfiler())
executor.Run(tf).Wait()

if err := executor.Profile(os.Stdout); err != nil {
    log.Fatal(err)
}

The Profile method generates raw strings in flamegraph format. Use the flamegraph tool to create a flamegraph SVG.

Tracing Taskflows

To trace a taskflow, first enable the tracer with WithTracer(), then call Trace:

executor := gtf.NewExecutor(1000, gtf.WithTracer())
executor.Run(tf).Wait()

if err := executor.Trace(os.Stdout); err != nil {
    log.Fatal(err)
}

The Trace method outputs JSON in Chrome Trace Event format. Open it in chrome://tracing or Perfetto UI for visualization.

Stargazer