query

README

Query

This folder contains multiple Query mechanism implementations.

Implementations

• SimpleQuery is a simple query mechanism.

Documentation

Index

• type ClosestNodesIter
  • func NewClosestNodesIter[K kad.Key[K]](target K) *ClosestNodesIter[K]
  • func (iter *ClosestNodesIter[K]) Add(ni *NodeStatus[K])
  • func (iter *ClosestNodesIter[K]) Each(ctx context.Context, fn func(context.Context, *NodeStatus[K]) bool) bool
  • func (iter *ClosestNodesIter[K]) Find(k K) (*NodeStatus[K], bool)
• type EventPoolAddQuery
• type EventPoolMessageFailure
• type EventPoolMessageResponse
• type EventPoolPoll
• type EventPoolStopQuery
• type EventQueryCancel
• type EventQueryMessageFailure
• type EventQueryMessageResponse
• type NodeIter
• type NodeState
• type NodeStatus
• type Pool
  • func NewPool[K kad.Key[K], A kad.Address[A]](self kad.NodeID[K], cfg *PoolConfig) (*Pool[K, A], error)
  • func (p *Pool[K, A]) Advance(ctx context.Context, ev PoolEvent) PoolState
• type PoolConfig
  • func DefaultPoolConfig() *PoolConfig
  • func (cfg *PoolConfig) Validate() error
• type PoolEvent
• type PoolState
• type Query
  • func NewQuery[K kad.Key[K], A kad.Address[A]](self kad.NodeID[K], id QueryID, protocolID address.ProtocolID, ...) (*Query[K, A], error)
  • func (q *Query[K, A]) Advance(ctx context.Context, ev QueryEvent) QueryState
• type QueryConfig
  • func DefaultQueryConfig[K kad.Key[K]]() *QueryConfig[K]
  • func (cfg *QueryConfig[K]) Validate() error
• type QueryEvent
• type QueryID
• type QueryState
• type QueryStats
• type SequentialIter
  • func NewSequentialIter[K kad.Key[K]]() *SequentialIter[K]
  • func (iter *SequentialIter[K]) Add(ni *NodeStatus[K])
  • func (iter *SequentialIter[K]) Each(ctx context.Context, fn func(context.Context, *NodeStatus[K]) bool) bool
  • func (iter *SequentialIter[K]) Find(k K) (*NodeStatus[K], bool)
• type StateNodeFailed
• type StateNodeNotContacted
• type StateNodeSucceeded
• type StateNodeUnresponsive
• type StateNodeWaiting
• type StatePoolIdle
• type StatePoolQueryFinished
• type StatePoolQueryMessage
• type StatePoolQueryTimeout
• type StatePoolWaitingAtCapacity
• type StatePoolWaitingWithCapacity
• type StateQueryFinished
• type StateQueryWaitingAtCapacity
• type StateQueryWaitingMessage
• type StateQueryWaitingWithCapacity

Types

type ClosestNodesIter

type ClosestNodesIter[K kad.Key[K]] struct {
	// contains filtered or unexported fields
}

A ClosestNodesIter iterates nodes in order of ascending distance from a key.

func NewClosestNodesIter

func NewClosestNodesIter[K kad.Key[K]](target K) *ClosestNodesIter[K]

NewClosestNodesIter creates a new ClosestNodesIter

func (*ClosestNodesIter[K]) Add

func (iter *ClosestNodesIter[K]) Add(ni *NodeStatus[K])

func (*ClosestNodesIter[K]) Each

func (iter *ClosestNodesIter[K]) Each(ctx context.Context, fn func(context.Context, *NodeStatus[K]) bool) bool

func (*ClosestNodesIter[K]) Find

func (iter *ClosestNodesIter[K]) Find(k K) (*NodeStatus[K], bool)

type EventPoolAddQuery

type EventPoolAddQuery[K kad.Key[K], A kad.Address[A]] struct {
	QueryID           QueryID            // the id to use for the new query
	Target            K                  // the target key for the query
	ProtocolID        address.ProtocolID // the protocol that defines how the message should be interpreted
	Message           kad.Request[K, A]  // the message the query should send to each node it traverses
	KnownClosestNodes []kad.NodeID[K]    // an initial set of close nodes the query should use
}

EventPoolAddQuery is an event that attempts to add a new query

type EventPoolMessageFailure

type EventPoolMessageFailure[K kad.Key[K]] struct {
	QueryID QueryID       // the id of the query that sent the message
	NodeID  kad.NodeID[K] // the node the message was sent to
	Error   error         // the error that caused the failure, if any
}

EventPoolMessageFailure notifies a pool that a query that an attempt to send a message has failed.

type EventPoolMessageResponse

type EventPoolMessageResponse[K kad.Key[K], A kad.Address[A]] struct {
	QueryID  QueryID            // the id of the query that sent the message
	NodeID   kad.NodeID[K]      // the node the message was sent to
	Response kad.Response[K, A] // the message response sent by the node
}

EventPoolMessageResponse notifies a pool that a query that a sent message has received a successful response.

type EventPoolPoll

type EventPoolPoll struct{}

EventPoolPoll is an event that signals the pool that it can perform housekeeping work such as time out queries.

type EventPoolStopQuery

type EventPoolStopQuery struct {
	QueryID QueryID // the id of the query that should be stopped
}

EventPoolStopQuery notifies a pool to stop a query.

type EventQueryCancel

type EventQueryCancel struct{}

EventQueryMessageResponse notifies a query to stop all work and enter the finished state.

type EventQueryMessageFailure

type EventQueryMessageFailure[K kad.Key[K]] struct {
	NodeID kad.NodeID[K] // the node the message was sent to
	Error  error         // the error that caused the failure, if any
}

EventQueryMessageFailure notifies a query that an attempt to send a message has failed.

type EventQueryMessageResponse

type EventQueryMessageResponse[K kad.Key[K], A kad.Address[A]] struct {
	NodeID   kad.NodeID[K]      // the node the message was sent to
	Response kad.Response[K, A] // the message response sent by the node
}

EventQueryMessageResponse notifies a query that an attempt to send a message has received a successful response.

type NodeIter

type NodeIter[K kad.Key[K]] interface {
	// Add adds node information to the iterator
	Add(*NodeStatus[K])

	// Find returns the node information corresponding to the given Kademlia key
	Find(K) (*NodeStatus[K], bool)

	// Each applies fn to each entry in the iterator in order. Each stops and returns true if fn returns true.
	// Otherwise Each returns false when there are no further entries.
	Each(ctx context.Context, fn func(context.Context, *NodeStatus[K]) bool) bool
}

A NodeIter iterates nodes according to some strategy.

type NodeState

type NodeState interface {
	// contains filtered or unexported methods
}

type NodeStatus

type NodeStatus[K kad.Key[K]] struct {
	NodeID kad.NodeID[K]
	State  NodeState
}

type Pool

type Pool[K kad.Key[K], A kad.Address[A]] struct {
	// contains filtered or unexported fields
}

func NewPool

func NewPool[K kad.Key[K], A kad.Address[A]](self kad.NodeID[K], cfg *PoolConfig) (*Pool[K, A], error)

func (*Pool[K, A]) Advance

func (p *Pool[K, A]) Advance(ctx context.Context, ev PoolEvent) PoolState

Advance advances the state of the pool by attempting to advance one of its queries

type PoolConfig

type PoolConfig struct {
	Concurrency      int           // the maximum number of queries that may be waiting for message responses at any one time
	Timeout          time.Duration // the time to wait before terminating a query that is not making progress
	Replication      int           // the 'k' parameter defined by Kademlia
	QueryConcurrency int           // the maximum number of concurrent requests that each query may have in flight
	RequestTimeout   time.Duration // the timeout queries should use for contacting a single node
	Clock            clock.Clock   // a clock that may replaced by a mock when testing
}

PoolConfig specifies optional configuration for a Pool

func DefaultPoolConfig

func DefaultPoolConfig() *PoolConfig

DefaultPoolConfig returns the default configuration options for a Pool. Options may be overridden before passing to NewPool

func (*PoolConfig) Validate

func (cfg *PoolConfig) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type PoolEvent

type PoolEvent interface {
	// contains filtered or unexported methods
}

PoolEvent is an event intended to advance the state of a pool.

type PoolState

type PoolState interface {
	// contains filtered or unexported methods
}

type Query

type Query[K kad.Key[K], A kad.Address[A]] struct {
	// contains filtered or unexported fields
}

func NewQuery

func NewQuery[K kad.Key[K], A kad.Address[A]](self kad.NodeID[K], id QueryID, protocolID address.ProtocolID, msg kad.Request[K, A], iter NodeIter[K], knownClosestNodes []kad.NodeID[K], cfg *QueryConfig[K]) (*Query[K, A], error)

func (*Query[K, A]) Advance

func (q *Query[K, A]) Advance(ctx context.Context, ev QueryEvent) QueryState

type QueryConfig

type QueryConfig[K kad.Key[K]] struct {
	Concurrency    int           // the maximum number of concurrent requests that may be in flight
	NumResults     int           // the minimum number of nodes to successfully contact before considering iteration complete
	RequestTimeout time.Duration // the timeout for contacting a single node
	Clock          clock.Clock   // a clock that may replaced by a mock when testing
}

QueryConfig specifies optional configuration for a Query

func DefaultQueryConfig

func DefaultQueryConfig[K kad.Key[K]]() *QueryConfig[K]

DefaultQueryConfig returns the default configuration options for a Query. Options may be overridden before passing to NewQuery

func (*QueryConfig[K]) Validate

func (cfg *QueryConfig[K]) Validate() error

Validate checks the configuration options and returns an error if any have invalid values.

type QueryEvent

type QueryEvent interface {
	// contains filtered or unexported methods
}

type QueryID

type QueryID string

const InvalidQueryID QueryID = ""

type QueryState

type QueryState interface {
	// contains filtered or unexported methods
}

type QueryStats

type QueryStats struct {
	Start    time.Time
	End      time.Time
	Requests int
	Success  int
	Failure  int
}

type SequentialIter

type SequentialIter[K kad.Key[K]] struct {
	// contains filtered or unexported fields
}

A SequentialIter iterates nodes in the order they were added to the iterator.

func NewSequentialIter

func NewSequentialIter[K kad.Key[K]]() *SequentialIter[K]

NewSequentialIter creates a new SequentialIter

func (*SequentialIter[K]) Add

func (iter *SequentialIter[K]) Add(ni *NodeStatus[K])

func (*SequentialIter[K]) Each

func (iter *SequentialIter[K]) Each(ctx context.Context, fn func(context.Context, *NodeStatus[K]) bool) bool

func (*SequentialIter[K]) Find

func (iter *SequentialIter[K]) Find(k K) (*NodeStatus[K], bool)

Find returns the node information corresponding to the given Kademlia key. It uses a linear search which makes it unsuitable for large numbers of entries.

type StateNodeFailed

type StateNodeFailed struct{}

StateNodeFailed indicates that the attempt to contact the node failed.

type StateNodeNotContacted

type StateNodeNotContacted struct{}

StateNodeNotContacted indicates that the node has not been contacted yet.

type StateNodeSucceeded

type StateNodeSucceeded struct{}

StateNodeSucceeded indicates that the attempt to contact the node succeeded.

type StateNodeUnresponsive

type StateNodeUnresponsive struct{}

StateNodeUnresponsive indicates that the node did not respond within the configured timeout.

type StateNodeWaiting

type StateNodeWaiting struct {
	Deadline time.Time
}

StateNodeWaiting indicates that a query is waiting for a response from the node.

type StatePoolIdle

type StatePoolIdle struct{}

StatePoolIdle indicates that the pool is idle, i.e. there are no queries to process.

type StatePoolQueryFinished

type StatePoolQueryFinished struct {
	QueryID QueryID
	Stats   QueryStats
}

StatePoolQueryFinished indicates that a query has finished.

type StatePoolQueryMessage

type StatePoolQueryMessage[K kad.Key[K], A kad.Address[A]] struct {
	QueryID    QueryID
	NodeID     kad.NodeID[K]
	ProtocolID address.ProtocolID
	Message    kad.Request[K, A]
	Stats      QueryStats
}

StatePoolQueryMessage indicates that a pool query is waiting to message a node.

type StatePoolQueryTimeout

type StatePoolQueryTimeout struct {
	QueryID QueryID
	Stats   QueryStats
}

StatePoolQueryTimeout indicates that a query has timed out.

type StatePoolWaitingAtCapacity

type StatePoolWaitingAtCapacity struct{}

StatePoolWaitingAtCapacity indicates that at least one query is waiting for results and the pool has reached its maximum number of concurrent queries.

type StatePoolWaitingWithCapacity

type StatePoolWaitingWithCapacity struct{}

StatePoolWaitingWithCapacity indicates that at least one query is waiting for results but capacity to start more is available.

type StateQueryFinished

type StateQueryFinished struct {
	QueryID QueryID
	Stats   QueryStats
}

StateQueryFinished indicates that the Query has finished.

type StateQueryWaitingAtCapacity

type StateQueryWaitingAtCapacity struct {
	QueryID QueryID
	Stats   QueryStats
}

StateQueryWaitingAtCapacity indicates that the Query is waiting for results and is at capacity.

type StateQueryWaitingMessage

type StateQueryWaitingMessage[K kad.Key[K], A kad.Address[A]] struct {
	QueryID    QueryID
	Stats      QueryStats
	NodeID     kad.NodeID[K]
	ProtocolID address.ProtocolID
	Message    kad.Request[K, A]
}

StateQueryWaitingMessage indicates that the Query is waiting to send a message to a node.

type StateQueryWaitingWithCapacity

type StateQueryWaitingWithCapacity struct {
	QueryID QueryID
	Stats   QueryStats
}

StateQueryWaitingWithCapacity indicates that the Query is waiting for results but has no further nodes to contact.