Skip to content

Perf #110

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
noneback merged 1 commit into from
Jun 27, 2026
Merged

Perf #110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 44 additions & 22 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -182,35 +182,57 @@ For more examples, visit the [examples directory](https://github.com/noneback/go

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](https://github.com/taskflow/taskflow).

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

```plaintext
$ go test -bench=. -benchmem ./benchmark/
goos: darwin
goarch: arm64
pkg: github.com/noneback/go-taskflow/benchmark
cpu: Apple M4 Pro
BenchmarkConcurrent/N8-12 217042 5349 ns/op 1781 B/op 55 allocs/op
BenchmarkConcurrent/N32-12 47456 24439 ns/op 7566 B/op 213 allocs/op
BenchmarkConcurrent/N128-12 10000 116586 ns/op 32209 B/op 835 allocs/op
BenchmarkConcurrent/N512-12 2839 439930 ns/op 130337 B/op 3353 allocs/op
BenchmarkSerial/N8-12 126259 9339 ns/op 1905 B/op 63 allocs/op
BenchmarkSerial/N32-12 30313 39171 ns/op 7669 B/op 255 allocs/op
BenchmarkSerial/N128-12 7758 156781 ns/op 30725 B/op 1023 allocs/op
BenchmarkSerial/N512-12 1862 645739 ns/op 122952 B/op 4095 allocs/op
BenchmarkDiamond-12 181072 6662 ns/op 1441 B/op 47 allocs/op
BenchmarkDenseLayers/L4xW4-12 85122 13461 ns/op 4352 B/op 123 allocs/op
BenchmarkDenseLayers/L4xW8-12 42927 27127 ns/op 11764 B/op 270 allocs/op
BenchmarkDenseLayers/L8xW4-12 44412 27565 ns/op 8963 B/op 251 allocs/op
BenchmarkDenseLayers/L8xW8-12 20775 58088 ns/op 25071 B/op 556 allocs/op
BenchmarkSubflow-12 170228 6531 ns/op 1409 B/op 45 allocs/op
BenchmarkCondition-12 507645 2460 ns/op 704 B/op 23 allocs/op
BenchmarkConcurrencyScaling/C1-12 82400 14740 ns/op 15687 B/op 393 allocs/op
BenchmarkConcurrencyScaling/C12-12 23158 52602 ns/op 15700 B/op 422 allocs/op
BenchmarkConcurrencyScaling/C48-12 17500 68434 ns/op 15907 B/op 453 allocs/op
BenchmarkGraphBuild/N32-12 348994 3418 ns/op 6284 B/op 230 allocs/op
BenchmarkGraphBuild/N128-12 92428 13300 ns/op 24856 B/op 904 allocs/op
BenchmarkGraphBuild/N512-12 22143 55325 ns/op 101709 B/op 3850 allocs/op
cpu: Apple M4
BenchmarkConcurrent/N8-C10-10 329973 3765 ns/op 994 B/op 46 allocs/op
BenchmarkConcurrent/N8-C40-10 293101 4232 ns/op 1023 B/op 47 allocs/op
BenchmarkConcurrent/N8-C80-10 282382 4282 ns/op 1025 B/op 48 allocs/op
BenchmarkConcurrent/N32-C10-10 82747 14659 ns/op 4384 B/op 177 allocs/op
BenchmarkConcurrent/N32-C40-10 62128 18632 ns/op 4618 B/op 193 allocs/op
BenchmarkConcurrent/N32-C80-10 63214 19002 ns/op 4638 B/op 194 allocs/op
BenchmarkConcurrent/N128-C10-10 18846 61521 ns/op 19299 B/op 698 allocs/op
BenchmarkConcurrent/N128-C40-10 14550 79929 ns/op 20035 B/op 763 allocs/op
BenchmarkConcurrent/N128-C80-10 14028 85004 ns/op 20204 B/op 777 allocs/op
BenchmarkConcurrent/N512-C10-10 4621 250798 ns/op 79518 B/op 2794 allocs/op
BenchmarkConcurrent/N512-C40-10 3685 336991 ns/op 82147 B/op 3045 allocs/op
BenchmarkConcurrent/N512-C80-10 3523 338505 ns/op 82767 B/op 3093 allocs/op
BenchmarkSerial/N8-C10-10 103927 11773 ns/op 1080 B/op 55 allocs/op
BenchmarkSerial/N8-C40-10 105153 11625 ns/op 1080 B/op 55 allocs/op
BenchmarkSerial/N32-C10-10 24338 50010 ns/op 4346 B/op 223 allocs/op
BenchmarkSerial/N32-C40-10 24327 49822 ns/op 4346 B/op 223 allocs/op
BenchmarkSerial/N128-C10-10 6091 200769 ns/op 17410 B/op 895 allocs/op
BenchmarkSerial/N128-C40-10 6040 200494 ns/op 17411 B/op 895 allocs/op
BenchmarkSerial/N512-C10-10 1484 808528 ns/op 69668 B/op 3583 allocs/op
BenchmarkSerial/N512-C40-10 1568 809357 ns/op 69669 B/op 3583 allocs/op
BenchmarkDiamond-10 183015 6605 ns/op 816 B/op 41 allocs/op
BenchmarkDenseLayers/L4xW4-C10-10 109255 11290 ns/op 2421 B/op 107 allocs/op
BenchmarkDenseLayers/L4xW4-C40-10 107066 11000 ns/op 2433 B/op 107 allocs/op
BenchmarkDenseLayers/L4xW8-C10-10 58836 20358 ns/op 5553 B/op 210 allocs/op
BenchmarkDenseLayers/L4xW8-C40-10 56614 20908 ns/op 5639 B/op 215 allocs/op
BenchmarkDenseLayers/L8xW4-C10-10 52110 23374 ns/op 4969 B/op 218 allocs/op
BenchmarkDenseLayers/L8xW4-C40-10 51250 23861 ns/op 4995 B/op 219 allocs/op
BenchmarkDenseLayers/L8xW8-C10-10 27162 42796 ns/op 11628 B/op 429 allocs/op
BenchmarkDenseLayers/L8xW8-C40-10 26202 45285 ns/op 11787 B/op 439 allocs/op
BenchmarkSubflow-10 191072 9264 ns/op 800 B/op 39 allocs/op
BenchmarkCondition-10 426183 2805 ns/op 392 B/op 19 allocs/op
BenchmarkLoop/Iter3-10 127936 10096 ns/op 2257 B/op 90 allocs/op
BenchmarkLoop/Iter5-10 86608 13919 ns/op 2785 B/op 116 allocs/op
BenchmarkLoop/Iter10-10 49532 24666 ns/op 4106 B/op 181 allocs/op
BenchmarkConcurrencyScaling/C1-10 93998 13641 ns/op 9471 B/op 328 allocs/op
BenchmarkConcurrencyScaling/C10-10 37479 30258 ns/op 9291 B/op 351 allocs/op
BenchmarkConcurrencyScaling/C40-10 29713 39099 ns/op 9711 B/op 384 allocs/op
BenchmarkGraphBuild/N32-10 287893 4204 ns/op 6283 B/op 230 allocs/op
BenchmarkGraphBuild/N128-10 72253 16587 ns/op 24852 B/op 904 allocs/op
BenchmarkGraphBuild/N512-10 17781 67629 ns/op 101692 B/op 3850 allocs/op
```


## Understanding Conditional Tasks

Conditional nodes in go-taskflow behave similarly to those in [taskflow-cpp](https://github.com/taskflow/taskflow). They participate in both conditional control and looping. To avoid common pitfalls, refer to the [Conditional Tasking documentation](https://taskflow.github.io/taskflow/ConditionalTasking.html).
Expand Down
126 changes: 83 additions & 43 deletions benchmark/benchmark_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,39 +9,46 @@ import (
)

// --- Topology scaling: measure scheduling overhead across graph shapes and sizes ---
// concurrency is goroutine pool size, can be larger than NumCPU since goroutines are lightweight

func BenchmarkConcurrent(b *testing.B) {
numCPU := runtime.NumCPU()
for _, n := range []int{8, 32, 128, 512} {
b.Run(fmt.Sprintf("N%d", n), func(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(runtime.NumCPU()))
tf := gotaskflow.NewTaskFlow("concurrent")
for i := 0; i < n; i++ {
tf.NewTask(fmt.Sprintf("T%d", i), func() {})
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec.Run(tf).Wait()
}
})
for _, c := range []int{numCPU, numCPU * 4, numCPU * 8} {
b.Run(fmt.Sprintf("N%d-C%d", n, c), func(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(c))
tf := gotaskflow.NewTaskFlow("concurrent")
for i := 0; i < n; i++ {
tf.NewTask(fmt.Sprintf("T%d", i), func() {})
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec.Run(tf).Wait()
}
})
}
}
}

func BenchmarkSerial(b *testing.B) {
numCPU := runtime.NumCPU()
for _, n := range []int{8, 32, 128, 512} {
b.Run(fmt.Sprintf("N%d", n), func(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(runtime.NumCPU()))
tf := gotaskflow.NewTaskFlow("serial")
prev := tf.NewTask("T0", func() {})
for i := 1; i < n; i++ {
next := tf.NewTask(fmt.Sprintf("T%d", i), func() {})
prev.Precede(next)
prev = next
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec.Run(tf).Wait()
}
})
for _, c := range []int{numCPU, numCPU * 4} {
b.Run(fmt.Sprintf("N%d-C%d", n, c), func(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(c))
tf := gotaskflow.NewTaskFlow("serial")
prev := tf.NewTask("T0", func() {})
for i := 1; i < n; i++ {
next := tf.NewTask(fmt.Sprintf("T%d", i), func() {})
prev.Precede(next)
prev = next
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec.Run(tf).Wait()
}
})
}
}
}

Expand All @@ -65,28 +72,31 @@ func BenchmarkDiamond(b *testing.B) {
}

func BenchmarkDenseLayers(b *testing.B) {
numCPU := runtime.NumCPU()
for _, layers := range []int{4, 8} {
for _, width := range []int{4, 8} {
b.Run(fmt.Sprintf("L%dxW%d", layers, width), func(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(runtime.NumCPU()))
tf := gotaskflow.NewTaskFlow("dense_layers")
var curLayer, prevLayer []*gotaskflow.Task
for l := 0; l < layers; l++ {
for w := 0; w < width; w++ {
task := tf.NewTask(fmt.Sprintf("T%d_%d", l, w), func() {})
for _, p := range prevLayer {
p.Precede(task)
for _, c := range []int{numCPU, numCPU * 4} {
b.Run(fmt.Sprintf("L%dxW%d-C%d", layers, width, c), func(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(c))
tf := gotaskflow.NewTaskFlow("dense_layers")
var curLayer, prevLayer []*gotaskflow.Task
for l := 0; l < layers; l++ {
for w := 0; w < width; w++ {
task := tf.NewTask(fmt.Sprintf("T%d_%d", l, w), func() {})
for _, p := range prevLayer {
p.Precede(task)
}
curLayer = append(curLayer, task)
}
curLayer = append(curLayer, task)
prevLayer = curLayer
curLayer = nil
}
prevLayer = curLayer
curLayer = nil
}
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec.Run(tf).Wait()
}
})
b.ResetTimer()
for i := 0; i < b.N; i++ {
exec.Run(tf).Wait()
}
})
}
}
}
}
Expand Down Expand Up @@ -130,6 +140,36 @@ func BenchmarkCondition(b *testing.B) {
}
}

func BenchmarkLoop(b *testing.B) {
exec := gotaskflow.NewExecutor(uint(runtime.NumCPU()))
for _, iterations := range []int{3, 5, 10} {
b.Run(fmt.Sprintf("Iter%d", iterations), func(b *testing.B) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
tf := gotaskflow.NewTaskFlow("loop")
count := 0
init := tf.NewTask("init", func() {
count = 0
})
work := tf.NewTask("work", func() {
count++
})
cond := tf.NewCondition("cond", func() uint {
if count < iterations {
return 0 // continue loop
}
return 1 // exit
})
exit := tf.NewTask("exit", func() {})
init.Precede(work)
work.Precede(cond)
cond.Precede(work, exit) // 0 -> work (loop), 1 -> exit
exec.Run(tf).Wait()
}
})
}
}

// --- Concurrency scaling: fixed topology, varying executor concurrency ---

func BenchmarkConcurrencyScaling(b *testing.B) {
Expand Down
55 changes: 13 additions & 42 deletions executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@ import (
"runtime/debug"
"slices"
"sync"
"time"

"github.com/noneback/go-taskflow/utils"
)
Expand All @@ -26,8 +25,7 @@ type innerExecutorImpl struct {
pool *utils.Copool
wq *utils.Queue[*innerNode]
wg *sync.WaitGroup
profiler *profiler
tracer *tracer
obs *observer
mu *sync.Mutex
}

Expand All @@ -42,6 +40,7 @@ func NewExecutor(concurrency uint, opts ...Option) Executor {
wq: utils.NewQueue[*innerNode](false),
wg: &sync.WaitGroup{},
mu: &sync.Mutex{},
obs: newObserver(),
}
for _, opt := range opts {
opt(e)
Expand Down Expand Up @@ -102,17 +101,6 @@ func (e *innerExecutorImpl) sche_successors(node *innerNode) {
e.schedule(candidate...)
}

// record submits the span to active observers.
// ok=true means the node completed without panic; profiler only records successful spans.
func (e *innerExecutorImpl) record(s *span, ok bool) {
if ok && e.profiler != nil {
e.profiler.AddSpan(s)
}
if e.tracer != nil {
e.tracer.AddEvent(s)
}
}

// getDependentNames extracts predecessor task names from a node.
func getDependentNames(node *innerNode) []string {
if len(node.dependents) == 0 {
Expand All @@ -127,20 +115,14 @@ func getDependentNames(node *innerNode) []string {

func (e *innerExecutorImpl) invokeStatic(node *innerNode, parentSpan *span, p *Static) func() {
return func() {
span := span{extra: attr{
typ: nodeStatic,
name: node.name,
}, begin: time.Now(), parent: parentSpan, dependents: getDependentNames(node)}

s := e.obs.openSpan(node, parentSpan)
defer func() {
span.cost = time.Since(span.begin)
r := recover()
if r != nil {
node.g.canceled.Store(true)
log.Printf("[go-taskflow] graph %q canceled: static task %q panicked: %v\n%s", node.g.name, node.name, r, debug.Stack())
}
e.record(&span, r == nil)

e.obs.closeSpan(s, r == nil)
node.drop()
e.sche_successors(node)
node.g.deref()
Expand All @@ -156,22 +138,16 @@ func (e *innerExecutorImpl) invokeStatic(node *innerNode, parentSpan *span, p *S

func (e *innerExecutorImpl) invokeSubflow(node *innerNode, parentSpan *span, p *Subflow) func() {
return func() {
span := span{extra: attr{
typ: nodeSubflow,
name: node.name,
}, begin: time.Now(), parent: parentSpan, dependents: getDependentNames(node)}

s := e.obs.openSpan(node, parentSpan)
defer func() {
span.cost = time.Since(span.begin)
r := recover()
if r != nil {
log.Printf("[go-taskflow] graph %q canceled: subflow %q panicked: %v\n%s", node.g.name, node.name, r, debug.Stack())
node.g.canceled.Store(true)
p.g.canceled.Store(true)
}
e.record(&span, r == nil)

e.scheduleGraph(node.g, p.g, &span)
e.obs.closeSpan(s, r == nil)
e.scheduleGraph(node.g, p.g, s)
node.drop()
e.sche_successors(node)
node.g.deref()
Expand All @@ -191,19 +167,14 @@ func (e *innerExecutorImpl) invokeSubflow(node *innerNode, parentSpan *span, p *

func (e *innerExecutorImpl) invokeCondition(node *innerNode, parentSpan *span, p *Condition) func() {
return func() {
span := span{extra: attr{
typ: nodeCondition,
name: node.name,
}, begin: time.Now(), parent: parentSpan, dependents: getDependentNames(node)}

s := e.obs.openSpan(node, parentSpan)
defer func() {
span.cost = time.Since(span.begin)
r := recover()
if r != nil {
node.g.canceled.Store(true)
log.Printf("[go-taskflow] graph %q canceled: condition task %q panicked: %v\n%s", node.g.name, node.name, r, debug.Stack())
}
e.record(&span, r == nil)
e.obs.closeSpan(s, r == nil)
node.drop()
// e.sche_successors(node)
node.g.deref()
Expand Down Expand Up @@ -284,16 +255,16 @@ func (e *innerExecutorImpl) Wait() {

// Profile write flame graph raw text into w
func (e *innerExecutorImpl) Profile(w io.Writer) error {
if e.profiler == nil {
if e.obs.profiler == nil {
return nil
}
return e.profiler.draw(w)
return e.obs.profiler.draw(w)
}

// Trace write Chrome Trace Event data into w
func (e *innerExecutorImpl) Trace(w io.Writer) error {
if e.tracer == nil {
if e.obs.tracer == nil {
return nil
}
return e.tracer.draw(w)
return e.obs.tracer.draw(w)
}
4 changes: 2 additions & 2 deletions options.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,14 @@ type Option func(*innerExecutorImpl)
// WithProfiler enables flame graph profiling for task execution analysis.
func WithProfiler() Option {
return func(e *innerExecutorImpl) {
e.profiler = newProfiler()
e.obs.withProfiler(newProfiler())
}
}

// WithTracer enables Chrome Trace Event recording for task execution analysis.
// The trace output can be visualized in chrome://tracing or Perfetto UI.
func WithTracer() Option {
return func(e *innerExecutorImpl) {
e.tracer = newTracer()
e.obs.withTracer(newTracer())
}
}
Loading
Loading