dilithium

package module
v0.3.6-0...-49592a2 Latest Latest
Warning

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

 Go to latest Published: Feb 19, 2026 License: Apache-2.0 Imports: 17 Imported by: 0

README

The Dilithium Project

The Dilithium Project is a software framework and a set of tooling that supports the development of high-performance streaming protocols over unreliable message-passing systems (UDP datagrams for example).

Initially, we're focusing on the development of westworld, which is a reliable streaming protocol designed to maximize throughput across non-ideal wide-area network links.

Ultimately, Dilithium will become an extensible framework (implemented in golang) for the development of high-performance communication over any arbitrary unreliable message passing system.

Dilithium feeds into the Hanzo ZT project, in support of it's performance-oriented Transwarp data plane implementation. In this role, Dilithium provides both the UDP-based implementation, and also manages the higher-level overlay abstractions implemented as messages on top of existing streaming protocols.

Concepts

For a more detailed tour of Dilithium and the framework components it provides, see the Concepts Guide.

Loss Handling

Documentation

Index

Constants

View Source 
const (
	// 0x0 ... 0x7
	HELLO messageType = iota
	ACK
	DATA
	KEEPALIVE
	CLOSE
)
View Source 
const (
	// 0x8 ... 0x80
	RTT        messageFlag = 0x8
	INLINE_ACK messageFlag = 0x10
)

Variables

This section is empty.

Functions

func EncodeAcks 

func EncodeAcks(acks []Ack, data []byte) (n uint32, err error)

Types

type Ack 

type Ack struct {
	Start int32
	End   int32
}

func DecodeAcks 

func DecodeAcks(data []byte) (acks []Ack, sz uint32, err error)

type Adapter 

type Adapter interface {
	io.Reader
	io.Writer
	io.Closer
}

Adapter abstracts the underlying communication mechanism that dilithium is managing. Implementations are free to manage whatever state is necessary, and need to provide a basic `Read`, `Write`, and `Close` facility.

type Buffer 

type Buffer struct {
	Data []byte
	Size uint32
	Used uint32
	// contains filtered or unexported fields
}

func NewBuffer 

func NewBuffer(pool *Pool) *Buffer

func (*Buffer) Ref 

func (buf *Buffer) Ref()

func (*Buffer) Unref 

func (buf *Buffer) Unref()

type Closer 

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

Closer manages the state machine for the shutdown of a TxPortal and RxPortal pair (one side of a communication).

func NewCloser 

func NewCloser(seq *util.Sequence, closeHook func()) *Closer

func (*Closer) EmergencyStop 

func (c *Closer) EmergencyStop()

type Instrument 

type Instrument interface {
	NewInstance(id string) InstrumentInstance
}

func NewInstrument 

func NewInstrument(name string, config map[string]interface{}) (i Instrument, err error)

func NewMetricsInstrument 

func NewMetricsInstrument(config map[string]interface{}) (Instrument, error)

func NewMetricsInstrumentNoCf 

func NewMetricsInstrumentNoCf(path string, snapshotMs int, enabled bool) (Instrument, error)

func NewNilInstrument 

func NewNilInstrument() Instrument

func NewTraceInstrument 

func NewTraceInstrument(config map[string]interface{}) (Instrument, error)

type InstrumentInstance 

type InstrumentInstance interface {
	// connection
	//Listener(adapter Adapter)
	//Hello(adapter Adapter)
	//Connected(adapter Adapter)
	//ConnectionError(adapter Adapter, err error)
	Closed(adapter Adapter)

	// wire
	WireMessageTx(wm *WireMessage)
	WireMessageRetx(wm *WireMessage)
	WireMessageRx(wm *WireMessage)
	// UnknownPeer(peer *net.UDPAddr)
	ReadError(err error)
	WriteError(err error)
	UnexpectedMessageType(mt messageType)

	// control
	TxAck(wm *WireMessage)
	RxAck(wm *WireMessage)
	TxKeepalive(wm *WireMessage)
	RxKeepalive(wm *WireMessage)

	// txPortal
	TxPortalCapacityChanged(capacity int)
	TxPortalSzChanged(capacity int)
	TxPortalRxSzChanged(sz int)
	NewRetxMs(retxMs int)
	NewRetxScale(retxScale float64)
	DuplicateAck(ack int32)

	// rxPortal
	RxPortalSzChanged(capacity int)
	DuplicateRx(wm *WireMessage)

	// allocation
	Allocate(id string)

	// instrument lifecycle
	Shutdown()
}

type MetricsInstrument 

type MetricsInstrument struct {
	Config *MetricsInstrumentConfig
	// contains filtered or unexported fields
}

func (*MetricsInstrument) NewInstance 

func (self *MetricsInstrument) NewInstance(id string) InstrumentInstance

func (*MetricsInstrument) WriteAllSamples 

func (self *MetricsInstrument) WriteAllSamples() error

type MetricsInstrumentConfig 

type MetricsInstrumentConfig struct {
	Path       string `cf:"path"`
	SnapshotMs int    `cf:"snapshot_ms"`
	Enabled    bool   `cf:"enabled"`
}

type NilInstrumentInstance 

type NilInstrumentInstance struct{}

func (NilInstrumentInstance) Allocate 

func (n NilInstrumentInstance) Allocate(id string)

func (NilInstrumentInstance) Closed 

func (n NilInstrumentInstance) Closed(Adapter)

func (NilInstrumentInstance) DuplicateAck 

func (n NilInstrumentInstance) DuplicateAck(ack int32)

func (NilInstrumentInstance) DuplicateRx 

func (n NilInstrumentInstance) DuplicateRx(wm *WireMessage)

func (NilInstrumentInstance) NewRetxMs 

func (n NilInstrumentInstance) NewRetxMs(retxMs int)

func (NilInstrumentInstance) NewRetxScale 

func (n NilInstrumentInstance) NewRetxScale(retxScale float64)

func (NilInstrumentInstance) ReadError 

func (n NilInstrumentInstance) ReadError(err error)

func (NilInstrumentInstance) RxAck 

func (n NilInstrumentInstance) RxAck(wm *WireMessage)

func (NilInstrumentInstance) RxKeepalive 

func (n NilInstrumentInstance) RxKeepalive(wm *WireMessage)

func (NilInstrumentInstance) RxPortalSzChanged 

func (n NilInstrumentInstance) RxPortalSzChanged(capacity int)

func (NilInstrumentInstance) Shutdown 

func (n NilInstrumentInstance) Shutdown()

func (NilInstrumentInstance) TxAck 

func (n NilInstrumentInstance) TxAck(wm *WireMessage)

func (NilInstrumentInstance) TxKeepalive 

func (n NilInstrumentInstance) TxKeepalive(wm *WireMessage)

func (NilInstrumentInstance) TxPortalCapacityChanged 

func (n NilInstrumentInstance) TxPortalCapacityChanged(capacity int)

func (NilInstrumentInstance) TxPortalRxSzChanged 

func (n NilInstrumentInstance) TxPortalRxSzChanged(sz int)

func (NilInstrumentInstance) TxPortalSzChanged 

func (n NilInstrumentInstance) TxPortalSzChanged(capacity int)

func (NilInstrumentInstance) UnexpectedMessageType 

func (n NilInstrumentInstance) UnexpectedMessageType(mt messageType)

func (NilInstrumentInstance) WireMessageRetx 

func (n NilInstrumentInstance) WireMessageRetx(wm *WireMessage)

func (NilInstrumentInstance) WireMessageRx 

func (n NilInstrumentInstance) WireMessageRx(wm *WireMessage)

func (NilInstrumentInstance) WireMessageTx 

func (n NilInstrumentInstance) WireMessageTx(wm *WireMessage)

func (NilInstrumentInstance) WriteError 

func (n NilInstrumentInstance) WriteError(err error)

type Pool 

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

func NewPool 

func NewPool(id string, bufSize uint32, ii InstrumentInstance) *Pool

func (*Pool) Get 

func (pool *Pool) Get() *Buffer

func (*Pool) Put 

func (pool *Pool) Put(buf *Buffer)

type ReadSinkAdapter 

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

func (*ReadSinkAdapter) Accept 

func (self *ReadSinkAdapter) Accept(data []byte) error

func (*ReadSinkAdapter) Close 

func (self *ReadSinkAdapter) Close()

func (*ReadSinkAdapter) Read 

func (self *ReadSinkAdapter) Read(p []byte) (int, error)

type RxPortal 

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

func NewRxPortal 

func NewRxPortal(adapter Adapter, sink Sink, txp *TxPortal, seq *util.Sequence, closer *Closer, ii InstrumentInstance) *RxPortal

func (*RxPortal) Close 

func (rxp *RxPortal) Close()

func (*RxPortal) Rx 

func (rxp *RxPortal) Rx(wm *WireMessage) (err error)

func (*RxPortal) SetAccepted 

func (rxp *RxPortal) SetAccepted(accepted int32)

type RxRead 

type RxRead struct {
	Buf  []byte
	Size int
	Eof  bool
}

type Sink 

type Sink interface {
	Accept(data []byte) error
	Close()
}

func NewReadSinkAdapter 

func NewReadSinkAdapter(profile WestworldAlgorithm) Sink

type TxAlgorithm 

type TxAlgorithm interface {
	// SetLock will receive the shared lock from TxPortal.
	//
	SetLock(lock *sync.Mutex)

	// Tx will block the caller until there is sufficient space on the wire to enqueue the message.
	//
	Tx(int)

	// Success will unblock Tx callers by freeing up space on the wire from a successfully received message.
	//
	Success(int)

	// DuplicateAck is a signal that a duplicate transmission was received by the receiver.
	//
	DuplicateAck()

	// Retransmission is a signal that an acknowledgement was not received from the receiver within the round-trip time
	// window, and the transmitter sent another copy of the message.
	//
	Retransmission(int)

	// ProbeRTT will return bool when the transmitter is due to probe round trip time. It will also record that true
	// response and will not return true again until the algorithm wants another RTT probe.
	//
	ProbeRTT() bool

	// UpdateRTT pushes a round-trip time probe result onto the algorithm, allowing it to adapt its RetxMs calculation
	// accordingly.
	//
	UpdateRTT(rttMs int)

	// RetxMs returns the current timeout value for retransmission events.
	//
	RetxMs() int

	// RxPortalSize returns the currently observed size of the receiver's buffer.
	//
	RxPortalSize() int

	// UpdateRxPortalSize updates the observed size of the receiver's buffer.
	//
	UpdateRxPortalSize(int)

	// RxPortalPacing determines whether or not the RxPortal should send a keepalive in response to portal size changes.
	//
	RxPortalPacing(oldSize, newSize int) bool

	// Profile returns the requested tunables for this algorithm.
	//
	Profile() *TxProfile
}

TxAlgorithm is an abstraction of an extensible flow-control implementation, which can be plugged into a TxPortal instance.

func NewWestworldAlgorithm 

func NewWestworldAlgorithm(pf *WestworldProfile, ii InstrumentInstance) TxAlgorithm

type TxAlgorithmProfile 

type TxAlgorithmProfile interface {
	Create(ii InstrumentInstance) (TxAlgorithm, error)
}

type TxMonitor 

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

TxMonitor is responsible for managing in-flight payloads, retransmitting payloads when their timeout expires.

type TxPortal 

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

TxPortal manages the outgoing data transmitted by a communication instance. It is one half of a TxPortal->RxPortal communication pair. TxPortal is primarily concerned with optimizing the transmission rate over lossy Adapter implementations, while ensuring reliability.

func NewTxPortal 

func NewTxPortal(adapter Adapter, alg TxAlgorithm, closer *Closer, ii InstrumentInstance) *TxPortal

func (*TxPortal) Start 

func (txp *TxPortal) Start()

func (*TxPortal) Tx 

func (txp *TxPortal) Tx(p []byte, seq *util.Sequence) (n int, err error)

type TxProfile 

type TxProfile struct {
	MaxSegmentSize          int
	RetxBatchMs             int
	SendKeepalive           bool
	ConnectionTimeout       time.Duration
	MaxTreeSize             int
	ReadsQueueSize          int
	PoolBufferSize          int
	RxPortalPacingThreshold float64
	CloseCheckMs            int
}

TxProfile defines all of the configurable values that are requested by a flow control algorithm.

func DefaultTxProfile 

func DefaultTxProfile() *TxProfile

func (*TxProfile) NewPool 

func (txp *TxProfile) NewPool(id string, ii InstrumentInstance) *Pool

type WestworldAlgorithm 

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

WestworldAlgorithm implements the latest iteration of "westworld"-style flow control.

func (*WestworldAlgorithm) DuplicateAck 

func (wa *WestworldAlgorithm) DuplicateAck()

func (*WestworldAlgorithm) ProbeRTT 

func (wa *WestworldAlgorithm) ProbeRTT() bool

func (*WestworldAlgorithm) Profile 

func (wa *WestworldAlgorithm) Profile() *TxProfile

func (*WestworldAlgorithm) Retransmission 

func (wa *WestworldAlgorithm) Retransmission(_ int)

func (*WestworldAlgorithm) RetxMs 

func (wa *WestworldAlgorithm) RetxMs() int

func (*WestworldAlgorithm) RxPortalPacing 

func (wa *WestworldAlgorithm) RxPortalPacing(oldSize, newSize int) bool

func (*WestworldAlgorithm) RxPortalSize 

func (wa *WestworldAlgorithm) RxPortalSize() int

func (*WestworldAlgorithm) SetLock 

func (wa *WestworldAlgorithm) SetLock(lock *sync.Mutex)

func (*WestworldAlgorithm) Success 

func (wa *WestworldAlgorithm) Success(segmentSize int)

func (*WestworldAlgorithm) Tx 

func (wa *WestworldAlgorithm) Tx(segmentSize int)

func (*WestworldAlgorithm) UpdateRTT 

func (wa *WestworldAlgorithm) UpdateRTT(rttMs int)

func (*WestworldAlgorithm) UpdateRxPortalSize 

func (wa *WestworldAlgorithm) UpdateRxPortalSize(rxPortalSize int)

type WestworldProfile 

type WestworldProfile struct {
	StartSize           int
	MinSize             int
	MaxSize             int
	SuccessThresh       int
	SuccessScale        float64
	DupAckThresh        int
	DupAckCapacityScale float64
	DupAckSuccessScale  float64
	RetxStartMs         int
	RetxAddMs           int
	RetxThresh          int
	RetxCapacityScale   float64
	RetxSuccessScale    float64
	RxSizePressureScale float64
	RttProbeMs          int
	RttProbeAvg         int
	Txpf                *TxProfile
}

func NewBaselineWestworldProfile 

func NewBaselineWestworldProfile() *WestworldProfile

func (*WestworldProfile) Create 

func (wp *WestworldProfile) Create(ii InstrumentInstance) (TxAlgorithm, error)

type WireMessage 

type WireMessage struct {
	Seq int32
	Mt  messageType
	// contains filtered or unexported fields
}

Source Files

View all Source files

Directories

Path Synopsis
cmd
dilithium command
dilithium/ctrl
dilithium/dilithium
dilithium/echo
dilithium/influx
dilithium/loop
dilithium/tunnel
ditests command
protocol
loop
westlsworld3
westworld3

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.