WIP: Use rangefunc instead of a Manager

.github/FUNDING.yml

@@ -1 +1 @@
-github: carlmjohnson
+github: earthboundkid

.github/workflows/go.yml

@@ -9,9 +9,9 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
-    - uses: actions/setup-go@v3
+    - uses: actions/setup-go@v5
       with:
-        go-version: '1.21'
+        go-version: 'stable'
         cache: true
     - name: Get dependencies
       run: go mod download

README.md

@@ -1,4 +1,4 @@
-# Flowmatic [![GoDoc](https://pkg.go.dev/badge/github.com/carlmjohnson/flowmatic)](https://pkg.go.dev/github.com/carlmjohnson/flowmatic) [![Coverage Status](https://coveralls.io/repos/github/carlmjohnson/flowmatic/badge.svg)](https://coveralls.io/github/carlmjohnson/flowmatic) [![Go Report Card](https://goreportcard.com/badge/github.com/carlmjohnson/flowmatic)](https://goreportcard.com/report/github.com/carlmjohnson/flowmatic) [![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)
+# Flowmatic [![GoDoc](https://pkg.go.dev/badge/github.com/carlmjohnson/flowmatic)](https://pkg.go.dev/github.com/earthboundkid/flowmatic/v2) [![Coverage Status](https://coveralls.io/repos/github/earthboundkid/flowmatic/badge.svg)](https://coveralls.io/github/earthboundkid/flowmatic) [![Go Report Card](https://goreportcard.com/badge/github.com/earthboundkid/flowmatic)](https://goreportcard.com/report/github.com/earthboundkid/flowmatic) [![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)
 
 ![Flowmatic logo](https://github.com/carlmjohnson/flowmatic/assets/222245/c14936e9-bb35-405b-926e-4cfeb8003439)
 
@@ -261,20 +261,21 @@ task := func(u string) ([]string, error) {
     return getLinks(page), nil
 }
 
+// Process the tasks with as many workers as GOMAXPROCS
+manager := flowmatic.ManageTasks(flowmatic.MaxProcs, task)
+
 // Map from page to links
 // Doesn't need a lock because only the manager touches it
 results := map[string][]string{}
-var managerErr error
 
-// Manager keeps track of which pages have been visited and the results graph
-manager := func(req string, links []string, err error) ([]string, bool) {
-    // Halt execution after the first error
-    if err != nil {
-        managerErr = err
-        return nil, false
-    }
+// Prime the initial queue
+manager.Add("http://example.com/")
+
+// Start execution and track of which pages have been visited
+// and the results graph
+for url, links := range manager.Work() {
     // Save final results in map
-    results[req] = urls
+    results[url] = links
 
     // Check for new pages to scrape
     var newpages []string
@@ -288,18 +289,15 @@ manager := func(req string, links []string, err error) ([]string, bool) {
         // Add placeholder to map to prevent double scraping
         results[link] = nil
     }
-    return newpages, true
 }
 
-// Process the tasks with as many workers as GOMAXPROCS
-flowmatic.ManageTasks(flowmatic.MaxProcs, task, manager, "http://example.com/")
 // Check if anything went wrong
-if managerErr != nil {
-    fmt.Println("error", managerErr)
+if manager.HasErr() {
+    fmt.Println("error", manager.Err())
 }
 ```
 
-Normally, it is very difficult to keep track of concurrent code because any combination of events could occur in any order or simultaneously, and each combination has to be accounted for by the programmer. `flowmatic.ManageTasks` makes it simple to write concurrent code because everything follows a simple rule: **tasks happen concurrently; the manager runs serially**.
+Normally, it is very difficult to keep track of concurrent code because any combination of events could occur in any order or simultaneously, and each combination has to be accounted for by the programmer. `flowmatic.Manage` makes it simple to write concurrent code because everything follows a simple rule: **tasks happen concurrently; the manager runs serially**.
 
 Centralizing control in the manager makes reasoning about the code radically simpler. When writing locking code, if you have M states and N methods, you need to think about all N states in each of the M methods, giving you an M × N code explosion. By centralizing the logic, the N states only need to be considered in one location: the manager.
 

all.go

@@ -2,6 +2,7 @@ package flowmatic
 
 import (
 	"context"
+	"errors"
 )
 
 // All runs each task concurrently
@@ -33,3 +34,43 @@ func All(ctx context.Context, tasks ...func(context.Context) error) error {
 		return nil
 	})
 }
+
+// eachN starts numWorkers concurrent workers (or GOMAXPROCS workers if numWorkers < 1)
+// and starts a task for each number from 0 to numItems.
+// Errors returned by a task do not halt execution,
+// but are joined into a multierror return value.
+// If a task panics during execution,
+// the panic will be caught and rethrown in the parent Goroutine.
+func eachN(numWorkers, numItems int, task func(int) error) error {
+	type void struct{}
+	inch, ouch := TaskPool(numWorkers, func(pos int) (void, error) {
+		return void{}, task(pos)
+	})
+	var (
+		panicVal any
+		errs     []error
+	)
+	_ = Do(
+		func() error {
+			for i := 0; i < numItems; i++ {
+				inch <- i
+			}
+			close(inch)
+			return nil
+		},
+		func() error {
+			for r := range ouch {
+				if r.Panic != nil && panicVal == nil {
+					panicVal = r.Panic
+				}
+				if r.Err != nil {
+					errs = append(errs, r.Err)
+				}
+			}
+			return nil
+		})
+	if panicVal != nil {
+		panic(panicVal)
+	}
+	return errors.Join(errs...)
+}

all_example_test.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/carlmjohnson/flowmatic"
+	"github.com/earthboundkid/flowmatic/v2"
 )
 
 func ExampleAll() {

do.go

@@ -20,18 +20,15 @@ func Do(tasks ...func() error) error {
 	var wg sync.WaitGroup
 	errch := make(chan result, len(tasks))
 
-	wg.Add(len(tasks))
-	for i := range tasks {
-		fn := tasks[i]
-		go func() {
-			defer wg.Done()
+	for _, fn := range tasks {
+		wg.Go(func() {
 			defer func() {
 				if panicVal := recover(); panicVal != nil {
 					errch <- result{panic: panicVal}
 				}
 			}()
 			errch <- result{err: fn()}
-		}()
+		})
 	}
 	go func() {
 		wg.Wait()

do_example_test.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/carlmjohnson/flowmatic"
+	"github.com/earthboundkid/flowmatic/v2"
 )
 
 func ExampleDo() {

do_test.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/carlmjohnson/flowmatic"
+	"github.com/earthboundkid/flowmatic/v2"
 )
 
 func TestDo_err(t *testing.T) {

each.go

@@ -2,50 +2,43 @@ package flowmatic
 
 import (
 	"errors"
+	"iter"
 )
 
 // Each starts numWorkers concurrent workers (or GOMAXPROCS workers if numWorkers < 1)
-// and processes each item as a task.
+// and processes each item yielded by seq as a task.
 // Errors returned by a task do not halt execution,
 // but are joined into a multierror return value.
 // If a task panics during execution,
 // the panic will be caught and rethrown in the parent Goroutine.
-func Each[Input any](numWorkers int, items []Input, task func(Input) error) error {
-	return eachN(numWorkers, len(items), func(pos int) error {
-		return task(items[pos])
-	})
-}
-
-// eachN starts numWorkers concurrent workers (or GOMAXPROCS workers if numWorkers < 1)
-// and starts a task for each number from 0 to numItems.
-// Errors returned by a task do not halt execution,
-// but are joined into a multierror return value.
-// If a task panics during execution,
-// the panic will be caught and rethrown in the parent Goroutine.
-func eachN(numWorkers, numItems int, task func(int) error) error {
+func Each[Input any](numWorkers int, seq iter.Seq[Input], task func(Input) error) error {
 	type void struct{}
-	inch, ouch := TaskPool(numWorkers, func(pos int) (void, error) {
-		return void{}, task(pos)
+
+	inch, ouch := TaskPool(numWorkers, func(in Input) (void, error) {
+		return void{}, task(in)
 	})
+
 	var (
 		panicVal any
 		errs     []error
 	)
+
 	_ = Do(
 		func() error {
-			for i := 0; i < numItems; i++ {
-				inch <- i
+			defer close(inch)
+
+			for in := range seq {
+				inch <- in
 			}
-			close(inch)
 			return nil
 		},
 		func() error {
 			for r := range ouch {
 				if r.Panic != nil && panicVal == nil {
 					panicVal = r.Panic
 				}
-				if r.Err != nil {
-					errs = append(errs, r.Err)
+				if err := r.Err; err != nil {
+					errs = append(errs, err)
 				}
 			}
 			return nil

each_example_test.go

@@ -2,17 +2,18 @@ package flowmatic_test
 
 import (
 	"fmt"
+	"slices"
 	"time"
 
-	"github.com/carlmjohnson/flowmatic"
+	"github.com/earthboundkid/flowmatic/v2"
 )
 
 func ExampleEach() {
-	times := []time.Duration{
+	times := slices.Values([]time.Duration{
 		50 * time.Millisecond,
 		100 * time.Millisecond,
 		200 * time.Millisecond,
-	}
+	})
 	start := time.Now()
 	err := flowmatic.Each(3, times, func(d time.Duration) error {
 		time.Sleep(d)

each_test.go

@@ -2,15 +2,16 @@ package flowmatic_test
 
 import (
 	"errors"
+	"slices"
 	"testing"
 
-	"github.com/carlmjohnson/flowmatic"
+	"github.com/earthboundkid/flowmatic/v2"
 )
 
 func TestEach_err(t *testing.T) {
 	a := errors.New("a")
 	b := errors.New("b")
-	errs := flowmatic.Each(1, []int{1, 2, 3}, func(i int) error {
+	errs := flowmatic.Each(1, slices.Values([]int{1, 2, 3}), func(i int) error {
 		switch i {
 		case 1:
 			return a

go.mod

@@ -1,5 +1,5 @@
-module github.com/carlmjohnson/flowmatic
+module github.com/earthboundkid/flowmatic/v2
 
-go 1.21
+go 1.25
 
-require github.com/carlmjohnson/deque v0.23.1
+require github.com/earthboundkid/deque/v2 v2.24.2

go.sum

@@ -1,2 +1,2 @@
-github.com/carlmjohnson/deque v0.23.1 h1:X2HOJM9xcglY03deMZ0oZ1V2xtbqYV7dJDnZiSZN4Ak=
-github.com/carlmjohnson/deque v0.23.1/go.mod h1:LF5NJjICBrEOPx84pxPL4nCimy5n9NQjxKi5cXkh+8U=
+github.com/earthboundkid/deque/v2 v2.24.2 h1:U0vh6utzBx922tezr53ryt2tOIJ1GUKMABmYpxCgV48=
+github.com/earthboundkid/deque/v2 v2.24.2/go.mod h1:k/HnjdCUwuMdqNzbS2exS37GEXJBzHSpNBVppL/nEHg=