decision

package
v0.24.1 Latest Latest
Warning

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

 Go to latest Published: Oct 10, 2024 License: Apache-2.0, MIT Imports: 27 Imported by: 0

Documentation

Overview

Package decision implements the decision engine for the bitswap service.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type DefaultPeerLedger 

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

func NewDefaultPeerLedger 

func NewDefaultPeerLedger(maxEntriesPerPeer uint) *DefaultPeerLedger

func (*DefaultPeerLedger) CancelWant 

func (l *DefaultPeerLedger) CancelWant(p peer.ID, k cid.Cid) bool

func (*DefaultPeerLedger) CancelWantWithType 

func (l *DefaultPeerLedger) CancelWantWithType(p peer.ID, k cid.Cid, typ pb.Message_Wantlist_WantType)

func (*DefaultPeerLedger) ClearPeerWantlist 

func (l *DefaultPeerLedger) ClearPeerWantlist(p peer.ID)

ClearPeerWantlist does not take an effort to fully erase it from memory. This is intended when the peer is still connected and the map capacity could be reused. If the memory should be freed use PeerDisconnected instead.

func (*DefaultPeerLedger) CollectPeerIDs 

func (l *DefaultPeerLedger) CollectPeerIDs() []peer.ID

func (*DefaultPeerLedger) PeerDisconnected 

func (l *DefaultPeerLedger) PeerDisconnected(p peer.ID)

func (*DefaultPeerLedger) Peers 

func (l *DefaultPeerLedger) Peers(k cid.Cid) []PeerEntry

func (*DefaultPeerLedger) WantlistForPeer 

func (l *DefaultPeerLedger) WantlistForPeer(p peer.ID) []wl.Entry

func (*DefaultPeerLedger) WantlistSizeForPeer 

func (l *DefaultPeerLedger) WantlistSizeForPeer(p peer.ID) int

func (*DefaultPeerLedger) Wants 

func (l *DefaultPeerLedger) Wants(p peer.ID, e wl.Entry) bool

Wants adds an entry to the peer ledger. If adding the entry would make the peer ledger exceed the maxEntriesPerPeer limit, then the entry is not added and false is returned.

type DefaultScoreLedger 

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

DefaultScoreLedger is used by Engine as the default ScoreLedger.

func NewDefaultScoreLedger 

func NewDefaultScoreLedger() *DefaultScoreLedger

Creates a new instance of the default score ledger.

func NewTestScoreLedger 

func NewTestScoreLedger(peerSampleInterval time.Duration, sampleCh chan struct{}, clock clock.Clock) *DefaultScoreLedger

Creates a new instance of the default score ledger with testing parameters.

func (*DefaultScoreLedger) AddToReceivedBytes 

func (dsl *DefaultScoreLedger) AddToReceivedBytes(p peer.ID, n int)

Increments the received counter for the given peer.

func (*DefaultScoreLedger) AddToSentBytes 

func (dsl *DefaultScoreLedger) AddToSentBytes(p peer.ID, n int)

Increments the sent counter for the given peer.

func (*DefaultScoreLedger) GetReceipt 

func (dsl *DefaultScoreLedger) GetReceipt(p peer.ID) *Receipt

GetReceipt returns aggregated data communication with a given peer.

func (*DefaultScoreLedger) PeerConnected 

func (dsl *DefaultScoreLedger) PeerConnected(p peer.ID)

PeerConnected should be called when a new peer connects, meaning we should open accounting.

func (*DefaultScoreLedger) PeerDisconnected 

func (dsl *DefaultScoreLedger) PeerDisconnected(p peer.ID)

PeerDisconnected should be called when a peer disconnects to clean up the accounting.

func (*DefaultScoreLedger) Start 

func (dsl *DefaultScoreLedger) Start(scorePeer ScorePeerFunc)

Starts the default ledger sampling process.

func (*DefaultScoreLedger) Stop 

func (dsl *DefaultScoreLedger) Stop()

Stops the sampling process.

type Engine 

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

Engine manages sending requested blocks to peers.

func NewEngine 

func NewEngine(
	ctx context.Context,
	bs bstore.Blockstore,
	peerTagger PeerTagger,
	self peer.ID,
	opts ...Option,
) *Engine

NewEngine creates a new block sending engine for the given block store. maxOutstandingBytesPerPeer hints to the peer task queue not to give a peer more tasks if it has some maximum work already outstanding.

func (*Engine) LedgerForPeer 

func (e *Engine) LedgerForPeer(p peer.ID) *Receipt

LedgerForPeer returns aggregated data communication with a given peer.

func (*Engine) MessageReceived 

func (e *Engine) MessageReceived(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage) bool

MessageReceived is called when a message is received from a remote peer. For each item in the wantlist, add a want-have or want-block entry to the request queue (this is later popped off by the workerTasks). Returns true if the connection to the server must be closed.

func (*Engine) MessageSent 

func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage)

MessageSent is called when a message has successfully been sent out, to record changes.

func (*Engine) NotifyNewBlocks 

func (e *Engine) NotifyNewBlocks(blks []blocks.Block)

NotifyNewBlocks is called when new blocks becomes available locally, and in particular when the caller of bitswap decide to store those blocks and make them available on the network.

func (*Engine) Outbox 

func (e *Engine) Outbox() <-chan (<-chan *Envelope)

Outbox returns a channel of one-time use Envelope channels.

func (*Engine) PeerConnected 

func (e *Engine) PeerConnected(p peer.ID)

PeerConnected is called when a new peer connects, meaning we should start sending blocks.

func (*Engine) PeerDisconnected 

func (e *Engine) PeerDisconnected(p peer.ID)

PeerDisconnected is called when a peer disconnects.

func (*Engine) Peers 

func (e *Engine) Peers() []peer.ID

Peers returns a slice of Peers with whom the local node has active sessions.

func (*Engine) ReceivedBlocks 

func (e *Engine) ReceivedBlocks(from peer.ID, blks []blocks.Block)

ReceivedBlocks is called when new blocks are received from the network. This function also updates the receive side of the ledger.

func (*Engine) SetSendDontHaves 

func (e *Engine) SetSendDontHaves(send bool)

SetSendDontHaves indicates what to do when the engine receives a want-block for a block that is not in the blockstore. Either - Send a DONT_HAVE message - Simply don't respond Older versions of Bitswap did not respond, so this allows us to simulate those older versions for testing.

func (*Engine) StartWorkers 

func (e *Engine) StartWorkers(ctx context.Context, px process.Process)

Start up workers to handle requests from other nodes for the data on this node

func (*Engine) WantlistForPeer 

func (e *Engine) WantlistForPeer(p peer.ID) []wl.Entry

WantlistForPeer returns the list of keys that the given peer has asked for

type Envelope 

type Envelope struct {
	// Peer is the intended recipient.
	Peer peer.ID

	// Message is the payload.
	Message bsmsg.BitSwapMessage

	// A callback to notify the decision queue that the task is complete
	Sent func()
}

Envelope contains a message for a Peer.

type Option 

type Option func(*Engine)

func WithBlockstoreWorkerCount 

func WithBlockstoreWorkerCount(count int) Option

WithBlockstoreWorkerCount sets the number of worker threads used for blockstore operations in the decision engine

func WithMaxCidSize 

func WithMaxCidSize(n uint) Option

WithMaxQueuedWantlistEntriesPerPeer limits how much individual entries each peer is allowed to send. If a peer send us more than this we will truncate newest entries.

func WithMaxOutstandingBytesPerPeer 

func WithMaxOutstandingBytesPerPeer(count int) Option

WithMaxOutstandingBytesPerPeer describes approximately how much work we are will to have outstanding to a peer at any given time. Setting it to 0 will disable any limiting.

func WithMaxQueuedWantlistEntriesPerPeer 

func WithMaxQueuedWantlistEntriesPerPeer(count uint) Option

WithMaxQueuedWantlistEntriesPerPeer limits how many individual entries each peer is allowed to send. If a peer sends more than this, then the lowest priority entries are truncated to this limit. If there is insufficient space to enqueue new entries, then older existing wants with no associated blocks, and lower priority wants, are canceled to make room for the new wants.

func WithPeerBlockRequestFilter 

func WithPeerBlockRequestFilter(pbrf PeerBlockRequestFilter) Option

func WithPeerLedger 

func WithPeerLedger(peerLedger PeerLedger) Option

WithPeerLedger sets a custom PeerLedger to be used with this Engine.

func WithScoreLedger 

func WithScoreLedger(scoreledger ScoreLedger) Option

func WithSetSendDontHave 

func WithSetSendDontHave(send bool) Option

func WithTargetMessageSize 

func WithTargetMessageSize(size int) Option

func WithTaskComparator 

func WithTaskComparator(comparator TaskComparator) Option

func WithTaskWorkerCount 

func WithTaskWorkerCount(count int) Option

WithTaskWorkerCount sets the number of worker threads used inside the engine

type PeerBlockRequestFilter 

type PeerBlockRequestFilter func(p peer.ID, c cid.Cid) bool

PeerBlockRequestFilter is used to accept / deny requests for a CID coming from a PeerID It should return true if the request should be fullfilled.

type PeerEntry 

type PeerEntry struct {
	Peer     peer.ID
	Priority int32
	WantType pb.Message_Wantlist_WantType
}

type PeerLedger 

type PeerLedger interface {
	// Wants informs the ledger that [peer.ID] wants [wl.Entry].
	// If peer ledger exceed internal limit, then the entry is not added
	// and false is returned.
	Wants(p peer.ID, e wl.Entry) bool

	// CancelWant returns true if the [cid.Cid] was removed from the wantlist of [peer.ID].
	CancelWant(p peer.ID, k cid.Cid) bool

	// CancelWantWithType will not cancel WantBlock if we sent a HAVE message.
	CancelWantWithType(p peer.ID, k cid.Cid, typ pb.Message_Wantlist_WantType)

	// Peers returns all peers that want [cid.Cid].
	Peers(k cid.Cid) []PeerEntry

	// CollectPeerIDs returns all peers that the ledger has an active session with.
	CollectPeerIDs() []peer.ID

	// WantlistSizeForPeer returns the size of the wantlist for [peer.ID].
	WantlistSizeForPeer(p peer.ID) int

	// WantlistForPeer returns the wantlist for [peer.ID].
	WantlistForPeer(p peer.ID) []wl.Entry

	// ClearPeerWantlist clears the wantlist for [peer.ID].
	ClearPeerWantlist(p peer.ID)

	// PeerDisconnected informs the ledger that [peer.ID] is no longer connected.
	PeerDisconnected(p peer.ID)
}

PeerLedger is an external ledger dealing with peers and their want lists.

type PeerTagger 

type PeerTagger interface {
	TagPeer(peer.ID, string, int)
	UntagPeer(p peer.ID, tag string)
}

PeerTagger covers the methods on the connection manager used by the decision engine to tag peers

type Receipt 

type Receipt struct {
	Peer      string
	Value     float64
	Sent      uint64
	Recv      uint64
	Exchanged uint64
}

Receipt is a summary of the ledger for a given peer collecting various pieces of aggregated data for external reporting purposes.

type ScoreLedger 

type ScoreLedger interface {
	// Returns aggregated data communication with a given peer.
	GetReceipt(p peer.ID) *Receipt
	// Increments the sent counter for the given peer.
	AddToSentBytes(p peer.ID, n int)
	// Increments the received counter for the given peer.
	AddToReceivedBytes(p peer.ID, n int)
	// PeerConnected should be called when a new peer connects,
	// meaning the ledger should open accounting.
	PeerConnected(p peer.ID)
	// PeerDisconnected should be called when a peer disconnects to
	// clean up the accounting.
	PeerDisconnected(p peer.ID)
	// Starts the ledger sampling process.
	Start(scorePeer ScorePeerFunc)
	// Stops the sampling process.
	Stop()
}

ScoreLedger is an external ledger dealing with peer scores.

type ScorePeerFunc 

type ScorePeerFunc func(peer.ID, int)

Assigns a specific score to a peer

type TaskComparator 

type TaskComparator func(ta, tb *TaskInfo) bool

TaskComparator is used for task prioritization. It should return true if task 'ta' has higher priority than task 'tb'

type TaskInfo 

type TaskInfo struct {
	Peer peer.ID
	// The CID of the block
	Cid cid.Cid
	// Tasks can be want-have or want-block
	IsWantBlock bool
	// Whether to immediately send a response if the block is not found
	SendDontHave bool
	// The size of the block corresponding to the task
	BlockSize int
	// Whether the block was found
	HaveBlock bool
}

TaskInfo represents the details of a request from a peer.

Source Files

View all Source files

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.