@cryptotaxi247 / kubo / commits / fcff5a5c9

rework bitswap to reflect discussion on PR #32

Jeromy committed Aug 28, 2014 at 16:48 UTC fcff5a5c96541f0c26621c087cbf8940da01a4e5
11 files changed +141 -243
bitswap/bitswap.go
+89 -70
@@ -52,6 +52,8 @@ type BitSwap struct {
52 // wantList is the set of keys we want values for. a map for fast lookups.
53 wantList KeySet
54
55 + strategy StrategyFunc
56 +
57 haltChan chan struct{}
58 }
59
@@ -87,15 +89,11 @@ func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
89 go func() {
90 for p := range provs_ch {
91 go func(pr *peer.Peer) {
90 - ledger := bs.GetLedger(pr)
92 blk, err := bs.getBlock(k, pr, tleft)
93 if err != nil {
94 u.PErr("getBlock returned: %v\n", err)
95 return
96 }
96 - // NOTE: this credits everyone who sends us a block,
97 - // even if we dont use it
98 - ledger.ReceivedBytes(uint64(len(blk)))
97 select {
98 case valchan <- blk:
99 default:
@@ -115,30 +113,18 @@ func (bs *BitSwap) GetBlock(k u.Key, timeout time.Duration) (
113
114 func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byte, error) {
115 u.DOut("[%s] getBlock '%s' from [%s]\n", bs.peer.ID.Pretty(), k.Pretty(), p.ID.Pretty())
118 - //
119 - mes := new(PBMessage)
120 - mes.Id = proto.Uint64(swarm.GenerateMessageID())
121 - mes.Key = proto.String(string(k))
122 - typ := PBMessage_GET_BLOCK
123 - mes.Type = &typ
124 - //
116 +
117 + pmes := new(PBMessage)
118 + pmes.Wantlist = []string{string(k)}
119
120 after := time.After(timeout)
127 - resp := bs.listener.Listen(mes.GetId(), 1, timeout)
128 - smes := swarm.NewMessage(p, mes)
121 + resp := bs.listener.Listen(string(k), 1, timeout)
122 + smes := swarm.NewMessage(p, pmes)
123 bs.meschan.Outgoing <- smes
124
125 select {
126 case resp_mes := <-resp:
133 - pmes := new(PBMessage)
134 - err := proto.Unmarshal(resp_mes.Data, pmes)
135 - if err != nil {
136 - return nil, err
137 - }
138 - if pmes.GetSuccess() {
139 - return pmes.GetValue(), nil
140 - }
141 - return nil, u.ErrNotFound
127 + return resp_mes.Data, nil
128 case <-after:
129 u.PErr("getBlock for '%s' timed out.\n", k)
130 return nil, u.ErrTimeout
@@ -147,8 +133,26 @@ func (bs *BitSwap) getBlock(k u.Key, p *peer.Peer, timeout time.Duration) ([]byt
133
134 // HaveBlock announces the existance of a block to BitSwap, potentially sending
135 // it to peers (Partners) whose WantLists include it.
150 -func (bs *BitSwap) HaveBlock(k u.Key) error {
151 - return bs.routing.Provide(k)
136 +func (bs *BitSwap) HaveBlock(blk *blocks.Block) error {
137 + go func() {
138 + for _, ledger := range bs.partners {
139 + if _, ok := ledger.WantList[blk.Key()]; ok {
140 + //send block to node
141 + if ledger.ShouldSend() {
142 + bs.SendBlock(ledger.Partner, blk)
143 + }
144 + }
145 + }
146 + }()
147 + return bs.routing.Provide(blk.Key())
148 +}
149 +
150 +func (bs *BitSwap) SendBlock(p *peer.Peer, b *blocks.Block) {
151 + pmes := new(PBMessage)
152 + pmes.Blocks = [][]byte{b.Data}
153 +
154 + swarm_mes := swarm.NewMessage(p, pmes)
155 + bs.meschan.Outgoing <- swarm_mes
156 }
157
158 func (bs *BitSwap) handleMessages() {
@@ -161,18 +165,21 @@ func (bs *BitSwap) handleMessages() {
165 u.PErr("%v\n", err)
166 continue
167 }
164 - if pmes.GetResponse() {
165 - bs.listener.Respond(pmes.GetId(), mes)
166 - continue
168 + if pmes.Blocks != nil {
169 + for _, blkData := range pmes.Blocks {
170 + blk, err := blocks.NewBlock(blkData)
171 + if err != nil {
172 + u.PErr("%v\n", err)
173 + continue
174 + }
175 + go bs.blockReceive(mes.Peer, blk)
176 + }
177 }
178
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")
179 + if pmes.Wantlist != nil {
180 + for _, want := range pmes.Wantlist {
181 + go bs.peerWantsBlock(mes.Peer, want)
182 + }
183 }
184 case <-bs.haltChan:
185 return
@@ -180,52 +187,59 @@ func (bs *BitSwap) handleMessages() {
187 }
188 }
189
183 -func (bs *BitSwap) handleWantBlock(p *peer.Peer, pmes *PBMessage) {
184 - wants := pmes.GetWantlist()
190 +// peerWantsBlock will check if we have the block in question,
191 +// and then if we do, check the ledger for whether or not we should send it.
192 +func (bs *BitSwap) peerWantsBlock(p *peer.Peer, want string) {
193 + u.DOut("peer [%s] wants block [%s]\n", p.ID.Pretty(), u.Key(want).Pretty())
194 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 -}
195
192 -func (bs *BitSwap) handleGetBlock(p *peer.Peer, pmes *PBMessage) {
193 - u.DOut("handleGetBlock.\n")
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()))
196 + dsk := ds.NewKey(want)
197 + blk_i, err := bs.datastore.Get(dsk)
198 if err != nil {
199 - u.PErr("handleGetBlock datastore returned: %v\n", err)
199 if err == ds.ErrNotFound {
201 - return
200 + // TODO: this needs to be different. We need timeouts.
201 + ledg.WantList[u.Key(want)] = struct{}{}
202 }
203 + u.PErr("datastore get error: %v\n", err)
204 return
205 }
206
206 - u.DOut("found value!\n")
207 - data, ok := idata.([]byte)
207 + blk, ok := blk_i.([]byte)
208 if !ok {
209 - u.PErr("Failed casting data from datastore.")
209 + u.PErr("data conversion error.\n")
210 return
211 }
212
213 - if ledger.ShouldSend() {
214 - u.DOut("Sending value back!\n")
215 - resp := &Message{
216 - Value: data,
217 - Response: true,
218 - ID: pmes.GetId(),
219 - Type: PBMessage_GET_BLOCK,
220 - Success: true,
213 + if ledg.ShouldSend() {
214 + u.DOut("Sending block to peer.\n")
215 + bblk, err := blocks.NewBlock(blk)
216 + if err != nil {
217 + u.PErr("newBlock error: %v\n", err)
218 + return
219 }
222 - bs.meschan.Outgoing <- swarm.NewMessage(p, resp.ToProtobuf())
223 - ledger.SentBytes(uint64(len(data)))
224 - } else {
225 - u.DOut("Ledger decided not to send anything...\n")
220 + bs.SendBlock(p, bblk)
221 + ledg.SentBytes(len(blk))
222 }
223 }
224
225 +func (bs *BitSwap) blockReceive(p *peer.Peer, blk *blocks.Block) {
226 + u.DOut("blockReceive: %s\n", blk.Key().Pretty())
227 + err := bs.datastore.Put(ds.NewKey(string(blk.Key())), blk.Data)
228 + if err != nil {
229 + u.PErr("blockReceive error: %v\n", err)
230 + return
231 + }
232 +
233 + mes := &swarm.Message{
234 + Peer: p,
235 + Data: blk.Data,
236 + }
237 + bs.listener.Respond(string(blk.Key()), mes)
238 +
239 + ledger := bs.GetLedger(p)
240 + ledger.ReceivedBytes(len(blk.Data))
241 +}
242 +
243 func (bs *BitSwap) GetLedger(p *peer.Peer) *Ledger {
244 l, ok := bs.partners[p.Key()]
245 if ok {
@@ -240,16 +254,14 @@ func (bs *BitSwap) GetLedger(p *peer.Peer) *Ledger {
254 }
255
256 func (bs *BitSwap) SendWantList(wl KeySet) error {
243 - mes := Message{
244 - ID: swarm.GenerateMessageID(),
245 - Type: PBMessage_WANT_BLOCK,
246 - WantList: bs.wantList,
257 + pmes := new(PBMessage)
258 + for k, _ := range wl {
259 + pmes.Wantlist = append(pmes.Wantlist, string(k))
260 }
261
249 - pbmes := mes.ToProtobuf()
262 // Lets just ping everybody all at once
263 for _, ledger := range bs.partners {
252 - bs.meschan.Outgoing <- swarm.NewMessage(ledger.Partner, pbmes)
264 + bs.meschan.Outgoing <- swarm.NewMessage(ledger.Partner, pmes)
265 }
266
267 return nil
@@ -258,3 +270,10 @@ func (bs *BitSwap) SendWantList(wl KeySet) error {
270 func (bs *BitSwap) Halt() {
271 bs.haltChan <- struct{}{}
272 }
273 +
274 +func (bs *BitSwap) SetStrategy(sf StrategyFunc) {
275 + bs.strategy = sf
276 + for _, ledg := range bs.partners {
277 + ledg.Strategy = sf
278 + }
279 +}
bitswap/ledger.go
+10 -5
@@ -22,6 +22,9 @@ type Ledger struct {
22 // LastExchange is the time of the last data exchange.
23 LastExchange time.Time
24
25 + // Number of exchanges with this peer
26 + ExchangeCount uint64
27 +
28 // WantList is a (bounded, small) set of keys that Partner desires.
29 WantList KeySet
30
@@ -32,15 +35,17 @@ type Ledger struct {
35 type LedgerMap map[u.Key]*Ledger
36
37 func (l *Ledger) ShouldSend() bool {
35 - return l.Strategy(l.Accounting)
38 + return l.Strategy(l)
39 }
40
38 -func (l *Ledger) SentBytes(n uint64) {
41 +func (l *Ledger) SentBytes(n int) {
42 + l.ExchangeCount++
43 l.LastExchange = time.Now()
40 - l.Accounting.BytesSent += n
44 + l.Accounting.BytesSent += uint64(n)
45 }
46
43 -func (l *Ledger) ReceivedBytes(n uint64) {
47 +func (l *Ledger) ReceivedBytes(n int) {
48 + l.ExchangeCount++
49 l.LastExchange = time.Now()
45 - l.Accounting.BytesRecv += n
50 + l.Accounting.BytesRecv += uint64(n)
51 }
bitswap/message.go deleted
-41
@@ -1,41 +0,0 @@
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 - Type PBMessage_MessageType
10 - ID uint64
11 - Response bool
12 - Key u.Key
13 - Value []byte
14 - Success bool
15 - WantList KeySet
16 -}
17 -
18 -func (m *Message) ToProtobuf() *PBMessage {
19 - pmes := new(PBMessage)
20 - pmes.Id = &m.ID
21 - pmes.Type = &m.Type
22 - if m.Response {
23 - pmes.Response = proto.Bool(true)
24 - }
25 -
26 - if m.Success {
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
41 -}
bitswap/message.pb.go
+7 -81
@@ -20,103 +20,29 @@ import math "math"
20 var _ = proto.Marshal
21 var _ = math.Inf
22
23 -type PBMessage_MessageType int32
24 -
25 -const (
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{
35 - "GET_BLOCK": 0,
36 - "WANT_BLOCK": 1,
37 -}
38 -
39 -func (x PBMessage_MessageType) Enum() *PBMessage_MessageType {
40 - p := new(PBMessage_MessageType)
41 - *p = x
42 - return p
43 -}
44 -func (x PBMessage_MessageType) String() string {
45 - return proto.EnumName(PBMessage_MessageType_name, int32(x))
46 -}
47 -func (x *PBMessage_MessageType) UnmarshalJSON(data []byte) error {
48 - value, err := proto.UnmarshalJSONEnum(PBMessage_MessageType_value, data, "PBMessage_MessageType")
49 - if err != nil {
50 - return err
51 - }
52 - *x = PBMessage_MessageType(value)
53 - return nil
54 -}
55 -
23 type PBMessage struct {
57 - Type *PBMessage_MessageType `protobuf:"varint,1,req,enum=bitswap.PBMessage_MessageType" json:"Type,omitempty"`
58 - Id *uint64 `protobuf:"varint,2,req,name=id" json:"id,omitempty"`
59 - Key *string `protobuf:"bytes,3,req,name=key" json:"key,omitempty"`
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:"-"`
24 + Wantlist []string `protobuf:"bytes,1,rep,name=wantlist" json:"wantlist,omitempty"`
25 + Blocks [][]byte `protobuf:"bytes,2,rep,name=blocks" json:"blocks,omitempty"`
26 + XXX_unrecognized []byte `json:"-"`
27 }
28
29 func (m *PBMessage) Reset() { *m = PBMessage{} }
30 func (m *PBMessage) String() string { return proto.CompactTextString(m) }
31 func (*PBMessage) ProtoMessage() {}
32
71 -func (m *PBMessage) GetType() PBMessage_MessageType {
72 - if m != nil && m.Type != nil {
73 - return *m.Type
74 - }
75 - return PBMessage_GET_BLOCK
76 -}
77 -
78 -func (m *PBMessage) GetId() uint64 {
79 - if m != nil && m.Id != nil {
80 - return *m.Id
81 - }
82 - return 0
83 -}
84 -
85 -func (m *PBMessage) GetKey() string {
86 - if m != nil && m.Key != nil {
87 - return *m.Key
88 - }
89 - return ""
90 -}
91 -
92 -func (m *PBMessage) GetValue() []byte {
33 +func (m *PBMessage) GetWantlist() []string {
34 if m != nil {
94 - return m.Value
35 + return m.Wantlist
36 }
37 return nil
38 }
39
99 -func (m *PBMessage) GetResponse() bool {
100 - if m != nil && m.Response != nil {
101 - return *m.Response
102 - }
103 - return false
104 -}
105 -
106 -func (m *PBMessage) GetSuccess() bool {
107 - if m != nil && m.Success != nil {
108 - return *m.Success
109 - }
110 - return false
111 -}
112 -
113 -func (m *PBMessage) GetWantlist() []string {
40 +func (m *PBMessage) GetBlocks() [][]byte {
41 if m != nil {
115 - return m.Wantlist
42 + return m.Blocks
43 }
44 return nil
45 }
46
47 func init() {
121 - proto.RegisterEnum("bitswap.PBMessage_MessageType", PBMessage_MessageType_name, PBMessage_MessageType_value)
48 }
bitswap/message.proto
+2 -12
@@ -1,16 +1,6 @@
1 package bitswap;
2
3 message PBMessage {
4 - enum MessageType {
5 - GET_BLOCK = 0;
6 - WANT_BLOCK = 1;
7 - }
8 -
9 - required MessageType Type = 1;
10 - required uint64 id = 2;
11 - required string key = 3;
12 - optional bytes value = 4;
13 - optional bool response = 5;
14 - optional bool success = 6;
15 - repeated string wantlist = 7;
4 + repeated string wantlist = 1;
5 + repeated bytes blocks = 2;
6 }
bitswap/strategy.go
+4 -4
@@ -5,13 +5,13 @@ import (
5 "math/rand"
6 )
7
8 -type StrategyFunc func(debtRatio) bool
8 +type StrategyFunc func(*Ledger) bool
9
10 -func StandardStrategy(db debtRatio) bool {
11 - return rand.Float64() <= probabilitySend(db.Value())
10 +func StandardStrategy(l *Ledger) bool {
11 + return rand.Float64() <= probabilitySend(l.Accounting.Value())
12 }
13
14 -func YesManStrategy(db debtRatio) bool {
14 +func YesManStrategy(l *Ledger) bool {
15 return true
16 }
17
blockservice/blockservice.go
+1 -1
@@ -40,7 +40,7 @@ func (s *BlockService) AddBlock(b *blocks.Block) (u.Key, error) {
40 return k, err
41 }
42 if s.Remote != nil {
43 - err = s.Remote.HaveBlock(b.Key())
43 + err = s.Remote.HaveBlock(b)
44 }
45 return k, err
46 }
routing/dht/Message.go
+1 -1
@@ -11,7 +11,7 @@ type Message struct {
11 Key string
12 Value []byte
13 Response bool
14 - ID uint64
14 + ID string
15 Success bool
16 Peers []*peer.Peer
17 }
routing/dht/messages.pb.go
+12 -15
@@ -1,4 +1,4 @@
1 -// Code generated by protoc-gen-gogo.
1 +// Code generated by protoc-gen-go.
2 // source: messages.proto
3 // DO NOT EDIT!
4
@@ -13,7 +13,7 @@ It has these top-level messages:
13 */
14 package dht
15
16 -import proto "code.google.com/p/gogoprotobuf/proto"
16 +import proto "code.google.com/p/goprotobuf/proto"
17 import math "math"
18
19 // Reference imports to suppress errors if they are not otherwise used.
@@ -69,17 +69,14 @@ func (x *PBDHTMessage_MessageType) UnmarshalJSON(data []byte) error {
69 }
70
71 type PBDHTMessage struct {
72 - Type *PBDHTMessage_MessageType `protobuf:"varint,1,req,name=type,enum=dht.PBDHTMessage_MessageType" json:"type,omitempty"`
73 - Key *string `protobuf:"bytes,2,opt,name=key" json:"key,omitempty"`
74 - Value []byte `protobuf:"bytes,3,opt,name=value" json:"value,omitempty"`
75 - // Unique ID of this message, used to match queries with responses
76 - Id *uint64 `protobuf:"varint,4,req,name=id" json:"id,omitempty"`
77 - // Signals whether or not this message is a response to another message
78 - Response *bool `protobuf:"varint,5,opt,name=response" json:"response,omitempty"`
79 - Success *bool `protobuf:"varint,6,opt,name=success" json:"success,omitempty"`
80 - // Used for returning peers from queries (normally, peers closer to X)
81 - Peers []*PBDHTMessage_PBPeer `protobuf:"bytes,7,rep,name=peers" json:"peers,omitempty"`
82 - XXX_unrecognized []byte `json:"-"`
72 + Type *PBDHTMessage_MessageType `protobuf:"varint,1,req,name=type,enum=dht.PBDHTMessage_MessageType" json:"type,omitempty"`
73 + Key *string `protobuf:"bytes,2,opt,name=key" json:"key,omitempty"`
74 + Value []byte `protobuf:"bytes,3,opt,name=value" json:"value,omitempty"`
75 + Id *string `protobuf:"bytes,4,req,name=id" json:"id,omitempty"`
76 + Response *bool `protobuf:"varint,5,opt,name=response" json:"response,omitempty"`
77 + Success *bool `protobuf:"varint,6,opt,name=success" json:"success,omitempty"`
78 + Peers []*PBDHTMessage_PBPeer `protobuf:"bytes,7,rep,name=peers" json:"peers,omitempty"`
79 + XXX_unrecognized []byte `json:"-"`
80 }
81
82 func (m *PBDHTMessage) Reset() { *m = PBDHTMessage{} }
@@ -107,11 +104,11 @@ func (m *PBDHTMessage) GetValue() []byte {
104 return nil
105 }
106
110 -func (m *PBDHTMessage) GetId() uint64 {
107 +func (m *PBDHTMessage) GetId() string {
108 if m != nil && m.Id != nil {
109 return *m.Id
110 }
114 - return 0
111 + return ""
112 }
113
114 func (m *PBDHTMessage) GetResponse() bool {
routing/dht/messages.proto
+1 -1
@@ -23,7 +23,7 @@ message PBDHTMessage {
23 optional bytes value = 3;
24
25 // Unique ID of this message, used to match queries with responses
26 - required uint64 id = 4;
26 + required string id = 4;
27
28 // Signals whether or not this message is a response to another message
29 optional bool response = 5;
swarm/mes_listener.go
+14 -12
@@ -1,7 +1,7 @@
1 package swarm
2
3 import (
4 - "math/rand"
4 + crand "crypto/rand"
5 "sync"
6 "time"
7
@@ -9,16 +9,18 @@ import (
9 )
10
11 type MessageListener struct {
12 - listeners map[uint64]*listenInfo
12 + listeners map[string]*listenInfo
13 haltchan chan struct{}
14 - unlist chan uint64
14 + unlist chan string
15 nlist chan *listenInfo
16 send chan *respMes
17 }
18
19 // GenerateMessageID creates and returns a new message ID
20 -func GenerateMessageID() uint64 {
21 - return (uint64(rand.Uint32()) << 32) | uint64(rand.Uint32())
20 +func GenerateMessageID() string {
21 + buf := make([]byte, 16)
22 + crand.Read(buf)
23 + return string(buf)
24 }
25
26 // The listen info struct holds information about a message that is being waited for
@@ -38,21 +40,21 @@ type listenInfo struct {
40
41 closed bool
42
41 - id uint64
43 + id string
44 }
45
46 func NewMessageListener() *MessageListener {
47 ml := new(MessageListener)
48 ml.haltchan = make(chan struct{})
47 - ml.listeners = make(map[uint64]*listenInfo)
49 + ml.listeners = make(map[string]*listenInfo)
50 ml.nlist = make(chan *listenInfo, 16)
51 ml.send = make(chan *respMes, 16)
50 - ml.unlist = make(chan uint64, 16)
52 + ml.unlist = make(chan string, 16)
53 go ml.run()
54 return ml
55 }
56
55 -func (ml *MessageListener) Listen(id uint64, count int, timeout time.Duration) <-chan *Message {
57 +func (ml *MessageListener) Listen(id string, count int, timeout time.Duration) <-chan *Message {
58 li := new(listenInfo)
59 li.count = count
60 li.eol = time.Now().Add(timeout)
@@ -62,16 +64,16 @@ func (ml *MessageListener) Listen(id uint64, count int, timeout time.Duration) <
64 return li.resp
65 }
66
65 -func (ml *MessageListener) Unlisten(id uint64) {
67 +func (ml *MessageListener) Unlisten(id string) {
68 ml.unlist <- id
69 }
70
71 type respMes struct {
70 - id uint64
72 + id string
73 mes *Message
74 }
75
74 -func (ml *MessageListener) Respond(id uint64, mes *Message) {
76 +func (ml *MessageListener) Respond(id string, mes *Message) {
77 ml.send <- &respMes{
78 id: id,
79 mes: mes,