bitswap first working commit!
Jeromy committed
Aug 26, 2014 at 14:24 UTC
cfdf01d58a5e7b6de6e5b71a92f896083aeef74f
10 files changed
+95
-39
bitswap/bitswap.go
+16
-2
@@ -67,6 +67,7 @@ func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore, r routing.IpfsR
67
routing: r.(*dht.IpfsDHT),
68
meschan: net.GetChannel(swarm.PBWrapper_BITSWAP),
69
haltChan: make(chan struct{}),
70
+ listener: swarm.NewMesListener(),
71
}
72
73
go bs.handleMessages()
@@ -90,7 +91,7 @@ func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
91
ledger := bs.GetLedger(pr.Key())
92
blk, err := bs.getBlock(k, pr, tleft)
93
if err != nil {
93
- u.PErr("%v\n", err)
94
+ u.PErr("getBlock returned: %v\n", err)
95
return
96
}
97
// NOTE: this credits everyone who sends us a block,
@@ -106,6 +107,7 @@ func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
107
108
select {
109
case blkdata := <-valchan:
110
+ close(valchan)
111
return blocks.NewBlock(blkdata)
112
case <-after:
113
return nil, u.ErrTimeout
@@ -113,6 +115,7 @@ func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
115
}
116
117
func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byte, error) {
118
+ u.DOut("[%s] getBlock '%s' from [%s]\n", bs.peer.ID.Pretty(), k.Pretty(), p.ID.Pretty())
119
//
120
mes := new(PBMessage)
121
mes.Id = proto.Uint64(swarm.GenerateMessageID())
@@ -161,6 +164,7 @@ func (bs *BitSwap) handleMessages() {
164
}
165
if pmes.GetResponse() {
166
bs.listener.Respond(pmes.GetId(), mes)
167
+ continue
168
}
169
170
switch pmes.GetType() {
@@ -176,16 +180,20 @@ func (bs *BitSwap) handleMessages() {
180
}
181
182
func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
183
+ u.DOut("handleGetBlock.\n")
184
ledger := bs.GetLedger(p.Key())
185
186
+ u.DOut("finding [%s] in datastore.\n", u.Key(pmes.GetKey()).Pretty())
187
idata, err := bs.datastore.Get(ds.NewKey(pmes.GetKey()))
188
if err != nil {
189
+ u.PErr("handleGetBlock datastore returned: %v\n", err)
190
if err == ds.ErrNotFound {
191
return
192
}
186
- u.PErr("%v\n", err)
193
return
194
}
195
+
196
+ u.DOut("found value!\n")
197
data, ok := idata.([]byte)
198
if !ok {
199
u.PErr("Failed casting data from datastore.")
@@ -193,13 +201,18 @@ func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
201
}
202
203
if ledger.ShouldSend() {
204
+ u.DOut("Sending value back!\n")
205
resp := &Message{
206
Value: data,
207
Response: true,
208
ID: pmes.GetId(),
209
+ Type: PBMessage_GET_BLOCK,
210
+ Success: true,
211
}
212
bs.meschan.Outgoing <- swarm.NewMessage(p, resp.ToProtobuf())
213
ledger.SentBytes(uint64(len(data)))
214
+ } else {
215
+ u.DOut("Ledger decided not to send anything...\n")
216
}
217
}
218
@@ -210,6 +223,7 @@ func (bs *BitSwap) GetLedger(k u.Key) *Ledger {
223
}
224
225
l = new(Ledger)
226
+ l.Strategy = StandardStrategy
227
l.Partner = peer.ID(k)
228
bs.partners[k] = l
229
return l
bitswap/ledger.go
+3
-3
@@ -1,8 +1,6 @@
1
package bitswap
2
3
import (
4
- "math/rand"
5
-
4
peer "github.com/jbenet/go-ipfs/peer"
5
u "github.com/jbenet/go-ipfs/util"
6
@@ -26,13 +24,15 @@ type Ledger struct {
24
25
// WantList is a (bounded, small) set of keys that Partner desires.
26
WantList KeySet
27
+
28
+ Strategy StrategyFunc
29
}
30
31
// LedgerMap lists Ledgers by their Partner key.
32
type LedgerMap map[u.Key]*Ledger
33
34
func (l *Ledger) ShouldSend() bool {
35
- return rand.Float64() <= probabilitySend(l.Accounting.Value())
35
+ return l.Strategy(l.Accounting)
36
}
37
38
func (l *Ledger) SentBytes(n uint64) {
bitswap/message.go
+2
@@ -6,6 +6,7 @@ import (
6
)
7
8
type Message struct {
9
+ Type PBMessage_MessageType
10
ID uint64
11
Response bool
12
Key u.Key
@@ -16,6 +17,7 @@ type Message struct {
17
func (m *Message) ToProtobuf() *PBMessage {
18
pmes := new(PBMessage)
19
pmes.Id = &m.ID
20
+ pmes.Type = &m.Type
21
if m.Response {
22
pmes.Response = proto.Bool(true)
23
}
bitswap/strategy.go
+11
@@ -2,8 +2,19 @@ package bitswap
2
3
import (
4
"math"
5
+ "math/rand"
6
)
7
8
+type StrategyFunc func(debtRatio) bool
9
+
10
+func StandardStrategy(db debtRatio) bool {
11
+ return rand.Float64() <= probabilitySend(db.Value())
12
+}
13
+
14
+func YesManStrategy(db debtRatio) bool {
15
+ return true
16
+}
17
+
18
func probabilitySend(ratio float64) float64 {
19
x := 1 + math.Exp(6-3*ratio)
20
y := 1 / x
blockservice/blockservice.go
+30
-15
@@ -2,6 +2,7 @@ package blockservice
2
3
import (
4
"fmt"
5
+ "time"
6
7
ds "github.com/jbenet/datastore.go"
8
bitswap "github.com/jbenet/go-ipfs/bitswap"
@@ -19,18 +20,27 @@ type BlockService struct {
20
}
21
22
// NewBlockService creates a BlockService with given datastore instance.
22
-func NewBlockService(d ds.Datastore) (*BlockService, error) {
23
+func NewBlockService(d ds.Datastore, rem *bitswap.BitSwap) (*BlockService, error) {
24
if d == nil {
25
return nil, fmt.Errorf("BlockService requires valid datastore")
26
}
26
- return &BlockService{Datastore: d}, nil
27
+ if rem == nil {
28
+ return nil, fmt.Errorf("BlockService requires a valid bitswap")
29
+ }
30
+ return &BlockService{Datastore: d, Remote: rem}, nil
31
}
32
33
// AddBlock adds a particular block to the service, Putting it into the datastore.
34
func (s *BlockService) AddBlock(b *blocks.Block) (u.Key, error) {
35
k := b.Key()
36
dsk := ds.NewKey(string(k))
33
- return k, s.Datastore.Put(dsk, b.Data)
37
+ u.DOut("storing [%s] in datastore\n", k.Pretty())
38
+ err := s.Datastore.Put(dsk, b.Data)
39
+ if err != nil {
40
+ return k, err
41
+ }
42
+ err = s.Remote.HaveBlock(b.Key())
43
+ return k, err
44
}
45
46
// GetBlock retrieves a particular block from the service,
@@ -38,17 +48,22 @@ func (s *BlockService) AddBlock(b *blocks.Block) (u.Key, error) {
48
func (s *BlockService) GetBlock(k u.Key) (*blocks.Block, error) {
49
dsk := ds.NewKey(string(k))
50
datai, err := s.Datastore.Get(dsk)
41
- if err != nil {
42
- return nil, err
51
+ if err == nil {
52
+ bdata, ok := datai.([]byte)
53
+ if !ok {
54
+ return nil, fmt.Errorf("data associated with %s is not a []byte", k)
55
+ }
56
+ return &blocks.Block{
57
+ Multihash: mh.Multihash(k),
58
+ Data: bdata,
59
+ }, nil
60
+ } else if err == ds.ErrNotFound {
61
+ blk, err := s.Remote.GetBlock(k, time.Second*5)
62
+ if err != nil {
63
+ return nil, err
64
+ }
65
+ return blk, nil
66
+ } else {
67
+ return nil, u.ErrNotFound
68
}
44
-
45
- data, ok := datai.([]byte)
46
- if !ok {
47
- return nil, fmt.Errorf("data associated with %s is not a []byte", k)
48
- }
49
-
50
- return &blocks.Block{
51
- Multihash: mh.Multihash(k),
52
- Data: data,
53
- }, nil
69
}
core/core.go
+1
-1
@@ -59,7 +59,7 @@ func NewIpfsNode(cfg *config.Config) (*IpfsNode, error) {
59
return nil, err
60
}
61
62
- bs, err := bserv.NewBlockService(d)
62
+ bs, err := bserv.NewBlockService(d, nil)
63
if err != nil {
64
return nil, err
65
}
routing/dht/dht.go
+14
-8
@@ -53,11 +53,11 @@ type IpfsDHT struct {
53
}
54
55
// NewDHT creates a new DHT object with the given peer as the 'local' host
56
-func NewDHT(p *peer.Peer, net swarm.Network) *IpfsDHT {
56
+func NewDHT(p *peer.Peer, net swarm.Network, dstore ds.Datastore) *IpfsDHT {
57
dht := new(IpfsDHT)
58
dht.network = net
59
dht.netChan = net.GetChannel(swarm.PBWrapper_DHT_MESSAGE)
60
- dht.datastore = ds.NewMapDatastore()
60
+ dht.datastore = dstore
61
dht.self = p
62
dht.providers = NewProviderManager()
63
dht.shutdown = make(chan struct{})
@@ -322,6 +322,7 @@ type providerInfo struct {
322
323
func (dht *IpfsDHT) handleAddProvider(p *peer.Peer, pmes *PBDHTMessage) {
324
key := u.Key(pmes.GetKey())
325
+ u.DOut("[%s] Adding [%s] as a provider for '%s'\n", dht.self.ID.Pretty(), p.ID.Pretty(), peer.ID(key).Pretty())
326
dht.providers.AddProvider(key, p)
327
}
328
@@ -615,12 +616,8 @@ func (dht *IpfsDHT) addPeerList(key u.Key, peers []*PBDHTMessage_PBPeer) []*peer
616
p := dht.network.Find(u.Key(prov.GetId()))
617
if p == nil {
618
u.DOut("given provider %s was not in our network already.\n", peer.ID(prov.GetId()).Pretty())
618
- maddr, err := ma.NewMultiaddr(prov.GetAddr())
619
- if err != nil {
620
- u.PErr("error connecting to new peer: %s\n", err)
621
- continue
622
- }
623
- p, err = dht.network.GetConnection(peer.ID(prov.GetId()), maddr)
619
+ var err error
620
+ p, err = dht.peerFromInfo(prov)
621
if err != nil {
622
u.PErr("error connecting to new peer: %s\n", err)
623
continue
@@ -631,3 +628,12 @@ func (dht *IpfsDHT) addPeerList(key u.Key, peers []*PBDHTMessage_PBPeer) []*peer
628
}
629
return provArr
630
}
631
+
632
+func (dht *IpfsDHT) peerFromInfo(pbp *PBDHTMessage_PBPeer) (*peer.Peer, error) {
633
+ maddr, err := ma.NewMultiaddr(pbp.GetAddr())
634
+ if err != nil {
635
+ return nil, err
636
+ }
637
+
638
+ return dht.network.GetConnection(peer.ID(pbp.GetId()), maddr)
639
+}
routing/dht/dht_test.go
+6
-5
@@ -3,6 +3,7 @@ package dht
3
import (
4
"testing"
5
6
+ ds "github.com/jbenet/datastore.go"
7
peer "github.com/jbenet/go-ipfs/peer"
8
swarm "github.com/jbenet/go-ipfs/swarm"
9
u "github.com/jbenet/go-ipfs/util"
@@ -37,7 +38,7 @@ func setupDHTS(n int, t *testing.T) ([]*ma.Multiaddr, []*peer.Peer, []*IpfsDHT)
38
if err != nil {
39
t.Fatal(err)
40
}
40
- d := NewDHT(peers[i], net)
41
+ d := NewDHT(peers[i], net, ds.NewMapDatastore())
42
dhts = append(dhts, d)
43
d.Start()
44
}
@@ -69,14 +70,14 @@ func TestPing(t *testing.T) {
70
if err != nil {
71
t.Fatal(err)
72
}
72
- dhtA := NewDHT(peerA, neta)
73
+ dhtA := NewDHT(peerA, neta, ds.NewMapDatastore())
74
75
netb := swarm.NewSwarm(peerB)
76
err = netb.Listen()
77
if err != nil {
78
t.Fatal(err)
79
}
79
- dhtB := NewDHT(peerB, netb)
80
+ dhtB := NewDHT(peerB, netb, ds.NewMapDatastore())
81
82
dhtA.Start()
83
dhtB.Start()
@@ -120,14 +121,14 @@ func TestValueGetSet(t *testing.T) {
121
if err != nil {
122
t.Fatal(err)
123
}
123
- dhtA := NewDHT(peerA, neta)
124
+ dhtA := NewDHT(peerA, neta, ds.NewMapDatastore())
125
126
netb := swarm.NewSwarm(peerB)
127
err = netb.Listen()
128
if err != nil {
129
t.Fatal(err)
130
}
130
- dhtB := NewDHT(peerB, netb)
131
+ dhtB := NewDHT(peerB, netb, ds.NewMapDatastore())
132
133
dhtA.Start()
134
dhtB.Start()
routing/dht/ext_test.go
+4
-3
@@ -7,6 +7,7 @@ import (
7
8
"code.google.com/p/goprotobuf/proto"
9
10
+ ds "github.com/jbenet/datastore.go"
11
peer "github.com/jbenet/go-ipfs/peer"
12
swarm "github.com/jbenet/go-ipfs/swarm"
13
u "github.com/jbenet/go-ipfs/util"
@@ -89,7 +90,7 @@ func TestGetFailures(t *testing.T) {
90
local := new(peer.Peer)
91
local.ID = peer.ID("test_peer")
92
92
- d := NewDHT(local, fn)
93
+ d := NewDHT(local, fn, ds.NewMapDatastore())
94
95
other := &peer.Peer{ID: peer.ID("other_peer")}
96
@@ -177,7 +178,7 @@ func TestNotFound(t *testing.T) {
178
local := new(peer.Peer)
179
local.ID = peer.ID("test_peer")
180
180
- d := NewDHT(local, fn)
181
+ d := NewDHT(local, fn, ds.NewMapDatastore())
182
d.Start()
183
184
var ps []*peer.Peer
@@ -239,7 +240,7 @@ func TestLessThanKResponses(t *testing.T) {
240
local := new(peer.Peer)
241
local.ID = peer.ID("test_peer")
242
242
- d := NewDHT(local, fn)
243
+ d := NewDHT(local, fn, ds.NewMapDatastore())
244
d.Start()
245
246
var ps []*peer.Peer
routing/dht/routing.go
+8
-2
@@ -63,7 +63,7 @@ func (dht *IpfsDHT) GetValue(key u.Key, timeout time.Duration) ([]byte, error) {
63
val, err := dht.getLocal(key)
64
if err == nil {
65
ll.Success = true
66
- u.DOut("Found local, returning.")
66
+ u.DOut("Found local, returning.\n")
67
return val, nil
68
}
69
@@ -218,6 +218,9 @@ func (dht *IpfsDHT) FindProvidersAsync(key u.Key, count int, timeout time.Durati
218
//TODO: this function could also be done asynchronously
219
func (dht *IpfsDHT) addPeerListAsync(k u.Key, peers []*PBDHTMessage_PBPeer, ps *peerSet, count int, out chan *peer.Peer) {
220
for _, pbp := range peers {
221
+ if peer.ID(pbp.GetId()).Equal(dht.self.ID) {
222
+ continue
223
+ }
224
maddr, err := ma.NewMultiaddr(pbp.GetAddr())
225
if err != nil {
226
u.PErr("%v\n", err)
@@ -256,11 +259,14 @@ func (dht *IpfsDHT) FindProviders(key u.Key, timeout time.Duration) ([]*peer.Pee
259
return nil, err
260
}
261
if pmes.GetSuccess() {
262
+ u.DOut("Got providers back from findProviders call!\n")
263
provs := dht.addPeerList(key, pmes.GetPeers())
264
ll.Success = true
265
return provs, nil
266
}
267
268
+ u.DOut("Didnt get providers, just closer peers.\n")
269
+
270
closer := pmes.GetPeers()
271
if len(closer) == 0 {
272
level++
@@ -337,7 +343,7 @@ func (dht *IpfsDHT) FindPeer(id peer.ID, timeout time.Duration) (*peer.Peer, err
343
// Ping a peer, log the time it took
344
func (dht *IpfsDHT) Ping(p *peer.Peer, timeout time.Duration) error {
345
// Thoughts: maybe this should accept an ID and do a peer lookup?
340
- u.DOut("Enter Ping.")
346
+ u.DOut("Enter Ping.\n")
347
348
pmes := Message{ID: swarm.GenerateMessageID(), Type: PBDHTMessage_PING}
349
mes := swarm.NewMessage(p, pmes.ToProtobuf())