move some variables into strategy
Jeromy committed
Nov 19, 2014 at 23:32 UTC
0abc72c0625374c48db5ee7d2e97a4d120fddaff
7 files changed
+60
-8
blockservice/blockservice.go
+5
@@ -96,6 +96,11 @@ func (s *BlockService) GetBlock(ctx context.Context, k u.Key) (*blocks.Block, er
96
}
97
}
98
99
+func (s *BlockService) GetBlocks(ctx context.Context, ks []u.Key) (<-chan blocks.Block, error) {
100
+ // TODO:
101
+ return nil, nil
102
+}
103
+
104
// DeleteBlock deletes a block in the blockservice from the datastore
105
func (s *BlockService) DeleteBlock(k u.Key) error {
106
return s.Datastore.Delete(k.DsKey())
exchange/bitswap/bitswap.go
+10
-8
@@ -157,9 +157,10 @@ func (bs *bitswap) run(ctx context.Context) {
157
rebroadcastTime := time.Second * 5
158
159
var providers <-chan peer.Peer // NB: must be initialized to zero value
160
- broadcastSignal := time.After(rebroadcastPeriod)
161
- unsentKeys := 0
160
+ broadcastSignal := time.After(bs.strategy.GetRebroadcastDelay())
161
162
+ // Number of unsent keys for the current batch
163
+ unsentKeys := 0
164
for {
165
select {
166
case <-broadcastSignal:
@@ -170,14 +171,14 @@ func (bs *bitswap) run(ctx context.Context) {
171
if providers == nil {
172
// rely on semi randomness of maps
173
firstKey := wantlist[0]
173
- providers = bs.routing.FindProvidersAsync(ctx, firstKey, 6)
174
+ providers = bs.routing.FindProvidersAsync(ctx, firstKey, maxProvidersPerRequest)
175
}
176
err := bs.sendWantListTo(ctx, providers)
177
if err != nil {
178
log.Errorf("error sending wantlist: %s", err)
179
}
180
providers = nil
180
- broadcastSignal = time.After(rebroadcastPeriod)
181
+ broadcastSignal = time.After(bs.strategy.GetRebroadcastDelay())
182
183
case k := <-bs.blockRequests:
184
if unsentKeys == 0 {
@@ -185,19 +186,19 @@ func (bs *bitswap) run(ctx context.Context) {
186
}
187
unsentKeys++
188
188
- if unsentKeys >= numKeysPerBatch {
189
+ if unsentKeys >= bs.strategy.GetBatchSize() {
190
// send wantlist to providers
191
err := bs.sendWantListTo(ctx, providers)
192
if err != nil {
193
log.Errorf("error sending wantlist: %s", err)
194
}
195
unsentKeys = 0
195
- broadcastSignal = time.After(rebroadcastPeriod)
196
+ broadcastSignal = time.After(bs.strategy.GetRebroadcastDelay())
197
providers = nil
198
} else {
199
// set a timeout to wait for more blocks or send current wantlist
200
200
- broadcastSignal = time.After(batchDelay)
201
+ broadcastSignal = time.After(bs.strategy.GetBatchDelay())
202
}
203
case <-ctx.Done():
204
return
@@ -217,7 +218,7 @@ func (bs *bitswap) HasBlock(ctx context.Context, blk blocks.Block) error {
218
// TODO(brian): handle errors
219
func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsmsg.BitSwapMessage) (
220
peer.Peer, bsmsg.BitSwapMessage) {
220
- log.Debugf("ReceiveMessage from %v", p.Key())
221
+ log.Debugf("ReceiveMessage from %s", p)
222
log.Debugf("Message wantlist: %v", incoming.Wantlist())
223
224
if p == nil {
@@ -239,6 +240,7 @@ func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsm
240
for _, block := range incoming.Blocks() {
241
// TODO verify blocks?
242
if err := bs.blockstore.Put(&block); err != nil {
243
+ log.Criticalf("error putting block: %s", err)
244
continue // FIXME(brian): err ignored
245
}
246
bs.notifications.Publish(block)
exchange/bitswap/strategy/interface.go
+7
@@ -1,6 +1,8 @@
1
package strategy
2
3
import (
4
+ "time"
5
+
6
bsmsg "github.com/jbenet/go-ipfs/exchange/bitswap/message"
7
peer "github.com/jbenet/go-ipfs/peer"
8
u "github.com/jbenet/go-ipfs/util"
@@ -29,4 +31,9 @@ type Strategy interface {
31
NumBytesSentTo(peer.Peer) uint64
32
33
NumBytesReceivedFrom(peer.Peer) uint64
34
+
35
+ // Values determining bitswap behavioural patterns
36
+ GetBatchSize() int
37
+ GetBatchDelay() time.Duration
38
+ GetRebroadcastDelay() time.Duration
39
}
exchange/bitswap/strategy/strategy.go
+13
@@ -3,6 +3,7 @@ package strategy
3
import (
4
"errors"
5
"sync"
6
+ "time"
7
8
bsmsg "github.com/jbenet/go-ipfs/exchange/bitswap/message"
9
peer "github.com/jbenet/go-ipfs/peer"
@@ -139,3 +140,15 @@ func (s *strategist) ledger(p peer.Peer) *ledger {
140
}
141
return l
142
}
143
+
144
+func (s *strategist) GetBatchSize() int {
145
+ return 10
146
+}
147
+
148
+func (s *strategist) GetBatchDelay() time.Duration {
149
+ return time.Millisecond * 3
150
+}
151
+
152
+func (s *strategist) GetRebroadcastDelay() time.Duration {
153
+ return time.Second * 2
154
+}
net/interface.go
+4
@@ -42,6 +42,10 @@ type Network interface {
42
// the network since it was instantiated
43
GetBandwidthTotals() (uint64, uint64)
44
45
+ // GetMessageCounts returns the total number of messages passed through
46
+ // the network since it was instantiated
47
+ GetMessageCounts() (uint64, uint64)
48
+
49
// SendMessage sends given Message out
50
SendMessage(msg.NetMessage) error
51
net/mux/mux.go
+16
@@ -45,9 +45,11 @@ type Muxer struct {
45
46
bwiLock sync.Mutex
47
bwIn uint64
48
+ msgIn uint64
49
50
bwoLock sync.Mutex
51
bwOut uint64
52
+ msgOut uint64
53
54
*msg.Pipe
55
ctxc.ContextCloser
@@ -76,6 +78,18 @@ func (m *Muxer) GetPipe() *msg.Pipe {
78
return m.Pipe
79
}
80
81
+// GetMessageCounts return the in/out message count measured over this muxer.
82
+func (m *Muxer) GetMessageCounts() (in uint64, out uint64) {
83
+ m.bwiLock.Lock()
84
+ in = m.msgIn
85
+ m.bwiLock.Unlock()
86
+
87
+ m.bwoLock.Lock()
88
+ out = m.msgOut
89
+ m.bwoLock.Unlock()
90
+ return
91
+}
92
+
93
// GetBandwidthTotals return the in/out bandwidth measured over this muxer.
94
func (m *Muxer) GetBandwidthTotals() (in uint64, out uint64) {
95
m.bwiLock.Lock()
@@ -125,6 +139,7 @@ func (m *Muxer) handleIncomingMessage(m1 msg.NetMessage) {
139
m.bwiLock.Lock()
140
// TODO: compensate for overhead
141
m.bwIn += uint64(len(m1.Data()))
142
+ m.msgIn++
143
m.bwiLock.Unlock()
144
145
data, pid, err := unwrapData(m1.Data())
@@ -182,6 +197,7 @@ func (m *Muxer) handleOutgoingMessage(pid pb.ProtocolID, m1 msg.NetMessage) {
197
// TODO: compensate for overhead
198
// TODO(jbenet): switch this to a goroutine to prevent sync waiting.
199
m.bwOut += uint64(len(data))
200
+ m.msgOut++
201
m.bwoLock.Unlock()
202
203
m2 := msg.New(m1.Peer(), data)
net/net.go
+5
@@ -110,6 +110,11 @@ func (n *IpfsNetwork) GetBandwidthTotals() (in uint64, out uint64) {
110
return n.muxer.GetBandwidthTotals()
111
}
112
113
+// GetBandwidthTotals returns the total amount of messages transferred
114
+func (n *IpfsNetwork) GetMessageCounts() (in uint64, out uint64) {
115
+ return n.muxer.GetMessageCounts()
116
+}
117
+
118
// ListenAddresses returns a list of addresses at which this network listens.
119
func (n *IpfsNetwork) ListenAddresses() []ma.Multiaddr {
120
return n.swarm.ListenAddresses()