func (s *Session) handlePeriodicSearch(ctx context.Context) {
	randomWant := s.sw.RandomLiveWant()
	if !randomWant.Defined() {
		return
	}
	s.findMorePeers(ctx, randomWant)
	s.broadcastWantHaves(ctx, []cid.Cid{randomWant})
	s.periodicSearchTimer.Reset(s.periodicSearchDelay.NextWaitTime())
}
handlePeriodicSearch is called periodically to search for providers of a
randomly chosen CID in the sesssion.
package session
import (
	"context"
	"time"
	bsbpm "github.com/ipfs/go-bitswap/internal/blockpresencemanager"
	bsgetter "github.com/ipfs/go-bitswap/internal/getter"
	notifications "github.com/ipfs/go-bitswap/internal/notifications"
	bspm "github.com/ipfs/go-bitswap/internal/peermanager"
	bssim "github.com/ipfs/go-bitswap/internal/sessioninterestmanager"
	blocks "github.com/ipfs/go-block-format"
	cid "github.com/ipfs/go-cid"
	delay "github.com/ipfs/go-ipfs-delay"
	logging "github.com/ipfs/go-log"
	peer "github.com/libp2p/go-libp2p-core/peer"
	loggables "github.com/libp2p/go-libp2p-loggables"
	"go.uber.org/zap"
)
var log = logging.Logger("bs:sess")
var sflog = log.Desugar()
const (
	broadcastLiveWantsLimit = 64
)
type PeerManager interface {
	RegisterSession(peer.ID, bspm.Session)
	UnregisterSession(uint64)
	SendWants(ctx context.Context, peerId peer.ID, wantBlocks []cid.Cid, wantHaves []cid.Cid)
	BroadcastWantHaves(context.Context, []cid.Cid)
	SendCancels(context.Context, []cid.Cid)
}
type SessionManager interface {
	RemoveSession(sesid uint64)
	CancelSessionWants(sid uint64, wants []cid.Cid)
}
type SessionPeerManager interface {
	PeersDiscovered() bool
	Shutdown()
	AddPeer(peer.ID) bool
	RemovePeer(peer.ID) bool
	Peers() []peer.ID
	HasPeers() bool
	ProtectConnection(peer.ID)
}
type ProviderFinder interface {
	FindProvidersAsync(ctx context.Context, k cid.Cid) <-chan peer.ID
}
type opType int
const (
	opReceive opType = iota
	opWant
	opCancel
	opBroadcast
	opWantsSent
)
type op struct {
	op   opType
	keys []cid.Cid
}
type Session struct {
	ctx            context.Context
	shutdown       func()
	sm             SessionManager
	pm             PeerManager
	sprm           SessionPeerManager
	providerFinder ProviderFinder
	sim            *bssim.SessionInterestManager
	sw  sessionWants
	sws sessionWantSender
	latencyTrkr latencyTracker
	incoming      chan op
	tickDelayReqs chan time.Duration
	idleTick            *time.Timer
	periodicSearchTimer *time.Timer
	baseTickDelay       time.Duration
	consecutiveTicks    int
	initialSearchDelay  time.Duration
	periodicSearchDelay delay.D
	notif notifications.PubSub
	uuid  logging.Loggable
	id    uint64
	self peer.ID
}
func New(
	ctx context.Context,
	sm SessionManager,
	id uint64,
	sprm SessionPeerManager,
	providerFinder ProviderFinder,
	sim *bssim.SessionInterestManager,
	pm PeerManager,
	bpm *bsbpm.BlockPresenceManager,
	notif notifications.PubSub,
	initialSearchDelay time.Duration,
	periodicSearchDelay delay.D,
	self peer.ID) *Session {
	ctx, cancel := context.WithCancel(ctx)
	s := &Session{
		sw:                  newSessionWants(broadcastLiveWantsLimit),
		tickDelayReqs:       make(chan time.Duration),
		ctx:                 ctx,
		shutdown:            cancel,
		sm:                  sm,
		pm:                  pm,
		sprm:                sprm,
		providerFinder:      providerFinder,
		sim:                 sim,
		incoming:            make(chan op, 128),
		latencyTrkr:         latencyTracker{},
		notif:               notif,
		uuid:                loggables.Uuid("GetBlockRequest"),
		baseTickDelay:       time.Millisecond * 500,
		id:                  id,
		initialSearchDelay:  initialSearchDelay,
		periodicSearchDelay: periodicSearchDelay,
		self:                self,
	}
	s.sws = newSessionWantSender(id, pm, sprm, sm, bpm, s.onWantsSent, s.onPeersExhausted)
	go s.run(ctx)
	return s
}
func (s *Session) ID() uint64 {
	return s.id
}
func (s *Session) Shutdown() {
	s.shutdown()
}
func (s *Session) ReceiveFrom(from peer.ID, ks []cid.Cid, haves []cid.Cid, dontHaves []cid.Cid) {
	interestedRes := s.sim.FilterSessionInterested(s.id, ks, haves, dontHaves)
	ks = interestedRes[0]
	haves = interestedRes[1]
	dontHaves = interestedRes[2]
	s.logReceiveFrom(from, ks, haves, dontHaves)
	s.sws.Update(from, ks, haves, dontHaves)
	if len(ks) == 0 {
		return
	}
	select {
	case s.incoming <- op{op: opReceive, keys: ks}:
	case <-s.ctx.Done():
	}
}
func (s *Session) logReceiveFrom(from peer.ID, interestedKs []cid.Cid, haves []cid.Cid, dontHaves []cid.Cid) {
	if ce := sflog.Check(zap.DebugLevel, "Bitswap <- rcv message"); ce == nil {
		return
	}
	for _, c := range interestedKs {
		log.Debugw("Bitswap <- block", "local", s.self, "from", from, "cid", c, "session", s.id)
	}
	for _, c := range haves {
		log.Debugw("Bitswap <- HAVE", "local", s.self, "from", from, "cid", c, "session", s.id)
	}
	for _, c := range dontHaves {
		log.Debugw("Bitswap <- DONT_HAVE", "local", s.self, "from", from, "cid", c, "session", s.id)
	}
}
func (s *Session) GetBlock(parent context.Context, k cid.Cid) (blocks.Block, error) {
	return bsgetter.SyncGetBlock(parent, k, s.GetBlocks)
}
func (s *Session) GetBlocks(ctx context.Context, keys []cid.Cid) (<-chan blocks.Block, error) {
	ctx = logging.ContextWithLoggable(ctx, s.uuid)
	return bsgetter.AsyncGetBlocks(ctx, s.ctx, keys, s.notif,
		func(ctx context.Context, keys []cid.Cid) {
			select {
			case s.incoming <- op{op: opWant, keys: keys}:
			case <-ctx.Done():
			case <-s.ctx.Done():
			}
		},
		func(keys []cid.Cid) {
			select {
			case s.incoming <- op{op: opCancel, keys: keys}:
			case <-s.ctx.Done():
			}
		},
	)
}
func (s *Session) SetBaseTickDelay(baseTickDelay time.Duration) {
	select {
	case s.tickDelayReqs <- baseTickDelay:
	case <-s.ctx.Done():
	}
}
func (s *Session) onWantsSent(p peer.ID, wantBlocks []cid.Cid, wantHaves []cid.Cid) {
	allBlks := append(wantBlocks[:len(wantBlocks):len(wantBlocks)], wantHaves...)
	s.nonBlockingEnqueue(op{op: opWantsSent, keys: allBlks})
}
func (s *Session) onPeersExhausted(ks []cid.Cid) {
	s.nonBlockingEnqueue(op{op: opBroadcast, keys: ks})
}
func (s *Session) nonBlockingEnqueue(o op) {
	select {
	case s.incoming <- o:
	default:
		go func() {
			select {
			case s.incoming <- o:
			case <-s.ctx.Done():
			}
		}()
	}
}
func (s *Session) run(ctx context.Context) {
	go s.sws.Run()
	s.idleTick = time.NewTimer(s.initialSearchDelay)
	s.periodicSearchTimer = time.NewTimer(s.periodicSearchDelay.NextWaitTime())
	for {
		select {
		case oper := <-s.incoming:
			switch oper.op {
			case opReceive:
				s.handleReceive(oper.keys)
			case opWant:
				s.wantBlocks(ctx, oper.keys)
			case opCancel:
				s.sw.CancelPending(oper.keys)
				s.sws.Cancel(oper.keys)
			case opWantsSent:
				s.sw.WantsSent(oper.keys)
			case opBroadcast:
				s.broadcast(ctx, oper.keys)
			default:
				panic("unhandled operation")
			}
		case <-s.idleTick.C:
			s.broadcast(ctx, nil)
		case <-s.periodicSearchTimer.C:
			s.handlePeriodicSearch(ctx)
		case baseTickDelay := <-s.tickDelayReqs:
			s.baseTickDelay = baseTickDelay
		case <-ctx.Done():
			s.handleShutdown()
			return
		}
	}
}
func (s *Session) broadcast(ctx context.Context, wants []cid.Cid) {
	if wants == nil {
		wants = s.sw.PrepareBroadcast()
	}
	s.broadcastWantHaves(ctx, wants)
	if len(wants) > 0 && (s.consecutiveTicks == 0) {
		log.Debugw("FindMorePeers", "session", s.id, "cid", wants[0], "pending", len(wants))
		s.findMorePeers(ctx, wants[0])
	}
	s.resetIdleTick()
	if s.sw.HasLiveWants() {
		s.consecutiveTicks++
	}
}