README
¶
Go-Taskflow
A taskflow-like General-purpose Task-parallel Programming Framework for Go, inspired by taskflow-cpp, with Go's native capabilities and simplicity, suitable for complex dependency management in concurrent tasks.
Feature
-
High extensibility: Easily extend the framework to adapt to various specific use cases.
-
Native Go's concurrency model: Leverages Go's goroutines to manage concurrent task execution effectively.
-
User-friendly programming interface: Simplify complex task dependency management using Go.
-
Static\Subflow\Conditional\Cyclic tasking: Define static tasks, condition nodes, nested subflows and cyclic flow to enhance modularity and programmability.
Static Subflow Condition Cyclic 
-
Priority Task Schedule: Define tasks' priority, higher priority tasks will be scheduled first.
-
Built-in visualization & profiling tools: Generate visual representations of tasks and profile task execution performance using integrated tools, making debugging and optimization easier.
Use Cases
-
Data Pipeline: Orchestrate data processing stages that have complex dependencies.
-
Workflow Automation: Define and run automation workflows where tasks have a clear sequence and dependency structure.
-
Parallel Tasking: Execute independent tasks concurrently to fully utilize CPU resources.
Example
import latest version: go get -u github.com/noneback/go-taskflow
Understand Condition Task Correctly
Condition Node is special in taskflow-cpp. It not only enrolls in Condition Control but also in Looping.
Our repo keeps almost the same behavior. You should read ConditionTasking to avoid common pitfalls.
Error Handling in go-taskflow
errors in golang are values. It is the user's job to handle it correctly.
Only unrecovered panic needs to be addressed by the framework. Now, if it happens, the whole parent graph will be canceled, leaving the rest tasks undone. This behavior may evolve someday. If you have any good thoughts, feel free to let me know.
How to use visualize taskflow
if err := tf.Dump(os.Stdout); err != nil {
log.Fatal(err)
}
tf.Dump generates raw strings in dot format, use dot to draw a Graph svg.
How to use profile taskflow
if err :=exector.Profile(os.Stdout);err != nil {
log.Fatal(err)
}
Profile generates raw strings in flamegraph format, use flamegraph to draw a flamegraph svg.
What's more
Any Features Request or Discussions are all welcomed.
Documentation
¶
Index ¶
- Constants
- type Condition
- type Executor
- type Static
- type Subflow
- type Task
- type TaskFlow
- func (tf *TaskFlow) Dump(writer io.Writer) error
- func (tf *TaskFlow) Name() string
- func (tf *TaskFlow) NewCondition(name string, predict func() uint) *Task
- func (tf *TaskFlow) NewSubflow(name string, f func(sf *Subflow)) *Task
- func (tf *TaskFlow) NewTask(name string, f func()) *Task
- func (tf *TaskFlow) Reset()
- type TaskPriority
Constants ¶
const ( HIGH = TaskPriority(iota + 1) NORMAL LOW )
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Condition ¶ added in v0.0.5
type Condition struct {
// contains filtered or unexported fields
}
Condition Wrapper
type Executor ¶
type Executor interface {
Wait() // Wait block until all tasks finished
Profile(w io.Writer) error // Profile write flame graph raw text into w
Run(tf *TaskFlow) Executor // Run start to schedule and execute taskflow
}
Executor schedule and execute taskflow
func NewExecutor ¶
NewExecutor return a Executor with a specified max goroutine concurrency(recommend a value bigger than Runtime.NumCPU)
type Subflow ¶
type Subflow struct {
// contains filtered or unexported fields
}
Subflow Wrapper
func (*Subflow) NewCondition ¶ added in v0.1.2
NewCondition returns a condition task. The predict func return value determines its successor.
func (*Subflow) NewSubflow ¶ added in v0.1.2
NewSubflow returns a subflow task
type Task ¶
type Task struct {
// contains filtered or unexported fields
}
Basic component of Taskflow
func (*Task) Precede ¶
Precede: Tasks all depend on *this*. In Addition, order of tasks is correspond to predict result, ranging from 0...len(tasks)
func (*Task) Priority ¶ added in v0.0.9
func (t *Task) Priority(p TaskPriority) *Task
Priority sets task's sche priority. Noted that due to goroutine concurrent mode, it can only assure task schedule priority, rather than its execution.
type TaskFlow ¶
type TaskFlow struct {
// contains filtered or unexported fields
}
TaskFlow represents a series of tasks organized in DAG. Tasks must be pushed via a `Push` api.
func (*TaskFlow) NewCondition ¶ added in v0.1.2
NewCondition returns a condition task. The predict func return value determines its successor.
func (*TaskFlow) NewSubflow ¶ added in v0.1.2
NewSubflow returns a subflow task
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
examples
|
|
|
conditional
command
|
|
|
loop
command
|
|
|
parallel_merge_sort
command
|
|
|
simple
command
|
|