concurrent

README

concurrent

• concurrent.Map: backport sync.Map for go below 1.9
• concurrent.Executor: goroutine with explicit ownership and cancellable

concurrent.Map

because sync.Map is only available in go 1.9, we can use concurrent.Map to make code portable

m := concurrent.NewMap()
m.Store("hello", "world")
elem, found := m.Load("hello")
// elem will be "world"
// found will be true

concurrent.Executor

executor := concurrent.NewUnboundedExecutor()
executor.Go(func(ctx context.Context) {
    everyMillisecond := time.NewTicker(time.Millisecond)
    for {
        select {
        case <-ctx.Done():
            fmt.Println("goroutine exited")
            return
        case <-everyMillisecond.C:
            // do something
        }
    }
})
time.Sleep(time.Second)
executor.StopAndWaitForever()
fmt.Println("executor stopped")

attach goroutine to executor instance, so that we can

• cancel it by stop the executor with Stop/StopAndWait/StopAndWaitForever
• handle panic by callback: the default behavior will no longer crash your application

Documentation

Index

• Variables
• type Executor
• type Map
  • func NewMap() *Map
• type UnboundedExecutor
  • func NewUnboundedExecutor() *UnboundedExecutor
  • func (executor *UnboundedExecutor) Go(handler func(ctx context.Context))
  • func (executor *UnboundedExecutor) Stop()
  • func (executor *UnboundedExecutor) StopAndWait(ctx context.Context)
  • func (executor *UnboundedExecutor) StopAndWaitForever()

Examples

• UnboundedExecutor.Go
• UnboundedExecutor.Go (Panic)
• UnboundedExecutor.StopAndWaitForever

Variables

var ErrorLogger = log.New(os.Stderr, "", 0)

ErrorLogger is used to print out error, can be set to writer other than stderr

var GlobalUnboundedExecutor = NewUnboundedExecutor()

GlobalUnboundedExecutor has the life cycle of the program itself any goroutine want to be shutdown before main exit can be started from this executor GlobalUnboundedExecutor expects the main function to call stop it does not magically knows the main function exits

var HandlePanic = func(recovered interface{}, funcName string) {
	ErrorLogger.Println(fmt.Sprintf("%s panic: %v", funcName, recovered))
	ErrorLogger.Println(string(debug.Stack()))
}

HandlePanic logs goroutine panic by default

var InfoLogger = log.New(ioutil.Discard, "", 0)

InfoLogger is used to print informational message, default to off

Types

type Executor

type Executor interface {
	// Go starts a new goroutine controlled by the context
	Go(handler func(ctx context.Context))
}

Executor replace go keyword to start a new goroutine the goroutine should cancel itself if the context passed in has been cancelled the goroutine started by the executor, is owned by the executor we can cancel all executors owned by the executor just by stop the executor itself however Executor interface does not Stop method, the one starting and owning executor should use the concrete type of executor, instead of this interface.

type Map

type Map struct {
	sync.Map
}

Map is a wrapper for sync.Map introduced in go1.9

func NewMap

func NewMap() *Map

NewMap creates a thread safe Map

type UnboundedExecutor

type UnboundedExecutor struct {
	HandlePanic func(recovered interface{}, funcName string)
	// contains filtered or unexported fields
}

UnboundedExecutor is a executor without limits on counts of alive goroutines it tracks the goroutine started by it, and can cancel them when shutdown

func NewUnboundedExecutor

func NewUnboundedExecutor() *UnboundedExecutor

NewUnboundedExecutor creates a new UnboundedExecutor, UnboundedExecutor can not be created by &UnboundedExecutor{} HandlePanic can be set with a callback to override global HandlePanic

func (*UnboundedExecutor) Go

func (executor *UnboundedExecutor) Go(handler func(ctx context.Context))

Go starts a new goroutine and tracks its lifecycle. Panic will be recovered and logged automatically, except for StopSignal

Example

package main

import (
	"context"
	"fmt"
	"github.com/modern-go/concurrent"
	"time"
)

func main() {
	executor := concurrent.NewUnboundedExecutor()
	executor.Go(func(ctx context.Context) {
		fmt.Println("abc")
	})
	time.Sleep(time.Second)
}

Output:

abc

Example (Panic)

package main

import (
	"context"
	"fmt"
	"github.com/modern-go/concurrent"
	"time"
)

func main() {
	concurrent.HandlePanic = func(recovered interface{}, funcName string) {
		fmt.Println(funcName)
	}
	executor := concurrent.NewUnboundedExecutor()
	executor.Go(willPanic)
	time.Sleep(time.Second)
}

func willPanic(ctx context.Context) {
	panic("!!!")
}

Output:

github.com/modern-go/concurrent_test.willPanic

func (*UnboundedExecutor) Stop

func (executor *UnboundedExecutor) Stop()

Stop cancel all goroutines started by this executor without wait

func (*UnboundedExecutor) StopAndWait

func (executor *UnboundedExecutor) StopAndWait(ctx context.Context)

StopAndWait cancel all goroutines started by this executor and wait. Wait can be cancelled by the context passed in.

func (*UnboundedExecutor) StopAndWaitForever

func (executor *UnboundedExecutor) StopAndWaitForever()

StopAndWaitForever cancel all goroutines started by this executor and wait until all goroutines exited

Example

package main

import (
	"context"
	"fmt"
	"github.com/modern-go/concurrent"
	"time"
)

func main() {
	executor := concurrent.NewUnboundedExecutor()
	executor.Go(func(ctx context.Context) {
		everyMillisecond := time.NewTicker(time.Millisecond)
		for {
			select {
			case <-ctx.Done():
				fmt.Println("goroutine exited")
				return
			case <-everyMillisecond.C:
				// do something
			}
		}
	})
	time.Sleep(time.Second)
	executor.StopAndWaitForever()
	fmt.Println("executor stopped")
}

Output:

goroutine exited
executor stopped