proc

README

proc

Goroutine coordinator and shutdown manager.

proc.Manager is not a general-purpose replacement for sync.WaitGroup. It is designed to simplify shutdown management and protect running code from being interrupted.

Key properties

• Worker registration (Add) is gated by an RWMutex shared with Stop, preventing Add from racing with Wait.

• Wait is not analogous to sync.WaitGroup's Wait: it will not return even if all workers have reported completion but the Manager is still active (not stopped).

• Add may be called at any time; if the manager is already stopped, the returned Worker is disabled and acts as a no-op.

• Add can be used in synchronous code as well.

A Manager has two phases: active and stopped.
Workers may be added while the manager is active.
Calling Stop transitions the manager into the stopped state and signals all registered workers to stop.

Use cases

type MyService struct {
    // ...
    proc proc.Manager
}

func New(...) *MyService {
    s := new(MyService)
    
    // ...
    
    go s.doSomething(s.proc.Add())
    // or
    s.proc.Go(s.doSomething)
    
    return s
}

func (s *MyService) doSomething(w proc.Worker) {
    defer w.Complete()
    
    // ...
    
    select {
    case <-w.Done(): // like context
        return
    default:
    }
    
    // ...
}

func (s *MyService) Close() error {
    return s.proc.Close(func() error {
        // close resources
    })
    
    // Close is a shorthand for
    // s.proc.Stop()
    // s.proc.Wait()
    // return s.proc.Once(func() error { ... })
    
    // there are also WaitContext/CloseContext and 
    // WaitTimeout/CloseTimeout variants
}

func (s *MyService) NormalMethodThatMustFinish() error {
    w := s.proc.Add()
    if w.Disabled() {
        return errors.New("service closed")
    }
    defer w.Complete()
    
    // ...
}

See GoDoc for reference

Documentation

Overview

Package proc provides a small work/goroutine coordination helper for graceful shutdown.

Index

• Variables
• func Term() <-chan struct{}
• func TermContext() context.Context
• type Manager
  • func (m *Manager) Add() Worker
  • func (m *Manager) Close(fn func() error) error
  • func (m *Manager) CloseContext(ctx context.Context, fn func() error) error
  • func (m *Manager) CloseTimeout(duration time.Duration, fn func() error) error
  • func (m *Manager) Context(parent context.Context) context.Context
  • func (m *Manager) Count() int
  • func (m *Manager) Done() <-chan struct{}
  • func (m *Manager) Go(fn func())
  • func (m *Manager) Once(fn func() error) (err error)
  • func (m *Manager) Stop()
  • func (m *Manager) Wait()
  • func (m *Manager) WaitContext(ctx context.Context) error
  • func (m *Manager) WaitTimeout(duration time.Duration) error
  • func (m *Manager) Watch(ch <-chan struct{})
  • func (m *Manager) WatchContext(ctx context.Context)
• type Worker
  • func (w Worker) Add() Worker
  • func (w Worker) Complete()
  • func (w Worker) Context(parent context.Context) context.Context
  • func (w Worker) Disabled() bool
  • func (w Worker) Done() <-chan struct{}
  • func (w Worker) Stopped() bool

Variables

var ErrTimeout = errors.New("timed out")

Functions

func Term

func Term() <-chan struct{}

Term returns a channel that is closed when one of the stop signals is received: os.Interrupt, syscall.SIGTERM, syscall.SIGHUP.

Signal notifiers are not initialized until Term or TermContext are called.

func TermContext

func TermContext() context.Context

TermContext returns a context that is canceled when one of the stop signals is received: os.Interrupt, syscall.SIGTERM, syscall.SIGHUP.

Signal notifiers are not initialized until Term or TermContext are called.

Types

type Manager

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

Manager coordinates graceful shutdown of workers and goroutines. The zero value is ready to use. A Manager must not be copied after first use.

func (*Manager) Add

func (m *Manager) Add() Worker

Add registers a unit of work with the Manager and returns a Worker handle. Users of the returned Worker must call Complete when their work is finished.

Add is safe to call even if the Manager is already stopped and its Wait method has already been called.

If the Manager is already stopped, the returned Worker is disabled: its methods will return closed channels and canceled contexts.

func (*Manager) Close

func (m *Manager) Close(fn func() error) error

Close is a shorthand for Stop, Wait, and Once. See mentioned methods for details.

The provided fn can be nil.

func (*Manager) CloseContext

func (m *Manager) CloseContext(ctx context.Context, fn func() error) error

CloseContext is a shorthand for Stop, WaitContext and Once. See mentioned methods for details.

Once is called regardless of the error returned by WaitContext. The error returned by CloseContext is either from Once or from WaitContext in that particular order.

The provided fn can be nil.

func (*Manager) CloseTimeout

func (m *Manager) CloseTimeout(duration time.Duration, fn func() error) error

CloseTimeout is a shorthand for Stop, WaitTimeout and Once. See mentioned methods for details.

Once is called regardless of the error returned by WaitTimeout. The error returned by CloseTimeout is either from Once or from WaitTimeout in that particular order.

The provided fn can be nil.

func (*Manager) Context

func (m *Manager) Context(parent context.Context) context.Context

Context returns a context that is canceled when the Manager is stopped, or when the parent context is canceled or expires.

func (*Manager) Count

func (m *Manager) Count() int

Count returns the number of active workers.

func (*Manager) Done

func (m *Manager) Done() <-chan struct{}

Done returns a channel that is closed when the Manager is stopped.

func (*Manager) Go

func (m *Manager) Go(fn func())

Go calls fn in a new goroutine and increases the number of active workers. When fn returns, the number of active workers is decreased. If the Manager is already stopped, fn is not called.

func (*Manager) Once

func (m *Manager) Once(fn func() error) (err error)

Once calls the provided fn exactly once. Subsequent calls to Once are no-ops. If fn is nil, Once ignores it but still counts as performed.

func (*Manager) Stop

func (m *Manager) Stop()

Stop signals all workers to stop.

func (*Manager) Wait

func (m *Manager) Wait()

Wait waits until the Manager is stopped and then until all workers report completion.

Wait is not analogous to sync.WaitGroup's Wait, it will not return even if all the workers report completion but the Manager is still in active state (not stopped).

func (*Manager) WaitContext

func (m *Manager) WaitContext(ctx context.Context) error

WaitContext waits until Manager is stopped and then until all workers report completion. If the provided context is canceled or expires during any of these steps, WaitContext returns ctx.Err().

WaitContext is not analogous to sync.WaitGroup's Wait, it will not return even if all the workers report completion but the Manager is still in active state (not stopped).

WaitContext may leak a goroutine if ctx is done and the Manager is never stopped.

func (*Manager) WaitTimeout

func (m *Manager) WaitTimeout(duration time.Duration) error

WaitTimeout waits until Manager is stopped and then until all workers report completion. If the provided timeout expires during any of these steps, WaitTimeout returns ErrTimeout.

WaitTimeout is not analogous to sync.WaitGroup's Wait, it will not return even if all the workers report completion but the Manager is still in active state (not stopped).

WaitTimeout may leak a goroutine if the timeout expires and the Manager is never stopped.

func (*Manager) Watch

func (m *Manager) Watch(ch <-chan struct{})

Watch stops the Manager when the provided channel is closed or a value is received from it.

func (*Manager) WatchContext

func (m *Manager) WatchContext(ctx context.Context)

WatchContext stops the Manager when the provided ctx is done.

type Worker

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

Worker represents a registered unit of work tied to a Manager.

func (Worker) Add

func (w Worker) Add() Worker

Add registers another unit of work in the same Manager.

func (Worker) Complete

func (w Worker) Complete()

Complete reports that the unit of work is finished. It must be called exactly once. If the Worker was created from a stopped Manager, Complete is a no-op.

func (Worker) Context

func (w Worker) Context(parent context.Context) context.Context

Context returns a context that is canceled when the worker is stopped or the provided parent is canceled or expires.

If the Worker was created from an already stopped Manager, Context returns a canceled context.

func (Worker) Disabled

func (w Worker) Disabled() bool

Disabled reports whether the Worker was created from a stopped Manager. It only checks a single pointer and is a bit faster than Stopped.

func (Worker) Done

func (w Worker) Done() <-chan struct{}

Done returns a channel that is closed when the worker should stop its work. If the Worker was created from an already stopped Manager, Done returns a closed channel.

func (Worker) Stopped

func (w Worker) Stopped() bool

Stopped reports whether the Worker is stopped.