func (e *Engine) PeerConnected(p peer.ID) {
	e.lock.Lock()
	defer e.lock.Unlock()
	_, ok := e.ledgerMap[p]
	if !ok {
		e.ledgerMap[p] = newLedger(p)
	}
	e.scoreLedger.PeerConnected(p)
}
PeerConnected is called when a new peer connects, meaning we should start
sending blocks.
package decision
import (
	"context"
	"fmt"
	"sync"
	"time"
	"github.com/google/uuid"
	bsmsg "github.com/ipfs/go-bitswap/message"
	pb "github.com/ipfs/go-bitswap/message/pb"
	wl "github.com/ipfs/go-bitswap/wantlist"
	blocks "github.com/ipfs/go-block-format"
	"github.com/ipfs/go-cid"
	bstore "github.com/ipfs/go-ipfs-blockstore"
	logging "github.com/ipfs/go-log"
	"github.com/ipfs/go-metrics-interface"
	"github.com/ipfs/go-peertaskqueue"
	"github.com/ipfs/go-peertaskqueue/peertask"
	process "github.com/jbenet/goprocess"
	"github.com/libp2p/go-libp2p-core/peer"
)
var log = logging.Logger("engine")
const (
	outboxChanBuffer = 0
	targetMessageSize = 16 * 1024
	tagFormat = "bs-engine-%s-%s"
	queuedTagWeight = 10
	maxBlockSizeReplaceHasWithBlock = 1024
)
type Envelope struct {
	Peer peer.ID
	Message bsmsg.BitSwapMessage
	Sent func()
}
type PeerTagger interface {
	TagPeer(peer.ID, string, int)
	UntagPeer(p peer.ID, tag string)
}
type ScorePeerFunc func(peer.ID, int)
type ScoreLedger interface {
	GetReceipt(p peer.ID) *Receipt
	AddToSentBytes(p peer.ID, n int)
	AddToReceivedBytes(p peer.ID, n int)
	PeerConnected(p peer.ID)
	PeerDisconnected(p peer.ID)
	Start(scorePeer ScorePeerFunc)
	Stop()
}
type Engine struct {
	peerRequestQueue *peertaskqueue.PeerTaskQueue
	workSignal chan struct{}
	outbox chan (<-chan *Envelope)
	bsm *blockstoreManager
	peerTagger PeerTagger
	tagQueued, tagUseful string
	lock sync.RWMutex  
	ledgerMap map[peer.ID]*ledger
	peerLedger *peerLedger
	scoreLedger ScoreLedger
	ticker *time.Ticker
	taskWorkerLock  sync.Mutex
	taskWorkerCount int
	maxBlockSizeReplaceHasWithBlock int
	sendDontHaves bool
	self peer.ID
	pendingGauge metrics.Gauge
	activeGauge metrics.Gauge
	metricsLock         sync.Mutex
	metricUpdateCounter int
}
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,
) *Engine {
	return newEngine(
		ctx,
		bs,
		bstoreWorkerCount,
		engineTaskWorkerCount,
		maxOutstandingBytesPerPeer,
		peerTagger,
		self,
		maxBlockSizeReplaceHasWithBlock,
		scoreLedger,
		pendingEngineGauge,
		activeEngineGauge,
		pendingBlocksGauge,
		activeBlocksGauge,
	)
}
func newEngine(
	ctx context.Context,
	bs bstore.Blockstore,
	bstoreWorkerCount,
	engineTaskWorkerCount, maxOutstandingBytesPerPeer int,
	peerTagger PeerTagger,
	self peer.ID,
	maxReplaceSize int,
	scoreLedger ScoreLedger,
	pendingEngineGauge metrics.Gauge,
	activeEngineGauge metrics.Gauge,
	pendingBlocksGauge metrics.Gauge,
	activeBlocksGauge metrics.Gauge,
) *Engine {
	if scoreLedger == nil {
		scoreLedger = NewDefaultScoreLedger()
	}
	e := &Engine{
		ledgerMap:                       make(map[peer.ID]*ledger),
		scoreLedger:                     scoreLedger,
		bsm:                             newBlockstoreManager(ctx, bs, bstoreWorkerCount, pendingBlocksGauge, activeBlocksGauge),
		peerTagger:                      peerTagger,
		outbox:                          make(chan (<-chan *Envelope), outboxChanBuffer),
		workSignal:                      make(chan struct{}, 1),
		ticker:                          time.NewTicker(time.Millisecond * 100),
		maxBlockSizeReplaceHasWithBlock: maxReplaceSize,
		taskWorkerCount:                 engineTaskWorkerCount,
		sendDontHaves:                   true,
		self:                            self,
		peerLedger:                      newPeerLedger(),
		pendingGauge:                    pendingEngineGauge,
		activeGauge:                     activeEngineGauge,
	}
	e.tagQueued = fmt.Sprintf(tagFormat, "queued", uuid.New().String())
	e.tagUseful = fmt.Sprintf(tagFormat, "useful", uuid.New().String())
	e.peerRequestQueue = peertaskqueue.New(
		peertaskqueue.OnPeerAddedHook(e.onPeerAdded),
		peertaskqueue.OnPeerRemovedHook(e.onPeerRemoved),
		peertaskqueue.TaskMerger(newTaskMerger()),
		peertaskqueue.IgnoreFreezing(true),
		peertaskqueue.MaxOutstandingWorkPerPeer(maxOutstandingBytesPerPeer))
	return e
}
func (e *Engine) updateMetrics() {
	e.metricsLock.Lock()
	c := e.metricUpdateCounter
	e.metricUpdateCounter++
	e.metricsLock.Unlock()
	if c%100 == 0 {
		stats := e.peerRequestQueue.Stats()
		e.activeGauge.Set(float64(stats.NumActive))
		e.pendingGauge.Set(float64(stats.NumPending))
	}
}
func (e *Engine) SetSendDontHaves(send bool) {
	e.sendDontHaves = send
}
func (e *Engine) startScoreLedger(px process.Process) {
	e.scoreLedger.Start(func(p peer.ID, score int) {
		if score == 0 {
			e.peerTagger.UntagPeer(p, e.tagUseful)
		} else {
			e.peerTagger.TagPeer(p, e.tagUseful, score)
		}
	})
	px.Go(func(ppx process.Process) {
		<-ppx.Closing()
		e.scoreLedger.Stop()
	})
}
func (e *Engine) StartWorkers(ctx context.Context, px process.Process) {
	e.bsm.start(px)
	e.startScoreLedger(px)
	e.taskWorkerLock.Lock()
	defer e.taskWorkerLock.Unlock()
	for i := 0; i < e.taskWorkerCount; i++ {
		px.Go(func(px process.Process) {
			e.taskWorker(ctx)
		})
	}
}
func (e *Engine) onPeerAdded(p peer.ID) {
	e.peerTagger.TagPeer(p, e.tagQueued, queuedTagWeight)
}
func (e *Engine) onPeerRemoved(p peer.ID) {
	e.peerTagger.UntagPeer(p, e.tagQueued)
}
func (e *Engine) WantlistForPeer(p peer.ID) []wl.Entry {
	partner := e.findOrCreate(p)
	partner.lk.Lock()
	entries := partner.wantList.Entries()
	partner.lk.Unlock()
	return entries
}
func (e *Engine) LedgerForPeer(p peer.ID) *Receipt {
	return e.scoreLedger.GetReceipt(p)
}
func (e *Engine) taskWorker(ctx context.Context) {
	defer e.taskWorkerExit()
	for {
		oneTimeUse := make(chan *Envelope, 1)  
		select {
		case <-ctx.Done():
			return
		case e.outbox <- oneTimeUse:
		}
		envelope, err := e.nextEnvelope(ctx)
		if err != nil {
			close(oneTimeUse)
			return  
		}
		oneTimeUse <- envelope  
		close(oneTimeUse)
	}
}
func (e *Engine) taskWorkerExit() {
	e.taskWorkerLock.Lock()
	defer e.taskWorkerLock.Unlock()
	e.taskWorkerCount--
	if e.taskWorkerCount == 0 {
		close(e.outbox)
	}
}
func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {
	for {
		p, nextTasks, pendingBytes := e.peerRequestQueue.PopTasks(targetMessageSize)
		e.updateMetrics()
		for len(nextTasks) == 0 {
			select {
			case <-ctx.Done():
				return nil, ctx.Err()
			case <-e.workSignal:
				p, nextTasks, pendingBytes = e.peerRequestQueue.PopTasks(targetMessageSize)
				e.updateMetrics()
			case <-e.ticker.C:
				e.peerRequestQueue.ThawRound()
				p, nextTasks, pendingBytes = e.peerRequestQueue.PopTasks(targetMessageSize)
				e.updateMetrics()
			}
		}
		msg := bsmsg.New(false)
		log.Debugw("Bitswap process tasks", "local", e.self, "taskCount", len(nextTasks))
		msg.SetPendingBytes(int32(pendingBytes))
		blockCids := make([]cid.Cid, 0, len(nextTasks))
		blockTasks := make(map[cid.Cid]*taskData, len(nextTasks))
		for _, t := range nextTasks {
			c := t.Topic.(cid.Cid)
			td := t.Data.(*taskData)
			if td.HaveBlock {
				if td.IsWantBlock {
					blockCids = append(blockCids, c)
					blockTasks[c] = td
				} else {
					msg.AddHave(c)
				}
			} else {
				msg.AddDontHave(c)
			}
		}
		blks, err := e.bsm.getBlocks(ctx, blockCids)
		if err != nil {
			return nil, err
		}
		for c, t := range blockTasks {
			blk := blks[c]
			if blk == nil {
				if t.SendDontHave {
					msg.AddDontHave(c)
				}
			} else {
				msg.AddBlock(blk)
			}
		}
		if msg.Empty() {
			e.peerRequestQueue.TasksDone(p, nextTasks...)
			continue
		}
		log.Debugw("Bitswap engine -> msg", "local", e.self, "to", p, "blockCount", len(msg.Blocks()), "presenceCount", len(msg.BlockPresences()), "size", msg.Size())
		return &Envelope{
			Peer:    p,
			Message: msg,
			Sent: func() {
				e.peerRequestQueue.TasksDone(p, nextTasks...)
				e.signalNewWork()
			},
		}, nil
	}
}
func (e *Engine) Outbox() <-chan (<-chan *Envelope) {
	return e.outbox
}
func (e *Engine) Peers() []peer.ID {
	e.lock.RLock()
	defer e.lock.RUnlock()
	response := make([]peer.ID, 0, len(e.ledgerMap))
	for _, ledger := range e.ledgerMap {
		response = append(response, ledger.Partner)
	}
	return response
}
func (e *Engine) MessageReceived(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage) {
	entries := m.Wantlist()
	if len(entries) > 0 {
		log.Debugw("Bitswap engine <- msg", "local", e.self, "from", p, "entryCount", len(entries))
		for _, et := range entries {
			if !et.Cancel {
				if et.WantType == pb.Message_Wantlist_Have {
					log.Debugw("Bitswap engine <- want-have", "local", e.self, "from", p, "cid", et.Cid)
				} else {
					log.Debugw("Bitswap engine <- want-block", "local", e.self, "from", p, "cid", et.Cid)
				}
			}
		}
	}
	if m.Empty() {
		log.Infof("received empty message from %s", p)
	}
	newWorkExists := false
	defer func() {
		if newWorkExists {
			e.signalNewWork()
		}
	}()
	wants, cancels := e.splitWantsCancels(entries)
	wantKs := cid.NewSet()
	for _, entry := range wants {
		wantKs.Add(entry.Cid)
	}
	blockSizes, err := e.bsm.getBlockSizes(ctx, wantKs.Keys())
	if err != nil {
		log.Info("aborting message processing", err)
		return
	}
	e.lock.Lock()
	for _, entry := range wants {
		e.peerLedger.Wants(p, entry.Cid)
	}
	for _, entry := range cancels {
		e.peerLedger.CancelWant(p, entry.Cid)
	}
	e.lock.Unlock()
	l := e.findOrCreate(p)
	l.lk.Lock()
	defer l.lk.Unlock()
	if m.Full() {
		l.wantList = wl.New()
	}
	var activeEntries []peertask.Task
	for _, entry := range cancels {
		log.Debugw("Bitswap engine <- cancel", "local", e.self, "from", p, "cid", entry.Cid)
		if l.CancelWant(entry.Cid) {
			e.peerRequestQueue.Remove(entry.Cid, p)
		}
	}
	for _, entry := range wants {
		c := entry.Cid
		blockSize, found := blockSizes[entry.Cid]
		l.Wants(c, entry.Priority, entry.WantType)
		if !found {
			log.Debugw("Bitswap engine: block not found", "local", e.self, "from", p, "cid", entry.Cid, "sendDontHave", entry.SendDontHave)
			if e.sendDontHaves && entry.SendDontHave {
				newWorkExists = true
				isWantBlock := false
				if entry.WantType == pb.Message_Wantlist_Block {
					isWantBlock = true
				}
				activeEntries = append(activeEntries, peertask.Task{
					Topic:    c,
					Priority: int(entry.Priority),
					Work:     bsmsg.BlockPresenceSize(c),
					Data: &taskData{
						BlockSize:    0,
						HaveBlock:    false,
						IsWantBlock:  isWantBlock,
						SendDontHave: entry.SendDontHave,
					},
				})
			}
		} else {
			newWorkExists = true
			isWantBlock := e.sendAsBlock(entry.WantType, blockSize)
			log.Debugw("Bitswap engine: block found", "local", e.self, "from", p, "cid", entry.Cid, "isWantBlock", isWantBlock)
			entrySize := blockSize
			if !isWantBlock {
				entrySize = bsmsg.BlockPresenceSize(c)
			}
			activeEntries = append(activeEntries, peertask.Task{
				Topic:    c,
				Priority: int(entry.Priority),
				Work:     entrySize,
				Data: &taskData{
					BlockSize:    blockSize,
					HaveBlock:    true,
					IsWantBlock:  isWantBlock,
					SendDontHave: entry.SendDontHave,
				},
			})
		}
	}
	if len(activeEntries) > 0 {
		e.peerRequestQueue.PushTasks(p, activeEntries...)
		e.updateMetrics()
	}
}
func (e *Engine) splitWantsCancels(es []bsmsg.Entry) ([]bsmsg.Entry, []bsmsg.Entry) {
	wants := make([]bsmsg.Entry, 0, len(es))
	cancels := make([]bsmsg.Entry, 0, len(es))
	for _, et := range es {
		if et.Cancel {
			cancels = append(cancels, et)
		} else {
			wants = append(wants, et)
		}
	}
	return wants, cancels
}
func (e *Engine) ReceiveFrom(from peer.ID, blks []blocks.Block) {
	if len(blks) == 0 {
		return
	}
	if from != "" {
		l := e.findOrCreate(from)
		l.lk.Lock()
		for _, blk := range blks {
			log.Debugw("Bitswap engine <- block", "local", e.self, "from", from, "cid", blk.Cid(), "size", len(blk.RawData()))
			e.scoreLedger.AddToReceivedBytes(l.Partner, len(blk.RawData()))
		}
		l.lk.Unlock()
	}
	blockSizes := make(map[cid.Cid]int, len(blks))
	for _, blk := range blks {
		blockSizes[blk.Cid()] = len(blk.RawData())
	}
	var work bool
	missingWants := make(map[peer.ID][]cid.Cid)
	for _, b := range blks {
		k := b.Cid()
		e.lock.RLock()
		peers := e.peerLedger.Peers(k)
		e.lock.RUnlock()
		for _, p := range peers {
			e.lock.RLock()
			ledger, ok := e.ledgerMap[p]
			e.lock.RUnlock()
			if !ok {
				log.Debugw("failed to find peer in ledger", "peer", p)
				missingWants[p] = append(missingWants[p], k)
				continue
			}
			ledger.lk.RLock()
			entry, ok := ledger.WantListContains(k)
			ledger.lk.RUnlock()
			if !ok {
				log.Debugw("wantlist index doesn't match peer's wantlist", "peer", p)
				missingWants[p] = append(missingWants[p], k)
				continue
			}
			work = true
			blockSize := blockSizes[k]
			isWantBlock := e.sendAsBlock(entry.WantType, blockSize)
			entrySize := blockSize
			if !isWantBlock {
				entrySize = bsmsg.BlockPresenceSize(k)
			}
			e.peerRequestQueue.PushTasks(p, peertask.Task{
				Topic:    entry.Cid,
				Priority: int(entry.Priority),
				Work:     entrySize,
				Data: &taskData{
					BlockSize:    blockSize,
					HaveBlock:    true,
					IsWantBlock:  isWantBlock,
					SendDontHave: false,
				},
			})
			e.updateMetrics()
		}
	}
	if len(missingWants) > 0 {
		e.lock.Lock()
		for p, wl := range missingWants {
			if ledger, ok := e.ledgerMap[p]; ok {
				ledger.lk.RLock()
				for _, k := range wl {
					if _, has := ledger.WantListContains(k); has {
						continue
					}
					e.peerLedger.CancelWant(p, k)
				}
				ledger.lk.RUnlock()
			} else {
				for _, k := range wl {
					e.peerLedger.CancelWant(p, k)
				}
			}
		}
		e.lock.Unlock()
	}
	if work {
		e.signalNewWork()
	}
}
func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage) {
	l := e.findOrCreate(p)
	l.lk.Lock()
	defer l.lk.Unlock()
	for _, block := range m.Blocks() {
		e.scoreLedger.AddToSentBytes(l.Partner, len(block.RawData()))
		l.wantList.RemoveType(block.Cid(), pb.Message_Wantlist_Block)
	}
	for _, bp := range m.BlockPresences() {
		if bp.Type == pb.Message_Have {
			l.wantList.RemoveType(bp.Cid, pb.Message_Wantlist_Have)
		}
	}
}