fixing up some bitswap stuff after the PR
Jeromy committed
Aug 28, 2014 at 12:01 UTC
af2f04ae89df4995396c559eb71f312172a79c56
11 files changed
+87
-40
bitswap/bitswap.go
+37
-11
@@ -32,14 +32,13 @@ type BitSwap struct {
32
net swarm.Network
33
meschan *swarm.Chan
34
35
- // datastore is the local database
36
- // Ledgers of known
35
+ // datastore is the local database // Ledgers of known
36
datastore ds.Datastore
37
38
// routing interface for communication
39
routing *dht.IpfsDHT
40
42
- listener *swarm.MesListener
41
+ listener *swarm.MessageListener
42
43
// partners is a map of currently active bitswap relationships.
44
// The Ledger has the peer.ID, and the peer connection works through net.
@@ -67,7 +66,7 @@ func NewBitSwap(p *peer.Peer, net swarm.Network, d ds.Datastore, r routing.IpfsR
66
routing: r.(*dht.IpfsDHT),
67
meschan: net.GetChannel(swarm.PBWrapper_BITSWAP),
68
haltChan: make(chan struct{}),
70
- listener: swarm.NewMesListener(),
69
+ listener: swarm.NewMessageListener(),
70
}
71
72
go bs.handleMessages()
@@ -84,11 +83,11 @@ func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
83
valchan := make(chan []byte)
84
after := time.After(tleft)
85
87
- // TODO: when the data is received, shut down this for loop
86
+ // TODO: when the data is received, shut down this for loop ASAP
87
go func() {
88
for p := range provs_ch {
89
go func(pr *peer.Peer) {
91
- ledger := bs.GetLedger(pr.Key())
90
+ ledger := bs.GetLedger(pr)
91
blk, err := bs.getBlock(k, pr, tleft)
92
if err != nil {
93
u.PErr("getBlock returned: %v\n", err)
@@ -170,6 +169,8 @@ func (bs *BitSwap) handleMessages() {
169
switch pmes.GetType() {
170
case PBMessage_GET_BLOCK:
171
go bs.handleGetBlock(mes.Peer, pmes)
172
+ case PBMessage_WANT_BLOCK:
173
+ go bs.handleWantBlock(mes.Peer, pmes)
174
default:
175
u.PErr("Invalid message type.\n")
176
}
@@ -179,9 +180,18 @@ func (bs *BitSwap) handleMessages() {
180
}
181
}
182
183
+func (bs *BitSwap) handleWantBlock(p *peer.Peer, pmes *PBMessage) {
184
+ wants := pmes.GetWantlist()
185
+ ledg := bs.GetLedger(p)
186
+ for _, s := range wants {
187
+ // TODO: this needs to be different. We need timeouts.
188
+ ledg.WantList[u.Key(s)] = struct{}{}
189
+ }
190
+}
191
+
192
func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
193
u.DOut("handleGetBlock.\n")
184
- ledger := bs.GetLedger(p.Key())
194
+ ledger := bs.GetLedger(p)
195
196
u.DOut("finding [%s] in datastore.\n", u.Key(pmes.GetKey()).Pretty())
197
idata, err := bs.datastore.Get(ds.NewKey(pmes.GetKey()))
@@ -216,19 +226,35 @@ func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
226
}
227
}
228
219
-func (bs *BitSwap) GetLedger(k u.Key) *Ledger {
220
- l, ok := bs.partners[k]
229
+func (bs *BitSwap) GetLedger(p *peer.Peer) *Ledger {
230
+ l, ok := bs.partners[p.Key()]
231
if ok {
232
return l
233
}
234
235
l = new(Ledger)
236
l.Strategy = StandardStrategy
227
- l.Partner = peer.ID(k)
228
- bs.partners[k] = l
237
+ l.Partner = p
238
+ bs.partners[p.Key()] = l
239
return l
240
}
241
242
+func (bs *BitSwap) SendWantList(wl KeySet) error {
243
+ mes := Message{
244
+ ID: swarm.GenerateMessageID(),
245
+ Type: PBMessage_WANT_BLOCK,
246
+ WantList: bs.wantList,
247
+ }
248
+
249
+ pbmes := mes.ToProtobuf()
250
+ // Lets just ping everybody all at once
251
+ for _, ledger := range bs.partners {
252
+ bs.meschan.Outgoing <- swarm.NewMessage(ledger.Partner, pbmes)
253
+ }
254
+
255
+ return nil
256
+}
257
+
258
func (bs *BitSwap) Halt() {
259
bs.haltChan <- struct{}{}
260
}
bitswap/ledger.go
+6
-4
@@ -10,17 +10,17 @@ import (
10
// Ledger stores the data exchange relationship between two peers.
11
type Ledger struct {
12
13
- // Partner is the ID of the remote Peer.
14
- Partner peer.ID
13
+ // Partner is the remote Peer.
14
+ Partner *peer.Peer
15
16
// Accounting tracks bytes sent and recieved.
17
Accounting debtRatio
18
19
// FirstExchnage is the time of the first data exchange.
20
- FirstExchange *time.Time
20
+ FirstExchange time.Time
21
22
// LastExchange is the time of the last data exchange.
23
- LastExchange *time.Time
23
+ LastExchange time.Time
24
25
// WantList is a (bounded, small) set of keys that Partner desires.
26
WantList KeySet
@@ -36,9 +36,11 @@ func (l *Ledger) ShouldSend() bool {
36
}
37
38
func (l *Ledger) SentBytes(n uint64) {
39
+ l.LastExchange = time.Now()
40
l.Accounting.BytesSent += n
41
}
42
43
func (l *Ledger) ReceivedBytes(n uint64) {
44
+ l.LastExchange = time.Now()
45
l.Accounting.BytesRecv += n
46
}
bitswap/message.go
+9
@@ -12,6 +12,7 @@ type Message struct {
12
Key u.Key
13
Value []byte
14
Success bool
15
+ WantList KeySet
16
}
17
18
func (m *Message) ToProtobuf() *PBMessage {
@@ -26,6 +27,14 @@ func (m *Message) ToProtobuf() *PBMessage {
27
pmes.Success = proto.Bool(true)
28
}
29
30
+ if m.WantList != nil {
31
+ var swant []string
32
+ for k, _ := range m.WantList {
33
+ swant = append(swant, string(k))
34
+ }
35
+ pmes.Wantlist = swant
36
+ }
37
+
38
pmes.Key = proto.String(string(m.Key))
39
pmes.Value = m.Value
40
return pmes
bitswap/message.pb.go
+13
-2
@@ -23,14 +23,17 @@ var _ = math.Inf
23
type PBMessage_MessageType int32
24
25
const (
26
- PBMessage_GET_BLOCK PBMessage_MessageType = 0
26
+ PBMessage_GET_BLOCK PBMessage_MessageType = 0
27
+ PBMessage_WANT_BLOCK PBMessage_MessageType = 1
28
)
29
30
var PBMessage_MessageType_name = map[int32]string{
31
0: "GET_BLOCK",
32
+ 1: "WANT_BLOCK",
33
}
34
var PBMessage_MessageType_value = map[string]int32{
33
- "GET_BLOCK": 0,
35
+ "GET_BLOCK": 0,
36
+ "WANT_BLOCK": 1,
37
}
38
39
func (x PBMessage_MessageType) Enum() *PBMessage_MessageType {
@@ -57,6 +60,7 @@ type PBMessage struct {
60
Value []byte `protobuf:"bytes,4,opt,name=value" json:"value,omitempty"`
61
Response *bool `protobuf:"varint,5,opt,name=response" json:"response,omitempty"`
62
Success *bool `protobuf:"varint,6,opt,name=success" json:"success,omitempty"`
63
+ Wantlist []string `protobuf:"bytes,7,rep,name=wantlist" json:"wantlist,omitempty"`
64
XXX_unrecognized []byte `json:"-"`
65
}
66
@@ -106,6 +110,13 @@ func (m *PBMessage) GetSuccess() bool {
110
return false
111
}
112
113
+func (m *PBMessage) GetWantlist() []string {
114
+ if m != nil {
115
+ return m.Wantlist
116
+ }
117
+ return nil
118
+}
119
+
120
func init() {
121
proto.RegisterEnum("bitswap.PBMessage_MessageType", PBMessage_MessageType_name, PBMessage_MessageType_value)
122
}
bitswap/message.proto
+2
@@ -3,6 +3,7 @@ package bitswap;
3
message PBMessage {
4
enum MessageType {
5
GET_BLOCK = 0;
6
+ WANT_BLOCK = 1;
7
}
8
9
required MessageType Type = 1;
@@ -11,4 +12,5 @@ message PBMessage {
12
optional bytes value = 4;
13
optional bool response = 5;
14
optional bool success = 6;
15
+ repeated string wantlist = 7;
16
}
blockservice/blocks_test.go
+1
-6
@@ -2,7 +2,6 @@ package blockservice
2
3
import (
4
"bytes"
5
- "fmt"
5
"testing"
6
7
ds "github.com/jbenet/datastore.go"
@@ -11,9 +10,8 @@ import (
10
)
11
12
func TestBlocks(t *testing.T) {
14
-
13
d := ds.NewMapDatastore()
16
- bs, err := NewBlockService(d)
14
+ bs, err := NewBlockService(d, nil)
15
if err != nil {
16
t.Error("failed to construct block service", err)
17
return
@@ -62,7 +60,4 @@ func TestBlocks(t *testing.T) {
60
if !bytes.Equal(b.Data, b2.Data) {
61
t.Error("Block data is not equal.")
62
}
65
-
66
- fmt.Printf("key: %s\n", b.Key())
67
- fmt.Printf("data: %v\n", b.Data)
63
}
blockservice/blockservice.go
+5
-3
@@ -25,7 +25,7 @@ func NewBlockService(d ds.Datastore, rem *bitswap.BitSwap) (*BlockService, error
25
return nil, fmt.Errorf("BlockService requires valid datastore")
26
}
27
if rem == nil {
28
- return nil, fmt.Errorf("BlockService requires a valid bitswap")
28
+ u.PErr("Caution: blockservice running in local (offline) mode.\n")
29
}
30
return &BlockService{Datastore: d, Remote: rem}, nil
31
}
@@ -39,7 +39,9 @@ func (s *BlockService) AddBlock(b *blocks.Block) (u.Key, error) {
39
if err != nil {
40
return k, err
41
}
42
- err = s.Remote.HaveBlock(b.Key())
42
+ if s.Remote != nil {
43
+ err = s.Remote.HaveBlock(b.Key())
44
+ }
45
return k, err
46
}
47
@@ -57,7 +59,7 @@ func (s *BlockService) GetBlock(k u.Key) (*blocks.Block, error) {
59
Multihash: mh.Multihash(k),
60
Data: bdata,
61
}, nil
60
- } else if err == ds.ErrNotFound {
62
+ } else if err == ds.ErrNotFound && s.Remote != nil {
63
blk, err := s.Remote.GetBlock(k, time.Second*5)
64
if err != nil {
65
return nil, err
routing/dht/dht.go
+2
-2
@@ -49,7 +49,7 @@ type IpfsDHT struct {
49
diaglock sync.Mutex
50
51
// listener is a server to register to listen for responses to messages
52
- listener *swarm.MesListener
52
+ listener *swarm.MessageListener
53
}
54
55
// NewDHT creates a new DHT object with the given peer as the 'local' host
@@ -66,7 +66,7 @@ func NewDHT(p *peer.Peer, net swarm.Network, dstore ds.Datastore) *IpfsDHT {
66
dht.routingTables[0] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*30)
67
dht.routingTables[1] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Millisecond*100)
68
dht.routingTables[2] = kb.NewRoutingTable(20, kb.ConvertPeerID(p.ID), time.Hour)
69
- dht.listener = swarm.NewMesListener()
69
+ dht.listener = swarm.NewMessageListener()
70
dht.birth = time.Now()
71
return dht
72
}
routing/kbucket/table_test.go
+1
-1
@@ -78,7 +78,7 @@ func TestTableUpdate(t *testing.T) {
78
for i := 0; i < 10000; i++ {
79
p := rt.Update(peers[rand.Intn(len(peers))])
80
if p != nil {
81
- t.Log("evicted peer.")
81
+ //t.Log("evicted peer.")
82
}
83
}
84
swarm/mes_listener.go
+8
-8
@@ -8,7 +8,7 @@ import (
8
u "github.com/jbenet/go-ipfs/util"
9
)
10
11
-type MesListener struct {
11
+type MessageListener struct {
12
listeners map[uint64]*listenInfo
13
haltchan chan struct{}
14
unlist chan uint64
@@ -41,8 +41,8 @@ type listenInfo struct {
41
id uint64
42
}
43
44
-func NewMesListener() *MesListener {
45
- ml := new(MesListener)
44
+func NewMessageListener() *MessageListener {
45
+ ml := new(MessageListener)
46
ml.haltchan = make(chan struct{})
47
ml.listeners = make(map[uint64]*listenInfo)
48
ml.nlist = make(chan *listenInfo, 16)
@@ -52,7 +52,7 @@ func NewMesListener() *MesListener {
52
return ml
53
}
54
55
-func (ml *MesListener) Listen(id uint64, count int, timeout time.Duration) <-chan *Message {
55
+func (ml *MessageListener) Listen(id uint64, count int, timeout time.Duration) <-chan *Message {
56
li := new(listenInfo)
57
li.count = count
58
li.eol = time.Now().Add(timeout)
@@ -62,7 +62,7 @@ func (ml *MesListener) Listen(id uint64, count int, timeout time.Duration) <-cha
62
return li.resp
63
}
64
65
-func (ml *MesListener) Unlisten(id uint64) {
65
+func (ml *MessageListener) Unlisten(id uint64) {
66
ml.unlist <- id
67
}
68
@@ -71,18 +71,18 @@ type respMes struct {
71
mes *Message
72
}
73
74
-func (ml *MesListener) Respond(id uint64, mes *Message) {
74
+func (ml *MessageListener) Respond(id uint64, mes *Message) {
75
ml.send <- &respMes{
76
id: id,
77
mes: mes,
78
}
79
}
80
81
-func (ml *MesListener) Halt() {
81
+func (ml *MessageListener) Halt() {
82
ml.haltchan <- struct{}{}
83
}
84
85
-func (ml *MesListener) run() {
85
+func (ml *MessageListener) run() {
86
for {
87
select {
88
case <-ml.haltchan:
swarm/mes_listener_test.go
+3
-3
@@ -8,8 +8,8 @@ import (
8
)
9
10
// Ensure that the Message Listeners basic functionality works
11
-func TestMesListenerBasic(t *testing.T) {
12
- ml := NewMesListener()
11
+func TestMessageListener(t *testing.T) {
12
+ ml := NewMessageListener()
13
a := GenerateMessageID()
14
resp := ml.Listen(a, 1, time.Minute)
15
@@ -20,7 +20,7 @@ func TestMesListenerBasic(t *testing.T) {
20
21
go ml.Respond(a, mes)
22
23
- del := time.After(time.Millisecond * 10)
23
+ del := time.After(time.Millisecond * 100)
24
select {
25
case get := <-resp:
26
if string(get.Data) != string(mes.Data) {