basic implementation of bitswap, needs testing/verification that it works
Jeromy committed
Aug 24, 2014 at 18:13 UTC
91e4675cbfd89e5df1b50168c7aac26cc57c656f
10 files changed
+221
-65
bitswap/bitswap.go
+105
-12
@@ -10,7 +10,6 @@ import (
10
11
ds "github.com/jbenet/datastore.go"
12
13
- "errors"
13
"time"
14
)
15
@@ -74,24 +73,52 @@ func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore, r routing.IpfsR
73
}
74
75
// GetBlock attempts to retrieve a particular block from peers, within timeout.
77
-func (bs *BitSwap) GetBlock(k u.Key, timeout time.Time) (
76
+func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
77
*blocks.Block, error) {
78
begin := time.Now()
80
- _, err := bs.routing.FindProviders(k, timeout)
79
+ provs, err := bs.routing.FindProviders(k, timeout)
80
if err != nil {
81
u.PErr("GetBlock error: %s\n", err)
83
- return
82
+ return nil, err
83
+ }
84
+ tleft := timeout - time.Now().Sub(begin)
85
+
86
+ valchan := make(chan []byte)
87
+ after := time.After(tleft)
88
+ for _, p := range provs {
89
+ go func(pr *peer.Peer) {
90
+ ledger := bs.GetLedger(pr.Key())
91
+ blk, err := bs.getBlock(k, pr, tleft)
92
+ if err != nil {
93
+ u.PErr("%v\n", err)
94
+ return
95
+ }
96
+ // NOTE: this credits everyone who sends us a block,
97
+ // even if we dont use it
98
+ ledger.ReceivedBytes(uint64(len(blk)))
99
+ select {
100
+ case valchan <- blk:
101
+ default:
102
+ }
103
+ }(p)
104
+ }
105
+
106
+ select {
107
+ case blkdata := <-valchan:
108
+ return blocks.NewBlock(blkdata)
109
+ case <-after:
110
+ return nil, u.ErrTimeout
111
}
85
- tleft := timeout.Sub(time.Now().Sub(begin))
86
- return nil, errors.New("not implemented")
112
}
113
114
func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byte, error) {
115
+ //
116
mes := new(PBMessage)
91
- mes.Id = proto.Uint64(swarm.GenerateID())
92
- mes.Key = proto.String(k)
117
+ mes.Id = proto.Uint64(swarm.GenerateMessageID())
118
+ mes.Key = proto.String(string(k))
119
typ := PBMessage_GET_BLOCK
120
mes.Type = &typ
121
+ //
122
123
after := time.After(timeout)
124
resp := bs.listener.Listen(mes.GetId(), 1, timeout)
@@ -100,23 +127,89 @@ func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byt
127
128
select {
129
case resp_mes := <-resp:
130
+ pmes := new(PBMessage)
131
+ err := proto.Unmarshal(resp_mes.Data, pmes)
132
+ if err != nil {
133
+ return nil, err
134
+ }
135
+ if pmes.GetSuccess() {
136
+ return pmes.GetValue(), nil
137
+ }
138
+ return nil, u.ErrNotFound
139
case <-after:
104
- u.PErr("getBlock for '%s' timed out.", k)
140
+ u.PErr("getBlock for '%s' timed out.\n", k)
141
return nil, u.ErrTimeout
142
}
143
}
144
145
// HaveBlock announces the existance of a block to BitSwap, potentially sending
146
// it to peers (Partners) whose WantLists include it.
111
-func (bs *BitSwap) HaveBlock(k u.Key) (*blocks.Block, error) {
112
- return nil, errors.New("not implemented")
147
+func (bs *BitSwap) HaveBlock(k u.Key) error {
148
+ return bs.routing.Provide(k)
149
}
150
151
func (bs *BitSwap) handleMessages() {
152
for {
153
select {
118
- case mes := bs.meschan.Incoming:
154
+ case mes := <-bs.meschan.Incoming:
155
+ pmes := new(PBMessage)
156
+ err := proto.Unmarshal(mes.Data, pmes)
157
+ if err != nil {
158
+ u.PErr("%v\n", err)
159
+ continue
160
+ }
161
+ if pmes.GetResponse() {
162
+ bs.listener.Respond(pmes.GetId(), mes)
163
+ }
164
+
165
+ switch pmes.GetType() {
166
+ case PBMessage_GET_BLOCK:
167
+ go bs.handleGetBlock(mes.Peer, pmes)
168
+ default:
169
+ u.PErr("Invalid message type.\n")
170
+ }
171
case <-bs.haltChan:
172
+ return
173
}
174
}
175
}
176
+
177
+func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
178
+ ledger := bs.GetLedger(p.Key())
179
+
180
+ idata, err := bs.datastore.Get(ds.NewKey(pmes.GetKey()))
181
+ if err != nil {
182
+ if err == ds.ErrNotFound {
183
+ return
184
+ }
185
+ u.PErr("%v\n", err)
186
+ return
187
+ }
188
+ data, ok := idata.([]byte)
189
+ if !ok {
190
+ u.PErr("Failed casting data from datastore.")
191
+ return
192
+ }
193
+
194
+ if ledger.ShouldSend() {
195
+ resp := &Message{
196
+ Value: data,
197
+ Response: true,
198
+ ID: pmes.GetId(),
199
+ }
200
+ bs.meschan.Outgoing <- swarm.NewMessage(p, resp.ToProtobuf())
201
+ ledger.SentBytes(uint64(len(data)))
202
+ }
203
+}
204
+
205
+func (bs *BitSwap) GetLedger(k u.Key) *Ledger {
206
+ l, ok := bs.partners[k]
207
+ if ok {
208
+ return l
209
+ }
210
+
211
+ l = new(Ledger)
212
+ l.Partner = peer.ID(k)
213
+ bs.partners[k] = l
214
+ return l
215
+}
bitswap/ledger.go
+16
-5
@@ -1,6 +1,8 @@
1
package bitswap
2
3
import (
4
+ "math/rand"
5
+
6
peer "github.com/jbenet/go-ipfs/peer"
7
u "github.com/jbenet/go-ipfs/util"
8
@@ -13,11 +15,8 @@ type Ledger struct {
15
// Partner is the ID of the remote Peer.
16
Partner peer.ID
17
16
- // BytesSent counts the number of bytes the local peer sent to Partner
17
- BytesSent uint64
18
-
19
- // BytesReceived counts the number of bytes local peer received from Partner
20
- BytesReceived uint64
18
+ // Accounting tracks bytes sent and recieved.
19
+ Accounting debtRatio
20
21
// FirstExchnage is the time of the first data exchange.
22
FirstExchange *time.Time
@@ -31,3 +30,15 @@ type Ledger struct {
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())
36
+}
37
+
38
+func (l *Ledger) SentBytes(n uint64) {
39
+ l.Accounting.BytesSent += n
40
+}
41
+
42
+func (l *Ledger) ReceivedBytes(n uint64) {
43
+ l.Accounting.BytesRecv += n
44
+}
bitswap/message.go
new
+30
@@ -0,0 +1,30 @@
1
+package bitswap
2
+
3
+import (
4
+ "code.google.com/p/goprotobuf/proto"
5
+ u "github.com/jbenet/go-ipfs/util"
6
+)
7
+
8
+type Message struct {
9
+ ID uint64
10
+ Response bool
11
+ Key u.Key
12
+ Value []byte
13
+ Success bool
14
+}
15
+
16
+func (m *Message) ToProtobuf() *PBMessage {
17
+ pmes := new(PBMessage)
18
+ pmes.Id = &m.ID
19
+ if m.Response {
20
+ pmes.Response = proto.Bool(true)
21
+ }
22
+
23
+ if m.Success {
24
+ pmes.Success = proto.Bool(true)
25
+ }
26
+
27
+ pmes.Key = proto.String(string(m.Key))
28
+ pmes.Value = m.Value
29
+ return pmes
30
+}
bitswap/message.pb.go
+8
@@ -56,6 +56,7 @@ type PBMessage struct {
56
Key *string `protobuf:"bytes,3,req,name=key" json:"key,omitempty"`
57
Value []byte `protobuf:"bytes,4,opt,name=value" json:"value,omitempty"`
58
Response *bool `protobuf:"varint,5,opt,name=response" json:"response,omitempty"`
59
+ Success *bool `protobuf:"varint,6,opt,name=success" json:"success,omitempty"`
60
XXX_unrecognized []byte `json:"-"`
61
}
62
@@ -98,6 +99,13 @@ func (m *PBMessage) GetResponse() bool {
99
return false
100
}
101
102
+func (m *PBMessage) GetSuccess() bool {
103
+ if m != nil && m.Success != nil {
104
+ return *m.Success
105
+ }
106
+ return false
107
+}
108
+
109
func init() {
110
proto.RegisterEnum("bitswap.PBMessage_MessageType", PBMessage_MessageType_name, PBMessage_MessageType_value)
111
}
bitswap/message.proto
+1
@@ -10,4 +10,5 @@ message PBMessage {
10
required string key = 3;
11
optional bytes value = 4;
12
optional bool response = 5;
13
+ optional bool success = 6;
14
}
blocks/blocks.go
-44
@@ -1,8 +1,6 @@
1
package blocks
2
3
import (
4
- "fmt"
5
- ds "github.com/jbenet/datastore.go"
4
u "github.com/jbenet/go-ipfs/util"
5
mh "github.com/jbenet/go-multihash"
6
)
@@ -27,45 +25,3 @@ func NewBlock(data []byte) (*Block, error) {
25
func (b *Block) Key() u.Key {
26
return u.Key(b.Multihash)
27
}
30
-
31
-// BlockService is a block datastore.
32
-// It uses an internal `datastore.Datastore` instance to store values.
33
-type BlockService struct {
34
- Datastore ds.Datastore
35
- // Remote *bitswap.BitSwap // eventually.
36
-}
37
-
38
-// NewBlockService creates a BlockService with given datastore instance.
39
-func NewBlockService(d ds.Datastore) (*BlockService, error) {
40
- if d == nil {
41
- return nil, fmt.Errorf("BlockService requires valid datastore")
42
- }
43
- return &BlockService{Datastore: d}, nil
44
-}
45
-
46
-// AddBlock adds a particular block to the service, Putting it into the datastore.
47
-func (s *BlockService) AddBlock(b *Block) (u.Key, error) {
48
- k := b.Key()
49
- dsk := ds.NewKey(string(k))
50
- return k, s.Datastore.Put(dsk, b.Data)
51
-}
52
-
53
-// GetBlock retrieves a particular block from the service,
54
-// Getting it from the datastore using the key (hash).
55
-func (s *BlockService) GetBlock(k u.Key) (*Block, error) {
56
- dsk := ds.NewKey(string(k))
57
- datai, err := s.Datastore.Get(dsk)
58
- if err != nil {
59
- return nil, err
60
- }
61
-
62
- data, ok := datai.([]byte)
63
- if !ok {
64
- return nil, fmt.Errorf("data associated with %s is not a []byte", k)
65
- }
66
-
67
- return &Block{
68
- Multihash: mh.Multihash(k),
69
- Data: data,
70
- }, nil
71
-}
blockservice/blockservice.go
new
+54
@@ -0,0 +1,54 @@
1
+package blockservice
2
+
3
+import (
4
+ "fmt"
5
+
6
+ ds "github.com/jbenet/datastore.go"
7
+ bitswap "github.com/jbenet/go-ipfs/bitswap"
8
+ blocks "github.com/jbenet/go-ipfs/blocks"
9
+ u "github.com/jbenet/go-ipfs/util"
10
+
11
+ mh "github.com/jbenet/go-multihash"
12
+)
13
+
14
+// BlockService is a block datastore.
15
+// It uses an internal `datastore.Datastore` instance to store values.
16
+type BlockService struct {
17
+ Datastore ds.Datastore
18
+ Remote *bitswap.BitSwap
19
+}
20
+
21
+// NewBlockService creates a BlockService with given datastore instance.
22
+func NewBlockService(d ds.Datastore) (*BlockService, error) {
23
+ if d == nil {
24
+ return nil, fmt.Errorf("BlockService requires valid datastore")
25
+ }
26
+ return &BlockService{Datastore: d}, nil
27
+}
28
+
29
+// AddBlock adds a particular block to the service, Putting it into the datastore.
30
+func (s *BlockService) AddBlock(b *blocks.Block) (u.Key, error) {
31
+ k := b.Key()
32
+ dsk := ds.NewKey(string(k))
33
+ return k, s.Datastore.Put(dsk, b.Data)
34
+}
35
+
36
+// GetBlock retrieves a particular block from the service,
37
+// Getting it from the datastore using the key (hash).
38
+func (s *BlockService) GetBlock(k u.Key) (*blocks.Block, error) {
39
+ dsk := ds.NewKey(string(k))
40
+ datai, err := s.Datastore.Get(dsk)
41
+ if err != nil {
42
+ return nil, err
43
+ }
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
54
+}
core/core.go
+4
-3
@@ -2,8 +2,9 @@ package core
2
3
import (
4
"fmt"
5
+
6
ds "github.com/jbenet/datastore.go"
6
- blocks "github.com/jbenet/go-ipfs/blocks"
7
+ bserv "github.com/jbenet/go-ipfs/blockservice"
8
config "github.com/jbenet/go-ipfs/config"
9
merkledag "github.com/jbenet/go-ipfs/merkledag"
10
path "github.com/jbenet/go-ipfs/path"
@@ -35,7 +36,7 @@ type IpfsNode struct {
36
// BitSwap *bitswap.BitSwap
37
38
// the block service, get/add blocks.
38
- Blocks *blocks.BlockService
39
+ Blocks *bserv.BlockService
40
41
// the merkle dag service, get/add objects.
42
DAG *merkledag.DAGService
@@ -58,7 +59,7 @@ func NewIpfsNode(cfg *config.Config) (*IpfsNode, error) {
59
return nil, err
60
}
61
61
- bs, err := blocks.NewBlockService(d)
62
+ bs, err := bserv.NewBlockService(d)
63
if err != nil {
64
return nil, err
65
}
merkledag/merkledag.go
+2
-1
@@ -4,6 +4,7 @@ import (
4
"fmt"
5
6
blocks "github.com/jbenet/go-ipfs/blocks"
7
+ bserv "github.com/jbenet/go-ipfs/blockservice"
8
u "github.com/jbenet/go-ipfs/util"
9
mh "github.com/jbenet/go-multihash"
10
)
@@ -93,7 +94,7 @@ func (n *Node) Key() (u.Key, error) {
94
// - the root is virtual (like a forest)
95
// - stores nodes' data in a BlockService
96
type DAGService struct {
96
- Blocks *blocks.BlockService
97
+ Blocks *bserv.BlockService
98
}
99
100
// Put adds a node to the DAGService, storing the block in the BlockService
routing/dht/dht.go
+1
@@ -603,6 +603,7 @@ func (dht *IpfsDHT) findProvidersSingle(p *peer.Peer, key u.Key, level int, time
603
}
604
}
605
606
+// TODO: Could be done async
607
func (dht *IpfsDHT) addPeerList(key u.Key, peers []*PBDHTMessage_PBPeer) []*peer.Peer {
608
var provArr []*peer.Peer
609
for _, prov := range peers {