cogroup

Golang coroutine group

Package cogroup provides an elegant goroutine group with context controls. It's designed to meet the following requirements.

• Tasks can be executed without order
• Group wait command will close the write access to the task queue
• Upstream context can cancel/stop the execution of the tasks
• When the context is canceled, the tasks in queue will be no longer consumed
• Only spawn specified number of goroutines to consume the task queue
• Panic recover for a single task execution
• Custom worker with upstream context and worker id provided.
• Wait will block until tasks are finished or canceled, and return with the queue length

Usage

Start a group and wait till all the tasks are finished.

import (
"context"
"time"
"github.com/devfans/cogroup"
)
func main() {
f := func(context.Context) error {
<-time.After(time.Second)
return nil
}
g := cogroup.Start(context.Background(), 2, 10, false)
for i := 0; i < 10; i++ {
g.Add(f)
}
g.Wait()
}

Start a group and cancel it later.

import (
"context"
"time"
"github.com/devfans/cogroup"
)
func main() {
f := func(ctx context.Context) error {
<-time.After(time.Second)
workerID := cogroup.GetWorkerID(ctx)
println(workerID, " did one task")
return nil
}
ctx, cancel := context.WithCancel(context.Background())
g := cogroup.Start(ctx, 2, 10, false)
go func() {
<-time.After(1 * time.Second)
cancel()
}()
for i := 0; i < 100; i++ {
g.Add(f)
}
println("Tasks left:", g.Wait())
}

Start a group with custom worker

import (
"context"
"time"
"github.com/devfans/cogroup"
)
func main() {
f := func(context.Context) error {
<-time.After(time.Second)
return nil
}
g := cogroup.New(context.Background(), 2, 10, false)
g.StartWithWorker(func(ctx context.Context, i int, f func(context.Context) error {
println("Worker is running with id", i)
f(ctx)
}))
for i := 0; i < 10; i++ {
g.Add(f)
}
g.Wait()
}

Misc

Blog：https://blog.devfans.io/create-a-golang-coroutine-group/