msync

README

msync

This repository defines a library of Go types to help in the management of concurrency.

Packages

• throttle a throttle for concurrent function calls (package docs)
• trigger an edge-triggered selectable condition variable (package docs)

Documentation

Overview

Package msync defines some helpful types for managing concurrency.

Index

• Variables
• type Collector
  • func Collect[T any](ch chan T) *Collector[T]
  • func (c *Collector[T]) Close() error
  • func (c *Collector[T]) Recv() <-chan T
  • func (c *Collector[T]) Send(ctx context.Context, v T) error
• type Flag
  • func NewFlag[T any]() *Flag[T]
  • func (f *Flag[T]) Ready() <-chan T
  • func (f *Flag[T]) Set(v T) bool
• type Linked
  • func (lv *Linked[T]) Get() T
  • func (lv *Linked[T]) StoreCond(v T) bool
  • func (lv *Linked[T]) Validate() bool
• type Value
  • func NewValue[T any](init T) *Value[T]
  • func (v *Value[T]) Get() T
  • func (v *Value[T]) LoadLink(lv *Linked[T]) *Linked[T]
  • func (v *Value[T]) Set(newValue T)
  • func (v *Value[T]) Wait() <-chan T

Variables

var ErrClosed = errors.New("collector is closed")

ErrClosed is the sentinel error reported by a collector that is closed before a value could be delivered.

Types

type Collector

type Collector[T any] struct {
	// contains filtered or unexported fields
}

A Collector manages a fan-in channel shared by multiple readers and writers. It wraps and behaves like a normal buffered channel, but when c is closed any pending writes are safely terminated and report errors rather than panicking.

func Collect

func Collect[T any](ch chan T) *Collector[T]

Collect returns a new collector wrapping the specified channel.

The collector takes ownership of ch, and the caller must not access ch except via the methods of the Collector. The channel may be buffered or unbuffered.

func (*Collector[T]) Close

func (c *Collector[T]) Close() error

Close closes the collector, which closes the receiver and causes any pending sends to fail. If c is already closed, Close returns ErrClosed. Close can be called repeatedly, but from at most one goroutine at a time.

func (*Collector[T]) Recv

func (c *Collector[T]) Recv() <-chan T

Recv returns a channel to which sent values are delivered. The returned channel is closed when c is closed. After c is closed, Recv returns a nil channel.

func (*Collector[T]) Send

func (c *Collector[T]) Send(ctx context.Context, v T) error

Send sends v to the collector. It blocks until v is delivered, c closes, or ctx ends. If c closes or ctx ends before v is sent, Send reports an error; otherwise Send returns nil.

type Flag

type Flag[T any] struct {
	// contains filtered or unexported fields
}

A Flag is a level-triggered single-value buffer shared by a producer and a consumer. A producer calls Flag.Set to make a value available, and a consumer calls Flag.Ready to obtain a channel which delivers the most recently-sent value.

Setting value on the flag does not block: Once a value is set, additional values are discarded until the buffered value is consumed.

The Flag.Ready method returns a channel that delivers buffered values to the consumer. Once a value has been consumed, the buffer is empty and the producer can send another.

func NewFlag

func NewFlag[T any]() *Flag[T]

NewFlag constructs a new empty flag.

func (*Flag[T]) Ready

func (f *Flag[T]) Ready() <-chan T

Ready returns a channel that delivers a value when one is available. Once a value is received, further reads on the channel will block until the flag is Set again.

func (*Flag[T]) Set

func (f *Flag[T]) Set(v T) bool

Set buffers or discards v, and reports whether v was buffered (true) or discarded (false). Set does not block.

type Linked

type Linked[T any] struct {
	// contains filtered or unexported fields
}

Linked is a snapshot of a Value acquired by a call to its LoadLink method.

A linked snapshot may be "valid" or "invalid". It is "valid" if a call to its StoreCond method could succeed at some point in the future; otherwise it is "invalid". A valid snapshot may become invalid, but an invalid snapshot is permanently so.

func (*Linked[T]) Get

func (lv *Linked[T]) Get() T

Get returns the current contents of the snapshot.

func (*Linked[T]) StoreCond

func (lv *Linked[T]) StoreCond(v T) bool

StoreCond attempts to update the linked Value with v, and reports whether doing so succeeded. Once StoreCond has been called, lv is invalid.

StoreCond succeeds if no successful StoreCond or Set operation has been applied to the underlying Value since the LoadLink that initialized lv.

func (*Linked[T]) Validate

func (lv *Linked[T]) Validate() bool

Validate reports whether a call to StoreCond would have succeeded given the current state of lv. When Validate reports true, it means lv was valid at the time of the call; it may have become invalid by the time the caller receives the result. If Validate reports false, lv is forever invalid.

type Value

type Value[T any] struct {
	// contains filtered or unexported fields
}

A Value is a mutable container for a single value of type T that can be concurrently accessed by multiple goroutines. A zero Value is ready for use, but must not be copied after its first use.

The Value takes ownership of the value in its custody. In particular, if a pointer or a slice is stored in a Value, the caller must not modify the contents without separate synchronization.

func NewValue

func NewValue[T any](init T) *Value[T]

NewValue creates a new Value with the given initial value.

func (*Value[T]) Get

func (v *Value[T]) Get() T

Get returns the current value stored in v.

func (*Value[T]) LoadLink

func (v *Value[T]) LoadLink(lv *Linked[T]) *Linked[T]

LoadLink links a view of the current value of v. If lv == nil, a new linked value is allocated and returned. Otherwise, the contents of *lv are replaced and lv is returned.

func (*Value[T]) Set

func (v *Value[T]) Set(newValue T)

Set updates the value stored in v to newValue. Calling Set also wakes any goroutines that are blocked in the Wait method. Set invalidates any linked snapshots open on v, even if the target value does not change.

func (*Value[T]) Wait

func (v *Value[T]) Wait() <-chan T

Wait returns a new channel that blocks until the value of v changes, then delivers the current value of v. Each call to Wait returns a new channel. Once a value has been delivered to the channel, it is closed.

If multiple goroutines set v concurrently, the channel will deliver the value from one of them, but not necessarily the first.