clarify synhronization constructs
Jeromy committed
May 19, 2015 at 11:26 UTC
b71a0aced018909ca2d56797394e4ae3b02516b8
1 file changed
+17
-21
exchange/bitswap/wantmanager.go
+17
-21
@@ -14,23 +14,17 @@ import (
14
)
15
16
type WantManager struct {
17
- receiver bsnet.Receiver
18
-
19
- incoming chan []*bsmsg.Entry
20
-
21
- // notification channel for new peers connecting
22
- connect chan peer.ID
23
-
24
- // notification channel for peers disconnecting
25
- disconnect chan peer.ID
17
+ // sync channels for Run loop
18
+ incoming chan []*bsmsg.Entry
19
+ connect chan peer.ID // notification channel for new peers connecting
20
+ disconnect chan peer.ID // notification channel for peers disconnecting
21
22
+ // synchronized by Run loop, only touch inside there
23
peers map[peer.ID]*msgQueue
28
-
29
- wl *wantlist.Wantlist
24
+ wl *wantlist.Wantlist
25
26
network bsnet.BitSwapNetwork
32
-
33
- ctx context.Context
27
+ ctx context.Context
28
}
29
30
func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantManager {
@@ -58,8 +52,9 @@ type cancellation struct {
52
type msgQueue struct {
53
p peer.ID
54
61
- outlk sync.Mutex
62
- out bsmsg.BitSwapMessage
55
+ outlk sync.Mutex
56
+ out bsmsg.BitSwapMessage
57
+ network bsnet.BitSwapNetwork
58
59
work chan struct{}
60
done chan struct{}
@@ -112,7 +107,7 @@ func (pm *WantManager) startPeerHandler(p peer.ID) *msgQueue {
107
return nil
108
}
109
115
- mq := newMsgQueue(p)
110
+ mq := pm.newMsgQueue(p)
111
112
// new peer, we will want to give them our full wantlist
113
fullwantlist := bsmsg.New(true)
@@ -123,7 +118,7 @@ func (pm *WantManager) startPeerHandler(p peer.ID) *msgQueue {
118
mq.work <- struct{}{}
119
120
pm.peers[p] = mq
126
- go pm.runQueue(mq)
121
+ go mq.runQueue(pm.ctx)
122
return mq
123
}
124
@@ -138,12 +133,12 @@ func (pm *WantManager) stopPeerHandler(p peer.ID) {
133
delete(pm.peers, p)
134
}
135
141
-func (pm *WantManager) runQueue(mq *msgQueue) {
136
+func (mq *msgQueue) runQueue(ctx context.Context) {
137
for {
138
select {
139
case <-mq.work: // there is work to be done
140
146
- err := pm.network.ConnectTo(pm.ctx, mq.p)
141
+ err := mq.network.ConnectTo(ctx, mq.p)
142
if err != nil {
143
log.Errorf("cant connect to peer %s: %s", mq.p, err)
144
// TODO: cant connect, what now?
@@ -161,7 +156,7 @@ func (pm *WantManager) runQueue(mq *msgQueue) {
156
mq.outlk.Unlock()
157
158
// send wantlist updates
164
- err = pm.network.SendMessage(pm.ctx, mq.p, wlm)
159
+ err = mq.network.SendMessage(ctx, mq.p, wlm)
160
if err != nil {
161
log.Error("bitswap send error: ", err)
162
// TODO: what do we do if this fails?
@@ -224,10 +219,11 @@ func (pm *WantManager) Run() {
219
}
220
}
221
227
-func newMsgQueue(p peer.ID) *msgQueue {
222
+func (wm *WantManager) newMsgQueue(p peer.ID) *msgQueue {
223
mq := new(msgQueue)
224
mq.done = make(chan struct{})
225
mq.work = make(chan struct{}, 1)
226
+ mq.network = wm.network
227
mq.p = p
228
229
return mq