From a3689968981cf1d30f45fbd3146e09304fa21e76 Mon Sep 17 00:00:00 2001 From: noneback Date: Fri, 26 Jun 2026 15:43:03 +0800 Subject: [PATCH] refactor(examples): rewrite with classic problems & meaningful names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - simple: MapReduce word count (split→map×4→shuffle→reduce×2→merge) - conditional: event severity routing (ingest→classify→3 processing paths) - loop: web crawler with retry & exponential backoff - ci_pipeline: CI/CD with nested subflows (build→test→deploy) - priority: fix typo, remove internal utils dependency - fibonacci: fix output label - parallel_merge_sort: add WithProfiler, remove internal pprof - README: add Quick Start with MapReduce, fix merge sort closure bug - Regenerate all SVGs with meaningful node names --- README.md | 169 +++++++++-- examples/ci_pipeline/ci_pipeline.go | 66 +++++ examples/conditional/condition.go | 180 ++++++++---- examples/fibonacci/fibonacci.go | 2 +- examples/loop/loop.go | 136 ++++++--- .../parallel_merge_sort.go | 40 +-- examples/priority/priority.go | 59 ++-- examples/simple/simple.go | 178 +++++++---- image/condition.svg | 243 ++++++--------- image/desc.svg | 278 +++++++----------- image/loop.svg | 124 ++++---- image/simple.svg | 211 ++++++------- image/subflow.svg | 236 +++++++-------- 13 files changed, 1086 insertions(+), 836 deletions(-) create mode 100644 examples/ci_pipeline/ci_pipeline.go diff --git a/README.md b/README.md index 6277c60..6245c77 100644 --- a/README.md +++ b/README.md @@ -47,9 +47,138 @@ go get -u github.com/noneback/go-taskflow [DeepWiki Page](https://deepwiki.com/noneback/go-taskflow) -## Example +## Quick Start -Below is an example of using go-taskflow to implement a parallel merge sort: +A MapReduce word-count pipeline with parallel mappers, hash-partitioned shuffle, and reducers: + +```go +package main + +import ( + "fmt" + "log" + "os" + "sort" + "strings" + "sync" + + gtf "github.com/noneback/go-taskflow" +) + +const ( + numMappers = 4 + numReducers = 2 +) + +var ( + mapOutputs [numMappers][numReducers]map[string]int + mu sync.Mutex +) + +func hashPartition(word string) int { + h := 0 + for _, c := range word { + h = 31*h + int(c) + } + if h < 0 { + h = -h + } + return h % numReducers +} + +func main() { + input := `the quick brown fox jumps over the lazy dog +the fox and the dog are friends the dog jumps over the fox +brown dog lazy fox the quick brown dog the lazy fox` + + var chunks [numMappers]string + executor := gtf.NewExecutor(8, gtf.WithProfiler()) + tf := gtf.NewTaskFlow("word-count") + + // Phase 1: Split input into chunks for parallel processing + splitTask := tf.NewTask("split_input", func() { + words := strings.Fields(input) + size := (len(words) + numMappers - 1) / numMappers + for i := 0; i < numMappers; i++ { + start, end := i*size, (i+1)*size + if end > len(words) { + end = len(words) + } + chunks[i] = strings.Join(words[start:end], " ") + } + }) + + // Phase 2: Map — count words per chunk, partition by hash + mapTasks := make([]*gtf.Task, numMappers) + for i := 0; i < numMappers; i++ { + idx := i + mapTasks[idx] = tf.NewTask(fmt.Sprintf("map_%d", idx), func() { + local := [numReducers]map[string]int{} + for r := 0; r < numReducers; r++ { + local[r] = make(map[string]int) + } + for _, w := range strings.Fields(chunks[idx]) { + local[hashPartition(w)][w]++ + } + mu.Lock() + for r := 0; r < numReducers; r++ { + mapOutputs[idx][r] = local[r] + } + mu.Unlock() + }) + } + splitTask.Precede(mapTasks...) + + // Phase 3: Reduce — aggregate each hash partition + reduceTasks := make([]*gtf.Task, numReducers) + reduceResults := make([]map[string]int, numReducers) + for r := 0; r < numReducers; r++ { + rIdx := r + reduceResults[rIdx] = make(map[string]int) + reduceTasks[rIdx] = tf.NewTask(fmt.Sprintf("reduce_%d", rIdx), func() { + for m := 0; m < numMappers; m++ { + for word, count := range mapOutputs[m][rIdx] { + reduceResults[rIdx][word] += count + } + } + }) + } + for _, mt := range mapTasks { + mt.Precede(reduceTasks...) + } + + // Phase 4: Merge final results + mergeTask := tf.NewTask("merge_results", func() { + final := make(map[string]int) + for r := 0; r < numReducers; r++ { + for w, c := range reduceResults[r] { + final[w] += c + } + } + keys := make([]string, 0, len(final)) + for k := range final { + keys = append(keys, k) + } + sort.Strings(keys) + for _, w := range keys { + fmt.Printf("%-12s %d\n", w, final[w]) + } + }) + for _, rt := range reduceTasks { + rt.Precede(mergeTask) + } + + executor.Run(tf).Wait() + + if err := tf.Dump(os.Stdout); err != nil { + log.Fatal(err) + } +} +``` + +## Example: Parallel Merge Sort + +A real-world example using go-taskflow to parallelize merge sort across multiple chunks: ```go package main @@ -66,7 +195,6 @@ import ( gtf "github.com/noneback/go-taskflow" ) -// mergeInto merges a sorted source array into a sorted destination array. func mergeInto(dest, src []int) []int { size := len(dest) + len(src) tmp := make([]int, 0, size) @@ -80,41 +208,40 @@ func mergeInto(dest, src []int) []int { j++ } } - if i < len(dest) { tmp = append(tmp, dest[i:]...) } else { tmp = append(tmp, src[j:]...) } - return tmp } func main() { - size := 100 - randomArr := make([][]int, 10) - sortedArr := make([]int, 0, 10*size) + chunks := 100 + chunkSize := 1000 + randomArr := make([][]int, chunks) + sortedArr := make([]int, 0, chunks*chunkSize) mutex := &sync.Mutex{} - for i := 0; i < 10; i++ { - for j := 0; j < size; j++ { + for i := 0; i < chunks; i++ { + for j := 0; j < chunkSize; j++ { randomArr[i] = append(randomArr[i], rand.Int()) } } - sortTasks := make([]*gtf.Task, 10) + sortTasks := make([]*gtf.Task, chunks) tf := gtf.NewTaskFlow("merge sort") - done := tf.NewTask("Done", func() { + done := tf.NewTask("done", func() { if !slices.IsSorted(sortedArr) { - log.Fatal("Sorting failed") + log.Fatal("sorting failed") } - fmt.Println("Sorted successfully") - fmt.Println(sortedArr[:1000]) + fmt.Println("sorted successfully") }) - for i := 0; i < 10; i++ { - sortTasks[i] = tf.NewTask("sort_"+strconv.Itoa(i), func() { - arr := randomArr[i] + for i := 0; i < chunks; i++ { + idx := i + sortTasks[idx] = tf.NewTask("sort_"+strconv.Itoa(idx), func() { + arr := randomArr[idx] slices.Sort(arr) mutex.Lock() defer mutex.Unlock() @@ -124,15 +251,13 @@ func main() { done.Succeed(sortTasks...) executor := gtf.NewExecutor(1000, gtf.WithProfiler()) - executor.Run(tf).Wait() if err := tf.Dump(os.Stdout); err != nil { - log.Fatal("Error dumping taskflow:", err) + log.Fatal(err) } - if err := executor.Profile(os.Stdout); err != nil { - log.Fatal("Error profiling taskflow:", err) + log.Fatal(err) } } ``` diff --git a/examples/ci_pipeline/ci_pipeline.go b/examples/ci_pipeline/ci_pipeline.go new file mode 100644 index 0000000..eeaae0d --- /dev/null +++ b/examples/ci_pipeline/ci_pipeline.go @@ -0,0 +1,66 @@ +package main + +import ( + "fmt" + "log" + "os" + + gotaskflow "github.com/noneback/go-taskflow" +) + +func main() { + executor := gotaskflow.NewExecutor(4, gotaskflow.WithProfiler()) + tf := gotaskflow.NewTaskFlow("ci-pipeline") + + // Stage 1: Build — parallel compile of modules + build := tf.NewSubflow("build", func(sf *gotaskflow.Subflow) { + frontend := sf.NewTask("compile_frontend", func() { + fmt.Println(" compiling frontend...") + }) + backend := sf.NewTask("compile_backend", func() { + fmt.Println(" compiling backend...") + }) + common := sf.NewTask("compile_common", func() { + fmt.Println(" compiling shared libs...") + }) + link := sf.NewTask("link", func() { + fmt.Println(" linking binaries...") + }) + frontend.Precede(link) + backend.Precede(link) + common.Precede(link) + }) + + // Stage 2: Test — parallel test suites + test := tf.NewSubflow("test", func(sf *gotaskflow.Subflow) { + unit := sf.NewTask("unit_test", func() { + fmt.Println(" running unit tests...") + }) + integration := sf.NewTask("integration_test", func() { + fmt.Println(" running integration tests...") + }) + e2e := sf.NewTask("e2e_test", func() { + fmt.Println(" running e2e tests...") + }) + report := sf.NewTask("test_report", func() { + fmt.Println(" generating test report...") + }) + unit.Precede(report) + integration.Precede(report) + e2e.Precede(report) + }) + + // Stage 3: Deploy + deploy := tf.NewTask("deploy", func() { + fmt.Println("deploying to production") + }) + + build.Precede(test) + test.Precede(deploy) + + executor.Run(tf).Wait() + + if err := tf.Dump(os.Stdout); err != nil { + log.Fatal(err) + } +} diff --git a/examples/conditional/condition.go b/examples/conditional/condition.go index 66e43db..5dccc85 100644 --- a/examples/conditional/condition.go +++ b/examples/conditional/condition.go @@ -3,86 +3,144 @@ package main import ( "fmt" "log" + "math/rand" "os" - "runtime" "time" gotaskflow "github.com/noneback/go-taskflow" ) -func main() { - executor := gotaskflow.NewExecutor(uint(runtime.NumCPU()-1) * 10000) - tf := gotaskflow.NewTaskFlow("G") - A, B, C := - tf.NewTask("A", func() { - fmt.Println("A") - }), - tf.NewTask("B", func() { - fmt.Println("B") - }), - tf.NewTask("C", func() { - fmt.Println("C") - }) +// An event-driven log processing pipeline with severity-based routing. +// +// DAG topology: +// +// ingest → classify → [severity_check] +// ├─→ critical_path: alert → escalate → page_oncall +// ├─→ warning_path: enrich → deduplicate → log_warning +// └─→ info_path: compress → archive +const eventTypes = 3 // 0=critical, 1=warning, 2=info - A1, B1, _ := - tf.NewTask("A1", func() { - fmt.Println("A1") - }), - tf.NewTask("B1", func() { - fmt.Println("B1") - }), - tf.NewTask("C1", func() { - fmt.Println("C1") - }) - A.Precede(B) - C.Precede(B) - A1.Precede(B) - C.Succeed(A1) - C.Succeed(B1) +var ( + severityLabels = []string{"critical", "warning", "info"} + eventBatch []map[string]string + processedLog []string +) - subflow := tf.NewSubflow("sub1", func(sf *gotaskflow.Subflow) { - A2, B2, C2 := - sf.NewTask("A2", func() { - fmt.Println("A2") - }), - sf.NewTask("B2", func() { - fmt.Println("B2") - }), - sf.NewTask("C2", func() { - fmt.Println("C2") +func main() { + executor := gotaskflow.NewExecutor(8, gotaskflow.WithProfiler()) + tf := gotaskflow.NewTaskFlow("event-processor") + + // Phase 1: Ingest — simulate receiving a batch of events + ingest := tf.NewTask("ingest", func() { + sources := []string{"webhook", "syslog", "api-gateway", "kafka", "sqs"} + eventBatch = make([]map[string]string, 0, 100) + for i := 0; i < 100; i++ { + severity := severityLabels[rand.Intn(3)] + eventBatch = append(eventBatch, map[string]string{ + "id": fmt.Sprintf("evt-%04d", i), + "source": sources[rand.Intn(len(sources))], + "severity": severity, + "msg": fmt.Sprintf("event payload %d", i), + "ts": time.Now().Format(time.RFC3339), }) - A2.Precede(B2) - C2.Precede(B2) + } + fmt.Printf("ingested %d events\n", len(eventBatch)) }) - subflow2 := tf.NewSubflow("sub2", func(sf *gotaskflow.Subflow) { - A3, B3, C3 := - sf.NewTask("A3", func() { - fmt.Println("A3") - }), - sf.NewTask("B3", func() { - fmt.Println("B3") - }), - sf.NewTask("C3", func() { - fmt.Println("C3") - // time.Sleep(10 * time.Second) - }) - A3.Precede(B3) - C3.Precede(B3) + // Phase 2: Classify — tag events with metadata + classify := tf.NewTask("classify", func() { + for _, evt := range eventBatch { + evt["region"] = fmt.Sprintf("us-%s", string("eastwest"[rand.Intn(2)*4:rand.Intn(2)*4+4])) + evt["tagged"] = "true" + } + fmt.Println("classified all events") }) + ingest.Precede(classify) - cond := tf.NewCondition("binary", func() uint { - return uint(time.Now().Second() % 2) + // Phase 3: Severity check — route by event severity + severityCheck := tf.NewCondition("severity_check", func() uint { + // Route based on the dominant severity in the batch + counts := map[string]int{} + for _, evt := range eventBatch { + counts[evt["severity"]]++ + } + if counts["critical"] > 20 { + fmt.Printf("routing → critical_path (critical=%d)\n", counts["critical"]) + return 0 + } else if counts["warning"] > 20 { + fmt.Printf("routing → warning_path (warning=%d)\n", counts["warning"]) + return 1 + } + fmt.Printf("routing → info_path (info=%d)\n", counts["info"]) + return 2 }) - B.Precede(cond) - cond.Precede(subflow, subflow2) + classify.Precede(severityCheck) + + // Critical path: alert → escalate → page on-call + criticalPath := tf.NewSubflow("critical_path", func(sf *gotaskflow.Subflow) { + alert := sf.NewTask("alert", func() { + critical := filterBySeverity("critical") + fmt.Printf(" ALERT: %d critical events detected\n", len(critical)) + }) + escalate := sf.NewTask("escalate", func() { + fmt.Println(" escalating to incident manager...") + }) + pageOnCall := sf.NewTask("page_oncall", func() { + fmt.Println(" paging on-call engineer via PagerDuty") + }) + alert.Precede(escalate) + escalate.Precede(pageOnCall) + }) + + // Warning path: enrich → deduplicate → log warning + warningPath := tf.NewSubflow("warning_path", func(sf *gotaskflow.Subflow) { + enrich := sf.NewTask("enrich", func() { + warnings := filterBySeverity("warning") + for _, evt := range warnings { + evt["enriched_with"] = "historical_context" + } + fmt.Printf(" enriched %d warning events\n", len(warnings)) + }) + dedup := sf.NewTask("deduplicate", func() { + fmt.Println(" deduplicating against known patterns...") + }) + logWarn := sf.NewTask("log_warning", func() { + fmt.Println(" logged warnings to monitoring dashboard") + }) + enrich.Precede(dedup) + dedup.Precede(logWarn) + }) + + // Info path: compress → archive + infoPath := tf.NewSubflow("info_path", func(sf *gotaskflow.Subflow) { + compress := sf.NewTask("compress", func() { + info := filterBySeverity("info") + fmt.Printf(" compressing %d info events\n", len(info)) + }) + archive := sf.NewTask("archive", func() { + fmt.Println(" archived to cold storage (S3 Glacier)") + }) + compress.Precede(archive) + }) + + severityCheck.Precede(criticalPath, warningPath, infoPath) + executor.Run(tf).Wait() - fmt.Println("Print DOT") + if err := tf.Dump(os.Stdout); err != nil { log.Fatal(err) } - fmt.Println("Print Flamegraph") if err := executor.Profile(os.Stdout); err != nil { log.Fatal(err) } } + +func filterBySeverity(severity string) []map[string]string { + var result []map[string]string + for _, evt := range eventBatch { + if evt["severity"] == severity { + result = append(result, evt) + } + } + return result +} diff --git a/examples/fibonacci/fibonacci.go b/examples/fibonacci/fibonacci.go index 6144ac8..3fb50cf 100644 --- a/examples/fibonacci/fibonacci.go +++ b/examples/fibonacci/fibonacci.go @@ -34,5 +34,5 @@ func main() { } executor.Run(tf).Wait() - fmt.Printf("F(%d) = %d\n", n+1, fib[n]) + fmt.Printf("F(%d) = %d\n", n, fib[n]) } diff --git a/examples/loop/loop.go b/examples/loop/loop.go index 0b2294f..8f53454 100644 --- a/examples/loop/loop.go +++ b/examples/loop/loop.go @@ -3,55 +3,117 @@ package main import ( "fmt" "log" + "math/rand" "os" - "runtime" "time" gotaskflow "github.com/noneback/go-taskflow" ) +// A web crawler with retry and exponential backoff. +// +// DAG topology: +// +// init → [fetch_pages] → parse → [validate_quality] +// ├─(pass)→ index → done +// └─(fail)→ backoff → fetch_pages (loop) +// +// Demonstrates: +// - Cyclic flow with retry logic +// - Exponential backoff +// - Quality validation before committing results +const maxAttempts = 5 + +var ( + attempt int + pageCount int + qualityOK bool + backoffMs int +) + func main() { + executor := gotaskflow.NewExecutor(4, gotaskflow.WithProfiler()) + tf := gotaskflow.NewTaskFlow("crawl-pipeline") + + // Initialize the crawl job + init := tf.NewTask("init", func() { + attempt = 0 + pageCount = 0 + qualityOK = false + backoffMs = 100 + fmt.Println("initializing crawl job...") + }) + + // Fetch pages — simulate fetching with random success rate + fetchPages := tf.NewTask("fetch_pages", func() { + attempt++ + // Simulate: each attempt fetches more pages + pagesFetched := 10 + attempt*5 + rand.Intn(20) + pageCount = pagesFetched + fmt.Printf("[attempt %d/%d] fetched %d pages\n", attempt, maxAttempts, pagesFetched) + time.Sleep(time.Duration(backoffMs) * time.Millisecond) + }) + + // Parse fetched pages + parse := tf.NewTask("parse", func() { + parsed := pageCount + fmt.Printf("parsed %d pages, extracted %d links\n", parsed, parsed*3) + }) + + // Validate quality — pass if we got enough pages + validateQuality := tf.NewCondition("validate_quality", func() uint { + threshold := 40 // need at least 40 pages + if pageCount >= threshold { + qualityOK = true + fmt.Printf("quality check PASSED (%d >= %d pages)\n", pageCount, threshold) + return 0 // pass → go to index + } + fmt.Printf("quality check FAILED (%d < %d pages)\n", pageCount, threshold) + if attempt >= maxAttempts { + fmt.Println("max attempts reached, giving up") + return 0 // force pass after max attempts + } + return 1 // fail → go to backoff and retry + }) - executor := gotaskflow.NewExecutor(uint(runtime.NumCPU()) * 100) - i := 0 - tf := gotaskflow.NewTaskFlow("G") - init, cond, body, back, done := - tf.NewTask("init", func() { - i = 0 - fmt.Println("i=0") - }), - tf.NewCondition("while i < 5", func() uint { - time.Sleep(100 * time.Millisecond) - if i < 5 { - return 0 - } else { - return 1 - } - }), - tf.NewTask("body", func() { - i += 1 - fmt.Println("i++ =", i) - }), - tf.NewCondition("back", func() uint { - fmt.Println("back") - return 0 - }), - tf.NewTask("done", func() { - fmt.Println("done") - }) - - init.Precede(cond) - cond.Precede(body, done) - body.Precede(back) - back.Precede(cond) + // Exponential backoff before retry — must be a condition node to allow cyclic flow + backoff := tf.NewCondition("backoff", func() uint { + backoffMs *= 2 + fmt.Printf("backing off %dms before retry...\n", backoffMs) + return 0 // always loop back to fetch_pages + }) + + // Index the parsed results + index := tf.NewTask("index", func() { + if qualityOK { + fmt.Printf("indexing %d pages into search engine\n", pageCount) + } else { + fmt.Printf("indexing %d pages (best effort after %d attempts)\n", pageCount, attempt) + } + }) + + // Final summary + done := tf.NewTask("done", func() { + fmt.Println("--- Crawl Summary ---") + fmt.Printf(" Total attempts: %d\n", attempt) + fmt.Printf(" Pages indexed: %d\n", pageCount) + fmt.Printf(" Quality: %v\n", qualityOK) + }) + + // Wire up the DAG + init.Precede(fetchPages) + fetchPages.Precede(parse) + parse.Precede(validateQuality) + validateQuality.Precede(index, backoff) + backoff.Precede(fetchPages) // loop back + index.Precede(done) executor.Run(tf).Wait() - if i < 5 { - log.Fatal("i < 5") - } if err := tf.Dump(os.Stdout); err != nil { log.Fatal(err) } - executor.Profile(os.Stdout) + if err := executor.Profile(os.Stdout); err != nil { + log.Fatal(err) + } } diff --git a/examples/parallel_merge_sort/parallel_merge_sort.go b/examples/parallel_merge_sort/parallel_merge_sort.go index 62ca586..df06225 100644 --- a/examples/parallel_merge_sort/parallel_merge_sort.go +++ b/examples/parallel_merge_sort/parallel_merge_sort.go @@ -10,10 +10,8 @@ import ( "sync" gtf "github.com/noneback/go-taskflow" - "github.com/noneback/go-taskflow/utils" ) -// merge sorted src to sorted dest func mergeInto(dest, src []int) []int { size := len(dest) + len(src) tmp := make([]int, 0, size) @@ -36,34 +34,31 @@ func mergeInto(dest, src []int) []int { return tmp } -func main() { - pprof := utils.NewPprofUtils(utils.CPU, "./out.prof") - pprof.StartProfile() - defer pprof.StopProfile() - size := 10000 - share := 1000 - randomArr := make([][]int, share) - sortedArr := make([]int, 0, share*size) +func main() { + chunks := 100 + chunkSize := 1000 + randomArr := make([][]int, chunks) + sortedArr := make([]int, 0, chunks*chunkSize) mutex := &sync.Mutex{} - for i := 0; i < share; i++ { - for j := 0; j < size; j++ { + for i := 0; i < chunks; i++ { + for j := 0; j < chunkSize; j++ { randomArr[i] = append(randomArr[i], rand.Int()) } } - sortTasks := make([]*gtf.Task, share) + sortTasks := make([]*gtf.Task, chunks) tf := gtf.NewTaskFlow("merge sort") - done := tf.NewTask("Done", func() { + done := tf.NewTask("done", func() { if !slices.IsSorted(sortedArr) { - log.Fatal("Failed") + log.Fatal("sorting failed") } - fmt.Println("Sorted") - fmt.Println(sortedArr[:1000]) + fmt.Println("sorted successfully") + fmt.Println("first 10:", sortedArr[:10]) }) - for i := 0; i < share; i++ { + for i := 0; i < chunks; i++ { idx := i sortTasks[idx] = tf.NewTask("sort_"+strconv.Itoa(idx), func() { arr := randomArr[idx] @@ -75,16 +70,13 @@ func main() { } done.Succeed(sortTasks...) - executor := gtf.NewExecutor(1000000) - + executor := gtf.NewExecutor(1000, gtf.WithProfiler()) executor.Run(tf).Wait() if err := tf.Dump(os.Stdout); err != nil { - log.Fatal("V->", err) + log.Fatal(err) } - if err := executor.Profile(os.Stdout); err != nil { - log.Fatal("P->", err) + log.Fatal(err) } - } diff --git a/examples/priority/priority.go b/examples/priority/priority.go index 773224e..ab7573e 100644 --- a/examples/priority/priority.go +++ b/examples/priority/priority.go @@ -2,39 +2,54 @@ package main import ( "fmt" + "sync" gotaskflow "github.com/noneback/go-taskflow" - "github.com/noneback/go-taskflow/utils" ) func main() { - exector := gotaskflow.NewExecutor(uint(2)) - q := utils.NewQueue[byte](true) - tf := gotaskflow.NewTaskFlow("G") + executor := gotaskflow.NewExecutor(2, gotaskflow.WithProfiler()) + tf := gotaskflow.NewTaskFlow("priority-demo") - tf.NewTask("B", func() { - fmt.Println("B") - q.Put('B') + var mu sync.Mutex + var order []string + + tf.NewTask("normal_task", func() { + mu.Lock() + order = append(order, "normal_task") + mu.Unlock() + fmt.Println("normal_task") }).Priority(gotaskflow.NORMAL) - tf.NewTask("C", func() { - fmt.Println("C") - q.Put('C') + + tf.NewTask("high_task", func() { + mu.Lock() + order = append(order, "high_task") + mu.Unlock() + fmt.Println("high_task") }).Priority(gotaskflow.HIGH) - tf.NewSubflow("sub1", func(sf *gotaskflow.Subflow) { - sf.NewTask("A2", func() { - fmt.Println("A2") - q.Put('a') + + tf.NewSubflow("low_subflow", func(sf *gotaskflow.Subflow) { + sf.NewTask("sub_low", func() { + mu.Lock() + order = append(order, "sub_low") + mu.Unlock() + fmt.Println(" sub_low") }).Priority(gotaskflow.LOW) - sf.NewTask("B2", func() { - fmt.Println("B2") - q.Put('b') + sf.NewTask("sub_high", func() { + mu.Lock() + order = append(order, "sub_high") + mu.Unlock() + fmt.Println(" sub_high") }).Priority(gotaskflow.HIGH) - sf.NewTask("C2", func() { - fmt.Println("C2") - q.Put('c') + sf.NewTask("sub_normal", func() { + mu.Lock() + order = append(order, "sub_normal") + mu.Unlock() + fmt.Println(" sub_normal") }).Priority(gotaskflow.NORMAL) - }).Priority(gotaskflow.LOW) - exector.Run(tf).Wait() + executor.Run(tf).Wait() + + fmt.Println("\nexecution order:", order) } diff --git a/examples/simple/simple.go b/examples/simple/simple.go index e10f504..ebe2cd3 100644 --- a/examples/simple/simple.go +++ b/examples/simple/simple.go @@ -4,81 +4,141 @@ import ( "fmt" "log" "os" - "runtime" + "sort" + "strings" + "sync" gotaskflow "github.com/noneback/go-taskflow" ) +// A simplified MapReduce word-count pipeline. +// +// DAG topology: +// +// split_input → [map_0, map_1, map_2, map_3] +// ↓ (shuffle by hash) +// [reduce_0, reduce_1] +// ↓ +// merge_results +const ( + numMappers = 4 + numReducers = 2 +) + +var ( + // intermediate results: mapOutputs[mapper][reducer] + mapOutputs [numMappers][numReducers]map[string]int + mu sync.Mutex +) + +func hashPartition(word string) int { + h := 0 + for _, c := range word { + h = 31*h + int(c) + } + if h < 0 { + h = -h + } + return h % numReducers +} + func main() { - executor := gotaskflow.NewExecutor(uint(runtime.NumCPU()-1) * 10000) + input := `the quick brown fox jumps over the lazy dog +the fox and the dog are friends +the dog jumps over the fox +brown dog lazy fox the quick brown dog +the lazy fox jumps over the quick dog +the dog and the fox play together +brown brown brown the fox the fox` - tf := gotaskflow.NewTaskFlow("G") - A, B, C := - tf.NewTask("A", func() { - fmt.Println("A") - }), - tf.NewTask("B", func() { - fmt.Println("B") - }), - tf.NewTask("C", func() { - fmt.Println("C") - }) + // Split input into chunks, one per mapper + var chunks [numMappers]string + splitChunks := func() { + words := strings.Fields(input) + size := (len(words) + numMappers - 1) / numMappers + for i := 0; i < numMappers; i++ { + start := i * size + end := start + size + if end > len(words) { + end = len(words) + } + chunks[i] = strings.Join(words[start:end], " ") + } + } - A1, B1, _ := - tf.NewTask("A1", func() { - fmt.Println("A1") - }), - tf.NewTask("B1", func() { - fmt.Println("B1") - }), - tf.NewTask("C1", func() { - fmt.Println("C1") + executor := gotaskflow.NewExecutor(8, gotaskflow.WithProfiler()) + tf := gotaskflow.NewTaskFlow("word-count") + + // Phase 1: Split input + splitTask := tf.NewTask("split_input", splitChunks) + + // Phase 2: Map — each mapper counts words and partitions by hash + mapTasks := make([]*gotaskflow.Task, numMappers) + for i := 0; i < numMappers; i++ { + idx := i + mapTasks[idx] = tf.NewTask(fmt.Sprintf("map_%d", idx), func() { + localCounts := [numReducers]map[string]int{} + for r := 0; r < numReducers; r++ { + localCounts[r] = make(map[string]int) + } + for _, w := range strings.Fields(chunks[idx]) { + localCounts[hashPartition(w)][w]++ + } + mu.Lock() + for r := 0; r < numReducers; r++ { + mapOutputs[idx][r] = localCounts[r] + } + mu.Unlock() }) - A.Precede(B) - C.Precede(B) - A1.Precede(B) - C.Succeed(A1) - C.Succeed(B1) + } + splitTask.Precede(mapTasks...) - subflow := tf.NewSubflow("sub1", func(sf *gotaskflow.Subflow) { - A2, B2, C2 := - sf.NewTask("A2", func() { - fmt.Println("A2") - }), - sf.NewTask("B2", func() { - fmt.Println("B2") - }), - sf.NewTask("C2", func() { - fmt.Println("C2") - }) - A2.Precede(B2) - C2.Precede(B2) - }) + // Phase 3: Reduce — each reducer aggregates a hash partition + reduceTasks := make([]*gotaskflow.Task, numReducers) + reduceResults := make([]map[string]int, numReducers) + for r := 0; r < numReducers; r++ { + rIdx := r + reduceResults[rIdx] = make(map[string]int) + reduceTasks[rIdx] = tf.NewTask(fmt.Sprintf("reduce_%d", rIdx), func() { + for m := 0; m < numMappers; m++ { + for word, count := range mapOutputs[m][rIdx] { + reduceResults[rIdx][word] += count + } + } + }) + } + // Every mapper feeds every reducer (shuffle) + for _, mt := range mapTasks { + mt.Precede(reduceTasks...) + } - subflow2 := tf.NewSubflow("sub2", func(sf *gotaskflow.Subflow) { - A3, B3, C3 := - sf.NewTask("A3", func() { - fmt.Println("A3") - }), - sf.NewTask("B3", func() { - fmt.Println("B3") - }), - sf.NewTask("C3", func() { - fmt.Println("C3") - // time.Sleep(10 * time.Second) - }) - A3.Precede(B3) - C3.Precede(B3) + // Phase 4: Merge final results + mergeTask := tf.NewTask("merge_results", func() { + final := make(map[string]int) + for r := 0; r < numReducers; r++ { + for w, c := range reduceResults[r] { + final[w] += c + } + } + keys := make([]string, 0, len(final)) + for k := range final { + keys = append(keys, k) + } + sort.Strings(keys) + fmt.Println("Word Count Results:") + for _, w := range keys { + fmt.Printf(" %-12s %d\n", w, final[w]) + } }) + for _, rt := range reduceTasks { + rt.Precede(mergeTask) + } - subflow.Precede(B) - subflow.Precede(subflow2) executor.Run(tf).Wait() - fmt.Println("Print DOT") + if err := tf.Dump(os.Stdout); err != nil { log.Fatal(err) } - fmt.Println("Print Flamegraph") if err := executor.Profile(os.Stdout); err != nil { log.Fatal(err) } diff --git a/image/condition.svg b/image/condition.svg index a8c09c7..ae17d78 100644 --- a/image/condition.svg +++ b/image/condition.svg @@ -1,185 +1,124 @@ - - - - -G - + + + +event-processor + -cluster_sub2 - -sub2 +cluster_critical_path + +critical_path -cluster_sub1 - -sub1 +cluster_warning_path + +warning_path - + +cluster_info_path + +info_path + + -A3 - -A3 +ingest + +ingest - + -B3 - -B3 +classify + +classify - + -A3->B3 - - +ingest->classify + + - + -C3 - -C3 +severity_check + +severity_check - + -C3->B3 - - +classify->severity_check + + - + -sub2 - +critical_path + + + + +severity_check->critical_path + + +0 - + -A2 - -A2 +warning_path + + + + +severity_check->warning_path + + +1 - + -B2 - -B2 +info_path + - - -A2->B2 - - + + +severity_check->info_path + + +2 - + -C2 - -C2 - - - -C2->B2 - - +alert + +alert - + -sub1 - +escalate + +escalate - - -A - -A + + +alert->escalate + + - - -B - -B + + +page_oncall + +page_oncall - - -A->B - - + + +escalate->page_oncall + + - - -cond - -cond - - -B->cond - - - - - -cond->sub2 - - -1 - - - -cond->sub1 - - -0 - - - -C - -C - - - -C->B - - - - - -A1 - -A1 - - - -A1->B - - - - - -A1->C - - - - - -B1 - -B1 - - - -B1->C - - - - - -C1 - -C1 - - - \ No newline at end of file + diff --git a/image/desc.svg b/image/desc.svg index b5ec205..019aa23 100644 --- a/image/desc.svg +++ b/image/desc.svg @@ -1,207 +1,145 @@ - - - - -G - - -cluster_sub2 - -sub2 - - -cluster_sub1 - -sub1 - - -cluster_sub in sub - -sub in sub - - + + + +word-count + + -A3 - -A3 +reduce_0 + +reduce_0 - - -B3 - -B3 + + +merge_results + +merge_results - - -A3->B3 - - + + +reduce_0->merge_results + + - - -C3 - -C3 + + +reduce_1 + +reduce_1 - - -C3->B3 - - + + +reduce_1->merge_results + + - + -sub2 - +split_input + +split_input - + -done - -done +map_0 + +map_0 - + + +split_input->map_0 + + + + -sub in sub - +map_1 + +map_1 - - -A2 - -A2 + + +split_input->map_1 + + - - -B2 - -B2 + + +map_2 + +map_2 - + -A2->B2 - - +split_input->map_2 + + - - -C2 - -C2 + + +map_3 + +map_3 - + -C2->B2 - - +split_input->map_3 + + - - -cond - -cond - - - -cond->sub in sub - - -0 - - + -cond->cond - - -1 +map_0->reduce_0 + + - - -sub1 - + + +map_0->reduce_1 + + - - -sub1->sub2 - - - - - -A - -A - - - -B - -B - - + -A->B - - - - - -C - -C +map_1->reduce_0 + + - + -C->B - - - - - -C1 - -C1 +map_1->reduce_1 + + - + -C->C1 - - +map_2->reduce_0 + + - - -C1->sub1 - - - - - -A1 - -A1 - - + -A1->C - - +map_2->reduce_1 + + - - -B1 - -B1 - - + -B1->C - - +map_3->reduce_0 + + + + + +map_3->reduce_1 + + - \ No newline at end of file + diff --git a/image/loop.svg b/image/loop.svg index e0a17f9..81bb89a 100644 --- a/image/loop.svg +++ b/image/loop.svg @@ -1,76 +1,100 @@ - - - - -G - + + + +crawl-pipeline + init - -init + +init - + -while i < 5 - -while i < 5 +fetch_pages + +fetch_pages - + -init->while i < 5 - - +init->fetch_pages + + - + -body - -body +parse + +parse - + -while i < 5->body - - -0 +fetch_pages->parse + + - + -done - -done +validate_quality + +validate_quality - + -while i < 5->done - - -1 +parse->validate_quality + + - + -back - -back +backoff + +backoff + + + +validate_quality->backoff + + +1 + + + +index + +index - + -body->back - - +validate_quality->index + + +0 - - -back->while i < 5 - - -0 + + +backoff->fetch_pages + + +0 + + + +done + +done + + + +index->done + + - \ No newline at end of file + diff --git a/image/simple.svg b/image/simple.svg index ab93a0c..cf1e320 100644 --- a/image/simple.svg +++ b/image/simple.svg @@ -1,138 +1,145 @@ - + + + + + -wordcount - - - -split_input - -split_input - - - -map_0 - -map_0 - - - -split_input->map_0 - - - +word-count + - + map_1 - -map_1 - - - -split_input->map_1 - - - - - -map_2 - -map_2 - - - -split_input->map_2 - - - - - -map_3 - -map_3 - - - -split_input->map_3 - - + +map_1 - + reduce_0 - -reduce_0 + +reduce_0 - - -map_0->reduce_0 - - + + +map_1->reduce_0 + + - + reduce_1 - -reduce_1 - - - -map_0->reduce_1 - - - - - -map_1->reduce_0 - - + +reduce_1 -map_1->reduce_1 - - +map_1->reduce_1 + + + + + +map_2 + +map_2 -map_2->reduce_0 - - +map_2->reduce_0 + + -map_2->reduce_1 - - +map_2->reduce_1 + + + + + +map_3 + +map_3 -map_3->reduce_0 - - +map_3->reduce_0 + + -map_3->reduce_1 - - +map_3->reduce_1 + + - + merge_results - -merge_results + +merge_results -reduce_0->merge_results - - +reduce_0->merge_results + + -reduce_1->merge_results - - +reduce_1->merge_results + + + + + +split_input + +split_input + + + +split_input->map_1 + + + + + +split_input->map_2 + + + + + +split_input->map_3 + + + + + +map_0 + +map_0 + + + +split_input->map_0 + + + + + +map_0->reduce_0 + + + + + +map_0->reduce_1 + + - \ No newline at end of file + diff --git a/image/subflow.svg b/image/subflow.svg index 9f9e776..b9905e7 100644 --- a/image/subflow.svg +++ b/image/subflow.svg @@ -1,171 +1,135 @@ - - - - -G - + + + +ci-pipeline + -cluster_sub2 - -sub2 +cluster_build + +build -cluster_sub1 - -sub1 +cluster_test + +test - + -A3 - -A3 +deploy + +deploy - + -B3 - -B3 +build + - - -A3->B3 - - - - + -C3 - -C3 +test + + + + +build->test + + - + -C3->B3 - - +test->deploy + + - + -sub2 - +compile_frontend + +compile_frontend - - -A2 - -A2 - - - -B2 - -B2 + + +link + +link - + -A2->B2 - - +compile_frontend->link + + - - -C2 - -C2 + + +compile_backend + +compile_backend - + -C2->B2 - - - - - -sub1 - - - - -sub1->sub2 - - +compile_backend->link + + - - -B - -B - - - -sub1->B - - - - - -A - -A + + +compile_common + +compile_common - + -A->B - - +compile_common->link + + + + + +unit_test + +unit_test - + -C - -C +test_report + +test_report - + -C->B - - +unit_test->test_report + + - - -A1 - -A1 + + +integration_test + +integration_test - + -A1->B - - +integration_test->test_report + + + + + +e2e_test + +e2e_test - + -A1->C - - - - - -B1 - -B1 - - - -B1->C - - - - - -C1 - -C1 - - - \ No newline at end of file +e2e_test->test_report + + + + +