@cryptotaxi247 / kubo / commits / d16591ad2

Make bitswap better

License: MIT Signed-off-by: Jeromy <why@ipfs.io>

Jeromy committed May 19, 2016 at 14:32 UTC d16591ad219af0bc0e5af317063cc9283c3dd8a4

11 files changed +161 -21

exchange/bitswap/decision/engine.go

+14 -7

@@ -3,6 +3,7 @@ package decision

4 import (

5 "sync"

6 + "time"

8 blocks "github.com/ipfs/go-ipfs/blocks"

9 bstore "github.com/ipfs/go-ipfs/blocks/blockstore"

@@ -68,7 +69,7 @@ type Engine struct {

69 // peerRequestQueue is a priority queue of requests received from peers.

70 // Requests are popped from the queue, packaged up, and placed in the

71 // outbox.

71 - peerRequestQueue peerRequestQueue

72 + peerRequestQueue *prq

74 // FIXME it's a bit odd for the client and the worker to both share memory

75 // (both modify the peerRequestQueue) and also to communicate over the

@@ -86,6 +87,8 @@ type Engine struct {

87 lock sync.Mutex // protects the fields immediatly below

88 // ledgerMap lists Ledgers by their Partner key.

89 ledgerMap map[peer.ID]*ledger

90 +

91 + ticker *time.Ticker

92 }

94 func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine {

@@ -95,6 +98,7 @@ func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine {

98 peerRequestQueue: newPRQ(),

99 outbox: make(chan (<-chan *Envelope), outboxChanBuffer),

100 workSignal: make(chan struct{}, 1),

101 + ticker: time.NewTicker(time.Millisecond * 100),

102 }

103 go e.taskWorker(ctx)

104 return e

@@ -142,6 +146,9 @@ func (e *Engine) nextEnvelope(ctx context.Context) (*Envelope, error) {

146 return nil, ctx.Err()

147 case <-e.workSignal:

148 nextTask = e.peerRequestQueue.Pop()

149 + case <-e.ticker.C:

150 + e.peerRequestQueue.thawRound()

151 + nextTask = e.peerRequestQueue.Pop()

152 }

153 }

154

@@ -191,9 +198,6 @@ func (e *Engine) Peers() []peer.ID {

198 // MessageReceived performs book-keeping. Returns error if passed invalid

199 // arguments.

200 func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {

194 - e.lock.Lock()

195 - defer e.lock.Unlock()

196 -

201 if len(m.Wantlist()) == 0 && len(m.Blocks()) == 0 {

202 log.Debugf("received empty message from %s", p)

203 }

@@ -206,6 +210,8 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {

210 }()

211

212 l := e.findOrCreate(p)

213 + l.lk.Lock()

214 + defer l.lk.Unlock()

215 if m.Full() {

216 l.wantList = wl.New()

217 }

@@ -236,10 +242,12 @@ func (e *Engine) addBlock(block blocks.Block) {

242 work := false

243

244 for _, l := range e.ledgerMap {

245 + l.lk.Lock()

246 if entry, ok := l.WantListContains(block.Key()); ok {

247 e.peerRequestQueue.Push(entry, l.Partner)

248 work = true

249 }

250 + l.lk.Unlock()

251 }

252

253 if work {

@@ -261,9 +269,6 @@ func (e *Engine) AddBlock(block blocks.Block) {

269 // send happen atomically

270

271 func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage) error {

264 - e.lock.Lock()

265 - defer e.lock.Unlock()

266 -

272 l := e.findOrCreate(p)

273 for _, block := range m.Blocks() {

274 l.SentBytes(len(block.Data()))

@@ -290,11 +295,13 @@ func (e *Engine) numBytesReceivedFrom(p peer.ID) uint64 {

295

296 // ledger lazily instantiates a ledger

297 func (e *Engine) findOrCreate(p peer.ID) *ledger {

298 + e.lock.Lock()

299 l, ok := e.ledgerMap[p]

300 if !ok {

301 l = newLedger(p)

302 e.ledgerMap[p] = l

303 }

304 + e.lock.Unlock()

305 return l

306 }

307

exchange/bitswap/decision/ledger.go

@@ -1,6 +1,7 @@

1 package decision

3 import (

4 + "sync"

5 "time"

7 key "github.com/ipfs/go-ipfs/blocks/key"

@@ -44,6 +45,8 @@ type ledger struct {

45 // sentToPeer is a set of keys to ensure we dont send duplicate blocks

46 // to a given peer

47 sentToPeer map[key.Key]time.Time

48 +

49 + lk sync.Mutex

50 }

52 type debtRatio struct {

exchange/bitswap/decision/peer_request_queue.go

+54 -3

@@ -15,14 +15,16 @@ type peerRequestQueue interface {

15 Pop() *peerRequestTask

16 Push(entry wantlist.Entry, to peer.ID)

17 Remove(k key.Key, p peer.ID)

18 +

19 // NB: cannot expose simply expose taskQueue.Len because trashed elements

20 // may exist. These trashed elements should not contribute to the count.

21 }

22 -func newPRQ() peerRequestQueue {

23 +func newPRQ() *prq {

24 return &prq{

25 taskMap: make(map[string]*peerRequestTask),

26 partners: make(map[peer.ID]*activePartner),

27 + frozen: make(map[peer.ID]*activePartner),

28 pQueue: pq.New(partnerCompare),

29 }

30 }

@@ -38,6 +40,8 @@ type prq struct {

40 pQueue pq.PQ

41 taskMap map[string]*peerRequestTask

42 partners map[peer.ID]*activePartner

43 +

44 + frozen map[peer.ID]*activePartner

45 }

47 // Push currently adds a new peerRequestTask to the end of the list

@@ -92,7 +96,7 @@ func (tl *prq) Pop() *peerRequestTask {

96 partner := tl.pQueue.Pop().(*activePartner)

98 var out *peerRequestTask

95 - for partner.taskQueue.Len() > 0 {

99 + for partner.taskQueue.Len() > 0 && partner.freezeVal == 0 {

100 out = partner.taskQueue.Pop().(*peerRequestTask)

101 delete(tl.taskMap, out.Key())

102 if out.trash {

@@ -120,11 +124,47 @@ func (tl *prq) Remove(k key.Key, p peer.ID) {

124 t.trash = true

125

126 // having canceled a block, we now account for that in the given partner

123 - tl.partners[p].requests--

127 + partner := tl.partners[p]

128 + partner.requests--

129 +

130 + // we now also 'freeze' that partner. If they sent us a cancel for a

131 + // block we were about to send them, we should wait a short period of time

132 + // to make sure we receive any other in-flight cancels before sending

133 + // them a block they already potentially have

134 + if partner.freezeVal == 0 {

135 + tl.frozen[p] = partner

136 + }

137 +

138 + partner.freezeVal++

139 + tl.pQueue.Update(partner.index)

140 }

141 tl.lock.Unlock()

142 }

143

144 +func (tl *prq) fullThaw() {

145 + tl.lock.Lock()

146 + defer tl.lock.Unlock()

147 +

148 + for id, partner := range tl.frozen {

149 + partner.freezeVal = 0

150 + delete(tl.frozen, id)

151 + tl.pQueue.Update(partner.index)

152 + }

153 +}

154 +

155 +func (tl *prq) thawRound() {

156 + tl.lock.Lock()

157 + defer tl.lock.Unlock()

158 +

159 + for id, partner := range tl.frozen {

160 + partner.freezeVal -= (partner.freezeVal + 1) / 2

161 + if partner.freezeVal <= 0 {

162 + delete(tl.frozen, id)

163 + }

164 + tl.pQueue.Update(partner.index)

165 + }

166 +}

167 +

168 type peerRequestTask struct {

169 Entry wantlist.Entry

170 Target peer.ID

@@ -196,6 +236,8 @@ type activePartner struct {

236 // for the PQ interface

237 index int

238

239 + freezeVal int

240 +

241 // priority queue of tasks belonging to this peer

242 taskQueue pq.PQ

243 }

@@ -208,6 +250,7 @@ func newActivePartner() *activePartner {

250 }

251

252 // partnerCompare implements pq.ElemComparator

253 +// returns true if peer 'a' has higher priority than peer 'b'

254 func partnerCompare(a, b pq.Elem) bool {

255 pa := a.(*activePartner)

256 pb := b.(*activePartner)

@@ -220,6 +263,14 @@ func partnerCompare(a, b pq.Elem) bool {

263 if pb.requests == 0 {

264 return true

265 }

266 +

267 + if pa.freezeVal > pb.freezeVal {

268 + return false

269 + }

270 + if pa.freezeVal < pb.freezeVal {

271 + return true

272 + }

273 +

274 if pa.active == pb.active {

275 // sorting by taskQueue.Len() aids in cleaning out trash entries faster

276 // if we sorted instead by requests, one peer could potentially build up

exchange/bitswap/decision/peer_request_queue_test.go

@@ -47,6 +47,8 @@ func TestPushPop(t *testing.T) {

47 prq.Remove(key.Key(consonant), partner)

48 }

50 + prq.fullThaw()

51 +

52 var out []string

53 for {

54 received := prq.Pop()

exchange/bitswap/network/interface.go

@@ -25,9 +25,16 @@ type BitSwapNetwork interface {

26 ConnectTo(context.Context, peer.ID) error

28 + NewMessageSender(context.Context, peer.ID) (MessageSender, error)

29 +

30 Routing

31 }

33 +type MessageSender interface {

34 + SendMsg(bsmsg.BitSwapMessage) error

35 + Close() error

36 +}

37 +

38 // Implement Receiver to receive messages from the BitSwapNetwork

39 type Receiver interface {

40 ReceiveMessage(

exchange/bitswap/network/ipfs_impl.go

+21

@@ -42,6 +42,27 @@ type impl struct {

42 receiver Receiver

43 }

45 +type streamMessageSender struct {

46 + s inet.Stream

47 +}

48 +

49 +func (s *streamMessageSender) Close() error {

50 + return s.s.Close()

51 +}

52 +

53 +func (s *streamMessageSender) SendMsg(msg bsmsg.BitSwapMessage) error {

54 + return msg.ToNet(s.s)

55 +}

56 +

57 +func (bsnet *impl) NewMessageSender(ctx context.Context, p peer.ID) (MessageSender, error) {

58 + s, err := bsnet.newStreamToPeer(ctx, p)

59 + if err != nil {

60 + return nil, err

61 + }

62 +

63 + return &streamMessageSender{s: s}, nil

64 +}

65 +

66 func (bsnet *impl) newStreamToPeer(ctx context.Context, p peer.ID) (inet.Stream, error) {

68 // first, make sure we're connected.

exchange/bitswap/testnet/virtual.go

+24

@@ -112,6 +112,30 @@ func (nc *networkClient) FindProvidersAsync(ctx context.Context, k key.Key, max

112 return out

113 }

114

115 +type messagePasser struct {

116 + net *network

117 + target peer.ID

118 + local peer.ID

119 + ctx context.Context

120 +}

121 +

122 +func (mp *messagePasser) SendMsg(m bsmsg.BitSwapMessage) error {

123 + return mp.net.SendMessage(mp.ctx, mp.local, mp.target, m)

124 +}

125 +

126 +func (mp *messagePasser) Close() error {

127 + return nil

128 +}

129 +

130 +func (n *networkClient) NewMessageSender(ctx context.Context, p peer.ID) (bsnet.MessageSender, error) {

131 + return &messagePasser{

132 + net: n.network,

133 + target: p,

134 + local: n.local,

135 + ctx: ctx,

136 + }, nil

137 +}

138 +

139 // Provide provides the key to the network

140 func (nc *networkClient) Provide(ctx context.Context, k key.Key) error {

141 return nc.routing.Provide(ctx, k)

exchange/bitswap/wantmanager.go

+31 -11

@@ -26,9 +26,11 @@ type WantManager struct {

27 network bsnet.BitSwapNetwork

28 ctx context.Context

29 + cancel func()

30 }

32 func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantManager {

33 + ctx, cancel := context.WithCancel(ctx)

34 return &WantManager{

35 incoming: make(chan []*bsmsg.Entry, 10),

36 connect: make(chan peer.ID, 10),

@@ -38,6 +40,7 @@ func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantMana

40 wl: wantlist.NewThreadSafe(),

41 network: network,

42 ctx: ctx,

43 + cancel: cancel,

44 }

45 }

@@ -58,6 +61,8 @@ type msgQueue struct {

61 out bsmsg.BitSwapMessage

62 network bsnet.BitSwapNetwork

64 + sender bsnet.MessageSender

65 +

66 refcnt int

68 work chan struct{}

@@ -150,6 +155,11 @@ func (pm *WantManager) stopPeerHandler(p peer.ID) {

155 }

156

157 func (mq *msgQueue) runQueue(ctx context.Context) {

158 + defer func() {

159 + if mq.sender != nil {

160 + mq.sender.Close()

161 + }

162 + }()

163 for {

164 select {

165 case <-mq.work: // there is work to be done

@@ -166,14 +176,25 @@ func (mq *msgQueue) doWork(ctx context.Context) {

176 // allow ten minutes for connections

177 // this includes looking them up in the dht

178 // dialing them, and handshaking

169 - conctx, cancel := context.WithTimeout(ctx, time.Minute*10)

170 - defer cancel()

179 + if mq.sender == nil {

180 + conctx, cancel := context.WithTimeout(ctx, time.Minute*10)

181 + defer cancel()

182 +

183 + err := mq.network.ConnectTo(conctx, mq.p)

184 + if err != nil {

185 + log.Infof("cant connect to peer %s: %s", mq.p, err)

186 + // TODO: cant connect, what now?

187 + return

188 + }

189

172 - err := mq.network.ConnectTo(conctx, mq.p)

173 - if err != nil {

174 - log.Infof("cant connect to peer %s: %s", mq.p, err)

175 - // TODO: cant connect, what now?

176 - return

190 + nsender, err := mq.network.NewMessageSender(ctx, mq.p)

191 + if err != nil {

192 + log.Infof("cant open new stream to peer %s: %s", mq.p, err)

193 + // TODO: cant open stream, what now?

194 + return

195 + }

196 +

197 + mq.sender = nsender

198 }

199

200 // grab outgoing message

@@ -186,13 +207,12 @@ func (mq *msgQueue) doWork(ctx context.Context) {

207 mq.out = nil

208 mq.outlk.Unlock()

209

189 - sendctx, cancel := context.WithTimeout(ctx, time.Minute*5)

190 - defer cancel()

191 -

210 // send wantlist updates

193 - err = mq.network.SendMessage(sendctx, mq.p, wlm)

211 + err := mq.sender.SendMsg(wlm)

212 if err != nil {

213 log.Infof("bitswap send error: %s", err)

214 + mq.sender.Close()

215 + mq.sender = nil

216 // TODO: what do we do if this fails?

217 return

218 }

test/integration/addcat_test.go

@@ -149,6 +149,8 @@ func DirectAddCat(data []byte, conf testutil.LatencyConfig) error {

149 if 0 != bytes.Compare(bufout.Bytes(), data) {

150 return errors.New("catted data does not match added data")

151 }

152 +

153 + cancel()

154 return nil

155 }

156

test/integration/grandcentral_test.go

@@ -73,6 +73,7 @@ func RunSupernodeBootstrappedAddCat(data []byte, conf testutil.LatencyConfig) er

73 if 0 != bytes.Compare(bufout.Bytes(), data) {

74 return errors.New("catted data does not match added data")

75 }

76 + cancel()

77 return nil

78 }

@@ -177,5 +178,6 @@ func RunSupernodePutRecordGetRecord(conf testutil.LatencyConfig) error {

178 if 0 != bytes.Compare(note, received) {

179 return errors.New("record doesn't match")

180 }

181 + cancel()

182 return nil

183 }

test/integration/three_legged_cat_test.go

@@ -128,5 +128,6 @@ func RunThreeLeggedCat(data []byte, conf testutil.LatencyConfig) error {

128 if 0 != bytes.Compare(bufout.Bytes(), data) {

129 return errors.New("catted data does not match added data")

130 }

131 + cancel()

132 return nil

133 }