From 9dc14738836bc1b6c43291020cee27bce854563d Mon Sep 17 00:00:00 2001 From: noneback Date: Sat, 27 Jun 2026 11:54:12 +0800 Subject: [PATCH] perf(executor): introduce observer pattern for 40-45% memory reduction - Add observer struct with openSpan/closeSpan lifecycle management - Replace profiler/tracer fields with observer in executor - Simplify invokeStatic/invokeSubflow/invokeCondition functions - Add concurrency scaling benchmarks (C10/C40/C80 variants) - Update README with benchmark results showing memory improvements Key findings: - Memory allocation reduced by ~40-45% (span not created when tracing disabled) - Allocs/op reduced by ~15-21% - Larger concurrency can hurt performance in high-concurrency small-task scenarios Co-Authored-By: Claude Opus 4.8 --- README.md | 66 ++++++++++++------- benchmark/benchmark_test.go | 126 ++++++++++++++++++++++++------------ executor.go | 55 ++++------------ options.go | 4 +- profiler.go | 45 +++++++++++++ validator_test.go | 4 +- 6 files changed, 189 insertions(+), 111 deletions(-) diff --git a/README.md b/README.md index 5ac1acc..ff83962 100644 --- a/README.md +++ b/README.md @@ -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). diff --git a/benchmark/benchmark_test.go b/benchmark/benchmark_test.go index a73640a..feeca83 100644 --- a/benchmark/benchmark_test.go +++ b/benchmark/benchmark_test.go @@ -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() + } + }) + } } } @@ -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() + } + }) + } } } } @@ -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) { diff --git a/executor.go b/executor.go index 7284f86..c39e82c 100644 --- a/executor.go +++ b/executor.go @@ -8,7 +8,6 @@ import ( "runtime/debug" "slices" "sync" - "time" "github.com/noneback/go-taskflow/utils" ) @@ -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 } @@ -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) @@ -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 { @@ -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() @@ -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() @@ -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() @@ -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) } diff --git a/options.go b/options.go index 8f5fe14..7aae25a 100644 --- a/options.go +++ b/options.go @@ -6,7 +6,7 @@ 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()) } } @@ -14,6 +14,6 @@ func WithProfiler() Option { // 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()) } } diff --git a/profiler.go b/profiler.go index 3e44a6a..20f11b5 100644 --- a/profiler.go +++ b/profiler.go @@ -74,3 +74,48 @@ func (t *profiler) draw(w io.Writer) error { } return nil } + +// observer 观察任务执行并记录 span +type observer struct { + profiler *profiler + tracer *tracer +} + +func newObserver() *observer { + return &observer{} +} + +// openSpan 创建 span（如果需要观察） +func (o *observer) openSpan(node *innerNode, parent *span) *span { + if o.profiler == nil && o.tracer == nil { + return nil + } + return &span{ + extra: attr{typ: node.Typ, name: node.name}, + begin: time.Now(), + parent: parent, + dependents: getDependentNames(node), + } +} + +// closeSpan 结束 span 并记录 +func (o *observer) closeSpan(s *span, ok bool) { + if s == nil { + return + } + s.cost = time.Since(s.begin) + if ok && o.profiler != nil { + o.profiler.AddSpan(s) + } + if o.tracer != nil { + o.tracer.AddEvent(s) + } +} + +func (o *observer) withProfiler(p *profiler) { + o.profiler = p +} + +func (o *observer) withTracer(t *tracer) { + o.tracer = t +} diff --git a/validator_test.go b/validator_test.go index d6d946d..05dbbd2 100644 --- a/validator_test.go +++ b/validator_test.go @@ -308,10 +308,10 @@ func TestValidatorResultString(t *testing.T) { // Returns nil if the executor was not created with WithTracer(). func mustSnapshot(e Executor) traceRecord { impl := e.(*innerExecutorImpl) - if impl.tracer == nil { + if impl.obs.tracer == nil { return nil } - return impl.tracer.snapshot() + return impl.obs.tracer.snapshot() } func containsStr(slice []string, s string) bool {