sync

Documentation

Overview

Package sync provides synchronization, error propagation, and Context cancelation for groups of goroutines working on subtasks of a common task.

Index

• func Go(ctx context.Context, n int, funcs ...func() error) error

Examples

• Go (JustErrors)
• Go (WithCancellation)

Functions

func Go

func Go(ctx context.Context, n int, funcs ...func() error) error

Go waits for a collection of goroutines to finish. It has almost similar semantics to sync.WaitGroup and/or golang.org/x/sync/errgroup.Group It limits the number of active goroutines to at most n. If n<=0, the limit is set to runtime.NumCPU

It calls each of the given functions in a new goroutine and blocks until the new goroutine can be added without the number of active goroutines in the group exceeding the configured limit.

It also blocks until all function calls have returned, then returns the concated non-nil errors(if any) from them. If any of those functions panic, Go will propagate that panic. Unlike golang.org/x/sync/errgroup.Group errors and panics do not cancel the context.

If callers of Go cancel ctx, it will return after the current executing func has finished.

Example (JustErrors)

JustErrors illustrates the use of a group in place of a sync.WaitGroup to simplify goroutine counting and error handling. This example is derived from the sync.WaitGroup example at https://golang.org/pkg/sync/#example_WaitGroup.

package main

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"github.com/komuw/ong/sync"
)

func main() {
	urls := []string{
		"http://www.example.org/",
		"http://www.example.com/",
		"http://www.nonExistentDomainName.com/",
	}

	funcs := []func() error{}
	for _, url := range urls {
		url := url // https://golang.org/doc/faq#closures_and_goroutines
		funcs = append(
			funcs,
			func() error {
				// Fetch the URL.
				ct, cancel := context.WithTimeout(context.Background(), 4*time.Second)
				defer cancel()

				req, err := http.NewRequestWithContext(ct, http.MethodGet, url, nil)
				if err != nil {
					return err
				}
				resp, err := http.DefaultClient.Do(req)
				if err != nil {
					return err
				}
				defer resp.Body.Close()
				return err
			},
		)
	}

	funcs = append(
		funcs,
		func() error {
			return nil
		},
	)

	err := sync.Go(
		context.Background(),
		2, // limit concurrency to 2 goroutines.
		funcs...,
	)
	fmt.Printf("\n\t err: %v\n\n", err)
}

Example (WithCancellation)

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/komuw/ong/sync"
)

func main() {
	// Read at most four files, with a concurrency of 2, but cancel the processing after 2 seconds.
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	err := sync.Go(
		ctx,
		2,
		func() error {
			_, err := os.ReadFile("/tmp/file1.txt")
			return err
		},
		func() error {
			_, err := os.ReadFile("/tmp/file2.txt")
			return err
		},
		func() error {
			_, err := os.ReadFile("/tmp/file3.txt")
			return err
		},
		func() error {
			_, err := os.ReadFile("/tmp/file4.txt")
			return err
		},
	)
	fmt.Printf("err: %v", err)
}