semgroup

README

semgroup

semgroup provides synchronization and error propagation, for groups of goroutines working on subtasks of a common task. It uses a weighted semaphore implementation to make sure that only a number of maximum tasks can be run at any time.

Unlike golang.org/x/sync/errgroup, it doesn't return the first non-nil error, rather it accumulates all errors and returns a set of errors, allowing each task to fullfil their task.

Install

go get github.com/fatih/semgroup

Example

With no errors:

package main

import (
	"context"
	"fmt"

	"github.com/fatih/semgroup"
)

func main() {
	const maxWorkers = 2
	s := semgroup.NewGroup(context.Background(), maxWorkers)

	visitors := []int{5, 2, 10, 8, 9, 3, 1}

	for _, v := range visitors {
		v := v

		s.Go(func() error {
			fmt.Println("Visits: ", v)
			return nil
		})
	}

	// Wait for all visits to complete. Any errors are accumulated.
	if err := s.Wait(); err != nil {
		fmt.Println(err)
	}

	// Output:
	// Visits: 2
	// Visits: 10
	// Visits: 8
	// Visits: 9
	// Visits: 3
	// Visits: 1
	// Visits: 5
}

With errors:

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/fatih/semgroup"
)

func main() {
	const maxWorkers = 2
	s := semgroup.NewGroup(context.Background(), maxWorkers)

	visitors := []int{1, 1, 1, 1, 2, 2, 1, 1, 2}

	for _, v := range visitors {
		v := v

		s.Go(func() error {
			if v != 1 {
				return errors.New("only one visitor is allowed")
			}
			return nil
		})
	}

	// Wait for all visits to complete. Any errors are accumulated.
	if err := s.Wait(); err != nil {
		fmt.Println(err)
	}

	// Output:
	// 3 error(s) occurred:
	// * only one visitor is allowed
	// * only one visitor is allowed
	// * only one visitor is allowed
}

Documentation

Overview

Package semgroup provides synchronization and error propagation, for groups of goroutines working on subtasks of a common task. It uses a weighted semaphore implementation to make sure that only a number of maximum tasks can be run at any time.

Unlike golang.org/x/sync/errgroup, it doesn't return the first non-nil error, rather it accumulates all errors and returns a set of errors, allowing each task to fullfil their task.

Index

• type Group
  • func NewGroup(ctx context.Context, maxWorkers int64) *Group
  • func (g *Group) Go(f func() error)
  • func (g *Group) Wait() error

Examples

• Group (Parallel)
• Group (WithErrors)

Types

type Group

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

A Group is a collection of goroutines working on subtasks that are part of the same overall task.

Example (Parallel)

This example increases a counter for each visit concurrently, using a SemGroup to block until all the visitors have finished. It only runs 2 tasks at any time.

package main

import (
	"context"
	"fmt"
	"sync"

	"github.com/fatih/semgroup"
)

func main() {
	const maxWorkers = 2
	s := semgroup.NewGroup(context.Background(), maxWorkers)

	var (
		counter int
		mu      sync.Mutex // protects visits
	)

	visitors := []int{5, 2, 10, 8, 9, 3, 1}

	for _, v := range visitors {
		v := v

		s.Go(func() error {
			mu.Lock()
			counter += v
			mu.Unlock()
			return nil
		})
	}

	// Wait for all visits to complete. Any errors are accumulated.
	if err := s.Wait(); err != nil {
		fmt.Println(err)
	}

	fmt.Printf("Counter: %d", counter)

}

Output:

Counter: 38

Example (WithErrors)

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/fatih/semgroup"
)

func main() {
	const maxWorkers = 2
	s := semgroup.NewGroup(context.Background(), maxWorkers)

	visitors := []int{1, 1, 1, 1, 2, 2, 1, 1, 2}

	for _, v := range visitors {
		v := v

		s.Go(func() error {
			if v != 1 {
				return errors.New("only one visitor is allowed")
			}
			return nil
		})
	}

	// Wait for all visits to complete. Any errors are accumulated.
	if err := s.Wait(); err != nil {
		fmt.Println(err)
	}

}

Output:

3 error(s) occurred:
* only one visitor is allowed
* only one visitor is allowed
* only one visitor is allowed

func NewGroup

func NewGroup(ctx context.Context, maxWorkers int64) *Group

NewGroup returns a new Group with the given maximum combined weight for concurrent access.

func (*Group) Go

func (g *Group) Go(f func() error)

On success, returns nil. On failure, returns ctx.Err() and leaves the semaphore unchanged. Any function call to return a non-nil error is accumulated; the accumulated errors will be returned by Wait.

func (*Group) Wait

func (g *Group) Wait() error

Wait blocks until all function calls from the Go method have returned, then returns all accumulated non-nil error (if any) from them.