README

go-waitgroup

How to use

An package that allows you to use the constructs of a sync.WaitGroup to create a pool of goroutines and control the concurrency.

Using it is just like a normal sync.WaitGroup. The only difference is the initialisation. When you use waitgroup.NewWaitGroup, you have the option to specify it's size.

Any int which is bigger than 0 will limit the number of concurrent goroutines. If you specify -1 or 0, all goroutines will run at once (just like a plain sync.WaitGroup).

package main

import (
    "fmt"
    "net/http"

    "github.com/pieterclaerhout/go-waitgroup"
)

func main() {
    
    urls := []string{
        "https://www.easyjet.com/",
        "https://www.skyscanner.de/",
        "https://www.ryanair.com",
        "https://wizzair.com/",
        "https://www.swiss.com/",
    }

    wg := waitgroup.NewWaitGroup(3)

	for _, url := range urls {
		wg.BlockAdd()
		go func(url string) {
			defer wg.Done()
			fmt.Printf("%s: checking\n", url)
			res, err := http.Get(url)
			if err != nil {
				fmt.Println("Error: %v")
			} else {
				defer res.Body.Close()
				fmt.Printf("%s: result: %v\n", url, err)
			}
		}(url)
	}

    wg.Wait()
    fmt.Println("Finished")

}

Using closures

There is also a way to use function closures to make it even more readable:

package main

import (
	"fmt"
	"net/http"

	"github.com/pieterclaerhout/go-waitgroup"
)

func main() {

	urls := []string{
		"https://www.easyjet.com/",
		"https://www.skyscanner.de/",
		"https://www.ryanair.com",
		"https://wizzair.com/",
		"https://www.swiss.com/",
	}

	wg := waitgroup.NewWaitGroup(3)

	for _, url := range urls {

		urlToCheck := url
		wg.Add(func() {
			fmt.Printf("%s: checking\n", urlToCheck)
			res, err := http.Get(urlToCheck)
			if err != nil {
				fmt.Println("Error: %v")
			} else {
				defer res.Body.Close()
				fmt.Printf("%s: result: %v\n", urlToCheck, err)
			}
		})

	}

	wg.Wait()
	fmt.Println("Finished")

}

Handling errors

If you want to handle errors, there is also an ErrorGroup. This uses the same principles as a normal WaitGroup with a small twist.

First of all, you can only add functions which returns just an error.

Second, as soon as one of the queued items fail, the rest will be cancelled:

package main

import (
	"context"
	"fmt"
	"os"

	"github.com/pieterclaerhout/go-waitgroup"
)

func main() {

	ctx := context.Background()

	wg, ctx := waitgroup.NewErrorGroup(ctx, tc.size)
	if err != nil {
		fmt.Println("Error: %v")
		os.Exit(1)
	}

	wg.Add(func() error {
		return nil
	})

	wg.Add(func() error {
		return errors.New("An error occurred")
	})

	if err := wg.Wait(); err != nil {
		fmt.Println("Error: %v")
		os.Exit(1)
	}

}

You can also add multiple functions in one step:

package main

import (
	"context"
	"errors"
	"fmt"
	"os"

	"github.com/pieterclaerhout/go-waitgroup"
)

func main() {

	ctx := context.Background()

	wg, ctx := waitgroup.NewErrorGroup(ctx, tc.size)
	if err != nil {
		fmt.Println("Error: %v")
		os.Exit(1)
	}

	wg.Add(
		func() error {
			return nil
		},
		func() error {
			return errors.New("An error occurred")
		},
	)

	if err := wg.Wait(); err != nil {
		fmt.Println("Error: %v")
		os.Exit(1)
	}

}

Documentation

Index

• type ErrorGroup
  • func NewErrorGroup(ctx context.Context, size int) (*ErrorGroup, context.Context)
  • func (g *ErrorGroup) Add(closures ...func() error)
  • func (g *ErrorGroup) Wait() error
• type WaitGroup
  • func NewWaitGroup(size int) *WaitGroup
  • func (wg *WaitGroup) Add(closures ...func())
  • func (wg *WaitGroup) BlockAdd()
  • func (wg *WaitGroup) Done()
  • func (wg *WaitGroup) PendingCount() int64
  • func (wg *WaitGroup) Wait()

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ErrorGroup

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

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

func NewErrorGroup

func NewErrorGroup(ctx context.Context, size int) (*ErrorGroup, context.Context)

NewErrorGroup returns a new ErrorGroup instance

func (*ErrorGroup) Add

func (g *ErrorGroup) Add(closures ...func() error)

Add calls the given function in a new goroutine.

The first call to return a non-nil error cancels the group; its error will be returned by Wait.

func (*ErrorGroup) Wait

func (g *ErrorGroup) Wait() error

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

type WaitGroup

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

WaitGroup implements a simple goruntine pool.

func NewWaitGroup

func NewWaitGroup(size int) *WaitGroup

NewWaitGroup creates a waitgroup with a specific size (the maximum number of goroutines to run at the same time). If you use -1 as the size, all items will run concurrently (just like a normal sync.WaitGroup)

func (*WaitGroup) Add

func (wg *WaitGroup) Add(closures ...func())

Add add the function close to the waitgroup

func (*WaitGroup) BlockAdd

func (wg *WaitGroup) BlockAdd()

BlockAdd pushes ‘one’ into the group. Blocks if the group is full.

func (*WaitGroup) Done

func (wg *WaitGroup) Done()

Done pops ‘one’ out the group.

func (*WaitGroup) PendingCount

func (wg *WaitGroup) PendingCount() int64

PendingCount returns the number of pending operations

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()

Wait waiting the group empty