decision

Documentation

Overview

Package decision implements the decision engine for the bitswap service.

Index

• type DefaultScoreLedger
  • func NewDefaultScoreLedger() *DefaultScoreLedger
  • func NewTestScoreLedger(peerSampleInterval time.Duration, sampleCh chan struct{}, clock clock.Clock) *DefaultScoreLedger
  • func (dsl *DefaultScoreLedger) AddToReceivedBytes(p peer.ID, n int)
  • func (dsl *DefaultScoreLedger) AddToSentBytes(p peer.ID, n int)
  • func (dsl *DefaultScoreLedger) GetReceipt(p peer.ID) *Receipt
  • func (dsl *DefaultScoreLedger) PeerConnected(p peer.ID)
  • func (dsl *DefaultScoreLedger) PeerDisconnected(p peer.ID)
  • func (dsl *DefaultScoreLedger) Start(scorePeer ScorePeerFunc)
  • func (dsl *DefaultScoreLedger) Stop()
• type Engine
  • func NewEngine(ctx context.Context, bs bstore.Blockstore, ...) *Engine
  • func (e *Engine) LedgerForPeer(p peer.ID) *Receipt
  • func (e *Engine) MessageReceived(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage)
  • func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage)
  • func (e *Engine) Outbox() <-chan (<-chan *Envelope)
  • func (e *Engine) PeerConnected(p peer.ID)
  • func (e *Engine) PeerDisconnected(p peer.ID)
  • func (e *Engine) Peers() []peer.ID
  • func (e *Engine) ReceiveFrom(from peer.ID, blks []blocks.Block)
  • func (e *Engine) SetSendDontHaves(send bool)
  • func (e *Engine) StartWorkers(ctx context.Context, px process.Process)
  • func (e *Engine) WantlistForPeer(p peer.ID) []wl.Entry
• type Envelope
• type Option
  • func WithTaskComparator(comparator TaskComparator) Option
• type PeerTagger
• type Receipt
• type ScoreLedger
• type ScorePeerFunc
• type TaskComparator
• type TaskInfo

Types

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,
	bstoreWorkerCount,
	engineTaskWorkerCount, maxOutstandingBytesPerPeer int,
	peerTagger PeerTagger,
	self peer.ID,
	scoreLedger ScoreLedger,
	pendingEngineGauge metrics.Gauge,
	activeEngineGauge metrics.Gauge,
	pendingBlocksGauge metrics.Gauge,
	activeBlocksGauge metrics.Gauge,
	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. 向特定块存储（peer）创建一个新的块引擎

func (*Engine) LedgerForPeer

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

LedgerForPeer returns aggregated data communication with a given peer. 获得某peer的（账本）收据信息

func (*Engine) MessageReceived

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

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) 接收到消息，将wantlist放入请求队列 在此进行权限管理

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) 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) ReceiveFrom

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

ReceiveFrom is called when new blocks are received and added to the block store, meaning there may be peers who want those blocks, so we should send the blocks to them.

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 获得某peer的wantlist

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 WithTaskComparator

func WithTaskComparator(comparator TaskComparator) Option

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.
	// 当新的peer连接上时，会调用，也就是会开始统计
	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. 用于处理score的外部账本

type ScorePeerFunc

type ScorePeerFunc func(peer.ID, int)

score: 对某个peer的信任情况 Assigns a specific score to a peer 对peer的score进行处理

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. 任务，即某peer的请求