rungroup

README

rungroup/v2

Note: This is the documentation for v2 of the rungroup module. For v1 documentation, see README.v1.md.

A Go module for managing and coordinating concurrent tasks with fine-grained control over cancellation.

rungroup/v2 is a Go module that provides a robust mechanism for managing and coordinating a collection of goroutines. It allows you to start multiple tasks concurrently and coordinate their execution with fine-grained control over cancellation behavior.

Improvements over v1

Version 2 of rungroup addresses several limitations present in v1:

• Non-terminating tasks: v1 automatically triggered cancellation when any task completed. v2 allows tasks to run independently with explicit control over cancellation behavior.
• Error propagation: v2 allows specifying an error when calling Cancel, providing more informative cancellation reasons.
• Zero value safety: v1 panicked when used with a zero value. v2 is safe to use with zero values, automatically initializing with context.Background.
• Nested tasks: v1 prevented starting new tasks within the same group from within a running task. v2 allows nested tasks to be started within the same group.

Installation

To install the rungroup/v2 module, use go get:

go get github.com/goaux/rungroup/v2

Usage

Here's a basic example of how to use the rungroup/v2 module:

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/goaux/rungroup/v2"
	"github.com/goaux/stacktrace/v2"
)

func main() {
	// Create a new group
	gr := rungroup.New(context.Background())

	// Ensure we clean up resources
	defer gr.Close()

	// Add a task that will finish after 150ms
	gr.Go(func(ctx context.Context) {
		select {
		case <-time.After(150 * time.Millisecond):
			fmt.Println("Task 1 completed")
		case <-ctx.Done():
			fmt.Println("Task 1 canceled")
		}
	})

	// Add a task that will finish after 300ms
	// This task will only be canceled if explicitly requested
	gr.Go(func(ctx context.Context) {
		select {
		case <-time.After(300 * time.Millisecond):
			fmt.Println("Task 2 completed")
		case <-ctx.Done():
			fmt.Println("Task 2 canceled")
		}
	})

	// Add a task that will cancel the group on completion
	gr.GoCancelOnFinish(func(ctx context.Context) error {
		time.Sleep(100 * time.Millisecond)
		fmt.Println("Task 3 completed, canceling group")
		return nil
	})

	// Wait for all tasks to complete or be canceled
	err := gr.Wait()
	if err != nil {
		fmt.Printf("Group canceled with error: %v\n", stacktrace.Format(err))
	} else {
		fmt.Println("All tasks completed successfully")
	}
}

API Overview

Creating a Group

// Create a new Group with a parent context
gr := rungroup.New(ctx)

// A zero value is also valid
var gr rungroup.Group

Starting Tasks

// Start a task without automatic cancellation
gr.Go(func(ctx context.Context) {
    // Your task logic here
})

// Start a task that cancels the group when it completes (success or failure)
gr.GoCancelOnFinish(func(ctx context.Context) error {
    // Your task logic here
    return nil
})

// Start a task that cancels the group only on successful completion
gr.GoCancelOnSuccess(func(ctx context.Context) error {
    // Your task logic here
    return nil
})

// Start a task that cancels the group only on error
gr.GoCancelOnError(func(ctx context.Context) error {
    // Your task logic here
    return errors.New("task failed")
})

Controlling the Group

// Cancel the group with a specific error
gr.Cancel(errors.New("operation aborted"))

// Cancel the group with ErrClosed
gr.Close()

// Set a timeout for the group
gr.SetTimeout(5 * time.Second)

// Wait for all tasks to complete
err := gr.Wait()

Resource Management

It's important to call either gr.Close() or gr.Cancel() when a Group is no longer needed to prevent resource leaks. This applies to both Groups created with New() and zero-value Groups.

Thread Safety

rungroup/v2 is designed to be thread-safe. You can start tasks from multiple goroutines, including from within other tasks in the same group.

License

Apache License Version 2.0

Documentation

Overview

Package rungroup provides a way to manage and synchronize concurrent tasks.

Its primary feature is the ability to cancel all goroutines when any one of them completes. This package is particularly useful for scenarios where you need to run multiple operations concurrently, but want to stop all of them as soon as any one operation completes or fails.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/goaux/rungroup"
	"github.com/goaux/timer"
)

func main() {
	// Create a new group with no timeout
	rg := rungroup.New(context.Background())

	// Add tasks to the group
	rg.Go(func(ctx context.Context) error {
		if err := timer.Sleep(ctx, 150*time.Millisecond); err != nil {
			fmt.Println("task1 canceled")
			return err
		}
		fmt.Println("task1 done")
		return nil
	})

	rg.Go(func(ctx context.Context) error {
		if err := timer.Sleep(ctx, 300*time.Millisecond); err != nil {
			fmt.Println("task2 canceled")
			return err
		}
		fmt.Println("task2 done")
		return nil
	})

	// Wait for all tasks to complete or be canceled
	err := rg.Wait()
	if err != nil {
		fmt.Printf("must not error: %v\n", err)
	} else {
		fmt.Println("ok")
	}
}

Output:
task1 done
task2 canceled
ok

Index

• type Group
  • func New(ctx context.Context) *Group
  • func NewTimeout(ctx context.Context, d time.Duration) *Group
  • func (g *Group) Cancel()
  • func (g *Group) Go(task func(context.Context) error)
  • func (g *Group) Wait() error

Examples

• Package
• NewTimeout

Types

type Group

type Group struct {
	// contains filtered or unexported fields
}

Group represents a collection of goroutines working on subtasks that are part of the same overall task. The Group is designed to manage concurrent execution and provides automatic cancellation of all running tasks when any single task completes.

• A Group must be created by New or NewTimeout, zero value must not be used.
• A Group must not be copied after first use.
• A Group must not be reused after calling Wait.

func New

func New(ctx context.Context) *Group

New creates a new Group with the given context. The returned Group's context is canceled when the first task completes (returns), regardless of whether it returns an error or nil. This cancellation triggers the termination of all other running tasks in the group.

func NewTimeout

func NewTimeout(ctx context.Context, d time.Duration) *Group

NewTimeout creates a new Group with the given context and timeout duration. The returned Group's context is canceled when the timeout expires, or when the first task completes (returns), whichever happens first. This cancellation triggers the termination of all other running tasks in the group.

The timeout is implemented as a separate task within the group, ensuring consistent behavior with other tasks.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/goaux/rungroup"
	"github.com/goaux/timer"
)

func main() {
	// Create a new group with a 300-milliseconds timeout
	rg := rungroup.NewTimeout(context.Background(), 300*time.Millisecond)

	// Add tasks to the group
	rg.Go(func(ctx context.Context) error {
		if err := timer.Sleep(ctx, 500*time.Millisecond); err != nil {
			fmt.Println("task1 canceled")
			return err
		}
		fmt.Println("task1 done")
		return nil
	})

	rg.Go(func(ctx context.Context) error {
		if err := timer.Sleep(ctx, 150*time.Millisecond); err != nil {
			fmt.Println("task2 canceled")
			return err
		}
		fmt.Println("task2 done")
		return nil
	})

	// Wait for all tasks to complete or be canceled
	err := rg.Wait()
	if err != nil {
		fmt.Printf("must not error: %v\n", err)
	} else {
		fmt.Println("ok")
	}
}

Output:
task2 done
task1 canceled
ok

func (*Group) Cancel

func (g *Group) Cancel()

Cancel explicitly cancels the Group's context, causing all tasks to be interrupted. This method can be used to manually trigger the cancellation of all running tasks.

func (*Group) Go

func (g *Group) Go(task func(context.Context) error)

Go starts a new goroutine in the Group. The provided function is executed in its own goroutine. If this function completes (either by returning nil or an error), it will trigger the cancellation of the Group's context, causing all other running tasks to be terminated.

Go will panic if called on a Group that has already been used (i.e., after Wait has been called).

func (*Group) Wait

func (g *Group) Wait() error

Wait blocks until all tasks in the Group have completed or been cancelled. It returns the first non-nil error (if any) from any of the tasks. If a task completes without error, Wait will still trigger the cancellation of all other tasks before returning.

If Wait is called without any tasks being started using the Go method, it returns nil immediately. Even in this case, the Group cannot be reused.