diodes

package module
v0.0.0-...-c756bfe Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jun 4, 2024 License: Apache-2.0 Imports: 5 Imported by: 67

README

diode

GoDoc

If you have any questions, or want to get attention for a PR or issue please reach out on the #logging-and-metrics channel in the cloudfoundry slack

Diodes are ring buffers manipulated via atomics.

Diodes are optimized for high throughput scenarios where losing data is acceptable. Unlike a channel, a diode will overwrite data on writes in lieu of blocking. A diode does its best to not "push back" on the producer. In other words, invoking Set() on a diode never blocks.

Installation

go get code.cloudfoundry.org/go-diodes

Example: Basic Use

d := diodes.NewOneToOne(1024, diodes.AlertFunc(func(missed int) {
	log.Printf("Dropped %d messages", missed)
}))

// writer
go func() {
	for i := 0; i < 2048; i++ {
		// Warning: Do not use i. By taking the address,
		// you would not get each value
		j := i
		d.Set(diodes.GenericDataType(&j))
	}
}()

// reader
poller := diodes.NewPoller(d)
for {
	i := poller.Next()
	fmt.Println(*(*int)(i))
}

Example: Creating a Concrete Shell

Diodes accept and return diodes.GenericDataType. It is recommended to not use these generic pointers directly. Rather, it is a much better experience to wrap the diode in a concrete shell that accepts the types your program works with and does the type casting for you. Here is an example of how to create a concrete shell for []byte:

type OneToOne struct {
	d *diodes.Poller
}

func NewOneToOne(size int, alerter diodes.Alerter) *OneToOne {
	return &OneToOne{
		d: diodes.NewPoller(diodes.NewOneToOne(size, alerter)),
	}
}

func (d *OneToOne) Set(data []byte) {
	d.d.Set(diodes.GenericDataType(&data))
}

func (d *OneToOne) TryNext() ([]byte, bool) {
	data, ok := d.d.TryNext()
	if !ok {
		return nil, ok
	}

	return *(*[]byte)(data), true
}

func (d *OneToOne) Next() []byte {
	data := d.d.Next()
	return *(*[]byte)(data)
}

Creating a concrete shell gives you the following advantages:

  • The compiler will tell you if you use a diode to read or write data of the wrong type.
  • The type casting syntax in go is not common and should be hidden.
  • It prevents the generic pointer type from escaping in to client code.

Dropping Data

The diode takes an Alerter as an argument to alert the user code to when the read noticed it missed data. It is important to note that the go-routine consuming from the diode is used to signal the alert.

When the diode notices it has fallen behind, it will move the read index to the new write index and therefore drop more than a single message.

There are two things to consider when choosing a diode:

  1. Storage layer
  2. Access layer

Storage Layer

OneToOne

The OneToOne diode is meant to be used by one producing (invoking Set()) go-routine and a (different) consuming (invoking TryNext()) go-routine. It is not thread safe for multiple readers or writers.

ManyToOne

The ManyToOne diode is optimized for many producing (invoking Set()) go-routines and a single consuming (invoking TryNext()) go-routine. It is not thread safe for multiple readers.

It is recommended to have a larger diode buffer size if the number of producers is high. This is to avoid the diode from having to mitigate write collisions (it will call its alert function if this occurs).

Access Layer

Poller

The Poller uses polling via time.Sleep(...) when Next() is invoked. While polling might seem sub-optimal, it allows the producer to be completely decoupled from the consumer. If you require very minimal push back on the producer, then the Poller is a better choice. However, if you require several diodes (e.g. one per connected client), then having several go-routines polling (sleeping) may be hard on the scheduler.

Waiter

The Waiter uses a conditional mutex to manage when the reader is alerted of new data. While this method is great for the scheduler, it does have extra overhead for the producer. Therefore, it is better suited for situations where you have several diodes and can afford slightly slower producers.

Benchmarks

There are benchmarks that compare the various storage and access layers to channels. To run them:

go test -bench=. -run=NoTest

Known Issues

If a diode was to be written to 18446744073709551615+1 times it would overflow a uint64. This will cause problems if the size of the diode is not a power of two (2^x). If you write into a diode at the rate of one message every nanosecond, without restarting your process, it would take you 584.54 years to encounter this issue.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AlertFunc

type AlertFunc func(missed int)

AlertFunc type is an adapter to allow the use of ordinary functions as Alert handlers.

func (AlertFunc) Alert

func (f AlertFunc) Alert(missed int)

Alert calls f(missed)

type Alerter

type Alerter interface {
	Alert(missed int)
}

Alerter is used to report how many values were overwritten since the last write.

type Diode

type Diode interface {
	Set(GenericDataType)
	TryNext() (GenericDataType, bool)
}

Diode is any implementation of a diode.

type GenericDataType

type GenericDataType unsafe.Pointer

GenericDataType is the data type the diodes operate on.

type ManyToOne

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

ManyToOne diode is optimal for many writers (go-routines B-n) and a single reader (go-routine A). It is not thread safe for multiple readers.

func NewManyToOne

func NewManyToOne(size int, alerter Alerter) *ManyToOne

NewManyToOne creates a new diode (ring buffer). The ManyToOne diode is optimzed for many writers (on go-routines B-n) and a single reader (on go-routine A). The alerter is invoked on the read's go-routine. It is called when it notices that the writer go-routine has passed it and wrote over data. A nil can be used to ignore alerts.

func (*ManyToOne) Set

func (d *ManyToOne) Set(data GenericDataType)

Set sets the data in the next slot of the ring buffer.

func (*ManyToOne) TryNext

func (d *ManyToOne) TryNext() (data GenericDataType, ok bool)

TryNext will attempt to read from the next slot of the ring buffer. If there is not data available, it will return (nil, false).

type OneToOne

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

OneToOne diode is meant to be used by a single reader and a single writer. It is not thread safe if used otherwise.

func NewOneToOne

func NewOneToOne(size int, alerter Alerter) *OneToOne

NewOneToOne creates a new diode is meant to be used by a single reader and a single writer. The alerter is invoked on the read's go-routine. It is called when it notices that the writer go-routine has passed it and wrote over data. A nil can be used to ignore alerts.

func (*OneToOne) Set

func (d *OneToOne) Set(data GenericDataType)

Set sets the data in the next slot of the ring buffer.

func (*OneToOne) TryNext

func (d *OneToOne) TryNext() (data GenericDataType, ok bool)

TryNext will attempt to read from the next slot of the ring buffer. If there is no data available, it will return (nil, false).

type Poller

type Poller struct {
	Diode
	// contains filtered or unexported fields
}

Poller will poll a diode until a value is available.

func NewPoller

func NewPoller(d Diode, opts ...PollerConfigOption) *Poller

NewPoller returns a new Poller that wraps the given diode.

func (*Poller) Next

func (p *Poller) Next() GenericDataType

Next polls the diode until data is available or until the context is done. If the context is done, then nil will be returned.

type PollerConfigOption

type PollerConfigOption func(*Poller)

PollerConfigOption can be used to setup the poller.

func WithPollingContext

func WithPollingContext(ctx context.Context) PollerConfigOption

WithPollingContext sets the context to cancel any retrieval (Next()). It will not change any results for adding data (Set()). Default is context.Background().

func WithPollingInterval

func WithPollingInterval(interval time.Duration) PollerConfigOption

WithPollingInterval sets the interval at which the diode is queried for new data. The default is 10ms.

type Waiter

type Waiter struct {
	Diode
	// contains filtered or unexported fields
}

Waiter will use a channel signal to alert the reader to when data is available.

func NewWaiter

func NewWaiter(d Diode, opts ...WaiterConfigOption) *Waiter

NewWaiter returns a new Waiter that wraps the given diode.

func (*Waiter) Next

func (w *Waiter) Next() GenericDataType

Next returns the next data point on the wrapped diode. If there is no new data, it will wait for Set to be called or the context to be done. If the context is done, then nil will be returned.

func (*Waiter) Set

func (w *Waiter) Set(data GenericDataType)

Set invokes the wrapped diode's Set with the given data and uses broadcast to wake up any readers.

type WaiterConfigOption

type WaiterConfigOption func(*Waiter)

WaiterConfigOption can be used to setup the waiter.

func WithWaiterContext

func WithWaiterContext(ctx context.Context) WaiterConfigOption

WithWaiterContext sets the context to cancel any retrieval (Next()). It will not change any results for adding data (Set()). Default is context.Background().

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL