refac(bitswap) inline helper methods
for readability
Brian Tiger Chow committed
Sep 17, 2014 at 18:22 UTC
b8fcc137a6c5114c6cf661a1096e7e09502f4573
1 file changed
+42
-73
bitswap/bitswap.go
+42
-73
@@ -10,6 +10,7 @@ import (
10
bsmsg "github.com/jbenet/go-ipfs/bitswap/message"
11
bsnet "github.com/jbenet/go-ipfs/bitswap/network"
12
notifications "github.com/jbenet/go-ipfs/bitswap/notifications"
13
+ strategy "github.com/jbenet/go-ipfs/bitswap/strategy"
14
blocks "github.com/jbenet/go-ipfs/blocks"
15
blockstore "github.com/jbenet/go-ipfs/blockstore"
16
peer "github.com/jbenet/go-ipfs/peer"
@@ -36,6 +37,7 @@ type bitswap struct {
37
sender bsnet.NetworkAdapter
38
39
// blockstore is the local database
40
+ // NB: ensure threadsafety
41
blockstore blockstore.Blockstore
42
43
// routing interface for communication
@@ -43,10 +45,11 @@ type bitswap struct {
45
46
notifications notifications.PubSub
47
46
- // partners is a map of currently active bitswap relationships.
47
- // The Ledger has the peer.ID, and the peer connection works through net.
48
- // Ledgers of known relationships (active or inactive) stored in datastore.
49
- // Changes to the Ledger should be committed to the datastore.
48
+ // strategist listens to network traffic and makes decisions about how to
49
+ // interact with partners.
50
+ // TODO(brian): save the strategist's state to the datastore
51
+ strategist strategy.Strategist
52
+
53
partners ledgerMap
54
55
// haveList is the set of keys we have values for. a map for fast lookups.
@@ -60,6 +63,7 @@ type bitswap struct {
63
// NewSession initializes a bitswap session.
64
func NewSession(parent context.Context, s bsnet.NetworkService, p *peer.Peer, d ds.Datastore, directory Directory) Exchange {
65
66
+ // FIXME(brian): instantiate a concrete Strategist
67
receiver := bsnet.Forwarder{}
68
bs := &bitswap{
69
peer: p,
@@ -79,7 +83,6 @@ func NewSession(parent context.Context, s bsnet.NetworkService, p *peer.Peer, d
83
// GetBlock attempts to retrieve a particular block from peers, within timeout.
84
func (bs *bitswap) Block(k u.Key, timeout time.Duration) (
85
*blocks.Block, error) {
82
- u.DOut("Bitswap GetBlock: '%s'\n", k.Pretty())
86
begin := time.Now()
87
tleft := timeout - time.Now().Sub(begin)
88
provs_ch := bs.routing.FindProvidersAsync(k, 20, timeout)
@@ -93,7 +96,6 @@ func (bs *bitswap) Block(k u.Key, timeout time.Duration) (
96
go func(pr *peer.Peer) {
97
blk, err := bs.getBlock(k, pr, tleft)
98
if err != nil {
96
- u.PErr("getBlock returned: %v\n", err)
99
return
100
}
101
select {
@@ -114,113 +116,80 @@ func (bs *bitswap) Block(k u.Key, timeout time.Duration) (
116
}
117
118
func (bs *bitswap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) (*blocks.Block, error) {
117
- u.DOut("[%s] getBlock '%s' from [%s]\n", bs.peer.ID.Pretty(), k.Pretty(), p.ID.Pretty())
119
120
ctx, _ := context.WithTimeout(context.Background(), timeout)
121
blockChannel := bs.notifications.Subscribe(ctx, k)
122
123
message := bsmsg.New()
124
message.AppendWanted(k)
125
+
126
+ // FIXME(brian): register the accountant on the service wrapper to ensure
127
+ // that accounting is _always_ performed when SendMessage and
128
+ // ReceiveMessage are called
129
bs.sender.SendMessage(ctx, p, message)
130
+ bs.strategist.MessageSent(p, message)
131
132
block, ok := <-blockChannel
133
if !ok {
128
- u.PErr("getBlock for '%s' timed out.\n", k.Pretty())
134
return nil, u.ErrTimeout
135
}
136
return &block, nil
137
}
138
139
+func (bs *bitswap) sendToPeersThatWant(block blocks.Block) {
140
+ for _, p := range bs.strategist.Peers() {
141
+ if bs.strategist.IsWantedByPeer(block.Key(), p) {
142
+ if bs.strategist.ShouldSendToPeer(block.Key(), p) {
143
+ go bs.send(p, block)
144
+ }
145
+ }
146
+ }
147
+}
148
+
149
// HasBlock announces the existance of a block to bitswap, potentially sending
150
// it to peers (Partners) whose WantLists include it.
151
func (bs *bitswap) HasBlock(blk blocks.Block) error {
137
- go func() {
138
- for _, ledger := range bs.partners {
139
- if ledger.WantListContains(blk.Key()) {
140
- //send block to node
141
- if ledger.ShouldSend() {
142
- bs.sendBlock(ledger.Partner, blk)
143
- }
144
- }
145
- }
146
- }()
152
+ go bs.sendToPeersThatWant(blk)
153
return bs.routing.Provide(blk.Key())
154
}
155
156
// TODO(brian): get a return value
151
-func (bs *bitswap) sendBlock(p *peer.Peer, b blocks.Block) {
152
- u.DOut("Sending block to peer.\n")
157
+func (bs *bitswap) send(p *peer.Peer, b blocks.Block) {
158
message := bsmsg.New()
154
- // TODO(brian): change interface to accept value instead of pointer
159
message.AppendBlock(b)
160
+ // FIXME(brian): pass ctx
161
bs.sender.SendMessage(context.Background(), p, message)
157
-}
158
-
159
-// peerWantsBlock will check if we have the block in question,
160
-// and then if we do, check the ledger for whether or not we should send it.
161
-func (bs *bitswap) peerWantsBlock(p *peer.Peer, wanted u.Key) {
162
- u.DOut("peer [%s] wants block [%s]\n", p.ID.Pretty(), wanted.Pretty())
163
-
164
- ledger := bs.getLedger(p)
165
-
166
- if !ledger.ShouldSend() {
167
- return
168
- }
169
-
170
- block, err := bs.blockstore.Get(wanted)
171
- if err != nil { // TODO(brian): log/return the error
172
- ledger.Wants(wanted)
173
- return
174
- }
175
- bs.sendBlock(p, *block)
176
- ledger.SentBytes(numBytes(*block))
177
-}
178
-
179
-// TODO(brian): return error
180
-func (bs *bitswap) blockReceive(p *peer.Peer, blk blocks.Block) {
181
- u.DOut("blockReceive: %s\n", blk.Key().Pretty())
182
- err := bs.blockstore.Put(blk)
183
- if err != nil {
184
- u.PErr("blockReceive error: %v\n", err)
185
- return
186
- }
187
-
188
- bs.notifications.Publish(blk)
189
-
190
- ledger := bs.getLedger(p)
191
- ledger.ReceivedBytes(len(blk.Data))
192
-}
193
-
194
-func (bs *bitswap) getLedger(p *peer.Peer) *ledger {
195
- l, ok := bs.partners[p.Key()]
196
- if ok {
197
- return l
198
- }
199
-
200
- l = new(ledger)
201
- l.Strategy = bs.strategy
202
- l.Partner = p
203
- bs.partners[p.Key()] = l
204
- return l
162
+ bs.strategist.MessageSent(p, message)
163
}
164
165
func (bs *bitswap) Halt() {
166
bs.haltChan <- struct{}{}
167
}
168
169
+// TODO(brian): handle errors
170
func (bs *bitswap) ReceiveMessage(
171
ctx context.Context, sender *peer.Peer, incoming bsmsg.BitSwapMessage) (
172
*peer.Peer, bsmsg.BitSwapMessage, error) {
173
+
174
+ bs.strategist.MessageReceived(sender, incoming)
175
+
176
if incoming.Blocks() != nil {
177
for _, block := range incoming.Blocks() {
216
- go bs.blockReceive(sender, block)
178
+ go bs.blockstore.Put(block) // FIXME(brian): err ignored
179
+ go bs.notifications.Publish(block)
180
}
181
}
182
183
if incoming.Wantlist() != nil {
221
- for _, want := range incoming.Wantlist() {
222
- // TODO(brian): return the block synchronously
223
- go bs.peerWantsBlock(sender, want)
184
+ for _, key := range incoming.Wantlist() {
185
+ if bs.strategist.ShouldSendToPeer(key, sender) {
186
+ block, errBlockNotFound := bs.blockstore.Get(key)
187
+ if errBlockNotFound != nil {
188
+ // TODO(brian): log/return the error
189
+ continue
190
+ }
191
+ go bs.send(sender, *block)
192
+ }
193
}
194
}
195
return nil, nil, errors.New("TODO implement")