concurrent

Documentation

Index

• Variables
• type Dispatchable
• type Dispatcher
  • func NewDispatcher[T, R any](ctx context.Context, numWorker int, disp Dispatchable[T, R]) *Dispatcher[T, R]
  • func (d *Dispatcher[T, R]) Process(task T) bool
  • func (d *Dispatcher[T, R]) Quit()
• type Listener
  • func (l *Listener) Close() error
  • func (l *Listener) Signal() <-chan Signal
• type Signal
• type Signaller
  • func NewSignaller() *Signaller
  • func (s *Signaller) Listener() (*Listener, error)
  • func (s *Signaller) Retire()
  • func (s *Signaller) Send(sig Signal) error
  • func (s *Signaller) SetLatency(d time.Duration)

Variables

var (
	ErrSigInactive   = errors.New("signaller inactive")
	ErrSigNoListener = errors.New("no signal listener")
)

Error codes

Types

type Dispatchable

type Dispatchable[T, R any] interface {

	// Worker using channels to read task and write results.
	Worker(ctx context.Context, n int, taskCh chan T, resCh chan R)

	// Eval receives results from workers
	Eval(result R) bool
}

Dispatchable interface

type Dispatcher

type Dispatcher[T, R any] struct {
	// contains filtered or unexported fields
}

Dispatcher managing worker go-routines

func NewDispatcher

func NewDispatcher[T, R any](ctx context.Context, numWorker int, disp Dispatchable[T, R]) *Dispatcher[T, R]

NewDispatcher runs a new dispatcher with given number of workers and a Dispatchanle implementation.

func (*Dispatcher[T, R]) Process

func (d *Dispatcher[T, R]) Process(task T) bool

Process a task. Returns false if the dispatcher is closed.

func (*Dispatcher[T, R]) Quit

func (d *Dispatcher[T, R]) Quit()

Quit dispatcher run

type Listener

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

Listener for signals managed by Signaller

func (*Listener) Close

func (l *Listener) Close() error

Close listener: This more an announcement than an operation as the channel is not closed immediately. It is possible for the listener to receive some more signals before it actually closes.

func (*Listener) Signal

func (l *Listener) Signal() <-chan Signal

Signal returns the channel from which to read the signal. If the returned signal is nil, no further signals will be received on this listener and the select-loop MUST terminate.

type Signal

type Signal interface{}

Signal can be any object (intrinsic or custom); it is the responsibility of the sender and the receiver of signals to handle them accordingly.

A signal in this context is a stand-alone unit of information. Its "meaning" does neither depend on other signals nor on their sequence. Its only purpose is to communicate state changes to listeners instead of sharing the state globally via memory reference ("Do not communicate by sharing memory; instead, share memory by communicating." -- https://go.dev/doc/effective_go.html)

type Signaller

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

Signaller dispatches signals to multiple concurrent listeners. The sequence in which listeners are served is stochastic.

In highly concurrent environments with a lot of signals the sequence of signals seen by a listener can vary. This is due to the fact that a signal gets dispatched in a Go routine, so the next signal can be dispatched before a listener got the first one if the second Go routine handles the listener earlier. It is therefore mandatory that received signals from a listener get handled in a Go routine as well to keep latency low. If a listener violates that promise, it got removed from the list.

func NewSignaller

func NewSignaller() *Signaller

NewSignaller instantiates a new signal manager:

func (*Signaller) Listener

func (s *Signaller) Listener() (*Listener, error)

Listener returns a new channel to listen on each time it is called. Function interested in listening should get the channel, start the for/select loop and drop the channel if the loop terminates. Requesting an listener and than not reading from it will block all other listeners of the signaller.

func (*Signaller) Retire

func (s *Signaller) Retire()

Retire a signaller: This will terminate the dispatch loop for signals; no further send operations are supported. A retired signaller cannot be re-activated. Running dispatches will not be interrupted.

func (*Signaller) Send

func (s *Signaller) Send(sig Signal) error

Send a signal to be dispatched to all listeners.

func (*Signaller) SetLatency

func (s *Signaller) SetLatency(d time.Duration)

SetLatency sets the max latency for listener. A listener is removed from the list if it violates this policy.