sessionmanager.go 3.89 KB
Newer Older
1 2 3
package sessionmanager

import (
4
	"context"
5
	"sync"
6
	"time"
7

8
	cid "github.com/ipfs/go-cid"
9
	delay "github.com/ipfs/go-ipfs-delay"
10

11
	notifications "github.com/ipfs/go-bitswap/notifications"
12
	bssession "github.com/ipfs/go-bitswap/session"
13
	exchange "github.com/ipfs/go-ipfs-exchange-interface"
Raúl Kripalani's avatar
Raúl Kripalani committed
14
	peer "github.com/libp2p/go-libp2p-core/peer"
15 16
)

17 18 19
// Session is a session that is managed by the session manager
type Session interface {
	exchange.Fetcher
20
	ReceiveFrom(peer.ID, []cid.Cid)
21
	IsWanted(cid.Cid) bool
22 23
}

24
type sesTrk struct {
25 26
	session Session
	pm      bssession.PeerManager
27
	srs     bssession.RequestSplitter
28 29
}

30
// SessionFactory generates a new session for the SessionManager to track.
31
type SessionFactory func(ctx context.Context, id uint64, pm bssession.PeerManager, srs bssession.RequestSplitter, notif notifications.PubSub, provSearchDelay time.Duration, rebroadcastDelay delay.D) Session
32 33 34

// RequestSplitterFactory generates a new request splitter for a session.
type RequestSplitterFactory func(ctx context.Context) bssession.RequestSplitter
35 36 37 38

// PeerManagerFactory generates a new peer manager for a session.
type PeerManagerFactory func(ctx context.Context, id uint64) bssession.PeerManager

39 40
// SessionManager is responsible for creating, managing, and dispatching to
// sessions.
41
type SessionManager struct {
42 43 44 45
	ctx                    context.Context
	sessionFactory         SessionFactory
	peerManagerFactory     PeerManagerFactory
	requestSplitterFactory RequestSplitterFactory
46
	notif                  notifications.PubSub
47

48 49
	// Sessions
	sessLk   sync.Mutex
50
	sessions []sesTrk
51 52 53 54 55 56

	// Session Index
	sessIDLk sync.Mutex
	sessID   uint64
}

57
// New creates a new SessionManager.
58 59
func New(ctx context.Context, sessionFactory SessionFactory, peerManagerFactory PeerManagerFactory,
	requestSplitterFactory RequestSplitterFactory, notif notifications.PubSub) *SessionManager {
60
	return &SessionManager{
61 62 63 64
		ctx:                    ctx,
		sessionFactory:         sessionFactory,
		peerManagerFactory:     peerManagerFactory,
		requestSplitterFactory: requestSplitterFactory,
65
		notif:                  notif,
66
	}
67 68
}

69 70
// NewSession initializes a session with the given context, and adds to the
// session manager.
71 72 73
func (sm *SessionManager) NewSession(ctx context.Context,
	provSearchDelay time.Duration,
	rebroadcastDelay delay.D) exchange.Fetcher {
74 75 76
	id := sm.GetNextSessionID()
	sessionctx, cancel := context.WithCancel(ctx)

77
	pm := sm.peerManagerFactory(sessionctx, id)
78
	srs := sm.requestSplitterFactory(sessionctx)
79
	session := sm.sessionFactory(sessionctx, id, pm, srs, sm.notif, provSearchDelay, rebroadcastDelay)
80
	tracked := sesTrk{session, pm, srs}
81
	sm.sessLk.Lock()
82
	sm.sessions = append(sm.sessions, tracked)
83
	sm.sessLk.Unlock()
84
	go func() {
hannahhoward's avatar
hannahhoward committed
85 86 87
		defer cancel()
		select {
		case <-sm.ctx.Done():
88
			sm.removeSession(tracked)
hannahhoward's avatar
hannahhoward committed
89
		case <-ctx.Done():
90
			sm.removeSession(tracked)
91 92 93 94
		}
	}()

	return session
95 96
}

97
func (sm *SessionManager) removeSession(session sesTrk) {
98 99 100 101 102 103 104 105 106 107 108
	sm.sessLk.Lock()
	defer sm.sessLk.Unlock()
	for i := 0; i < len(sm.sessions); i++ {
		if sm.sessions[i] == session {
			sm.sessions[i] = sm.sessions[len(sm.sessions)-1]
			sm.sessions = sm.sessions[:len(sm.sessions)-1]
			return
		}
	}
}

109
// GetNextSessionID returns the next sequentional identifier for a session.
110 111 112 113 114 115 116
func (sm *SessionManager) GetNextSessionID() uint64 {
	sm.sessIDLk.Lock()
	defer sm.sessIDLk.Unlock()
	sm.sessID++
	return sm.sessID
}

117
// ReceiveFrom receives block CIDs from a peer and dispatches to sessions.
118
func (sm *SessionManager) ReceiveFrom(from peer.ID, ks []cid.Cid) {
119 120 121 122
	sm.sessLk.Lock()
	defer sm.sessLk.Unlock()

	for _, s := range sm.sessions {
123
		s.session.ReceiveFrom(from, ks)
124 125
	}
}
126

127
// IsWanted indicates whether any of the sessions are waiting to receive
128
// the block with the given CID.
129
func (sm *SessionManager) IsWanted(cid cid.Cid) bool {
130 131 132 133
	sm.sessLk.Lock()
	defer sm.sessLk.Unlock()

	for _, s := range sm.sessions {
134
		if s.session.IsWanted(cid) {
135 136 137 138 139
			return true
		}
	}
	return false
}