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 @@
-
-
-
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 @@
-
-
-
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 @@
-
-
-
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 @@
-
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 @@
-
-
-