Merge pull request #4458 from ipfs/fix/bitswap-nits

bitswap nits

exchange/bitswap/bitswap.go

@@ -372,16 +372,6 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
 		return
 	}
 
-	// quickly send out cancels, reduces chances of duplicate block receives
-	var keys []*cid.Cid
-	for _, block := range iblocks {
-		if _, found := bs.wm.wl.Contains(block.Cid()); !found {
-			log.Infof("received un-asked-for %s from %s", block, p)
-			continue
-		}
-		keys = append(keys, block.Cid())
-	}
-
 	wg := sync.WaitGroup{}
 	for _, block := range iblocks {
 		wg.Add(1)
@@ -451,8 +441,9 @@ func (bs *Bitswap) Close() error {
 }
 
 func (bs *Bitswap) GetWantlist() []*cid.Cid {
-	var out []*cid.Cid
-	for _, e := range bs.wm.wl.Entries() {
+	entries := bs.wm.wl.Entries()
+	out := make([]*cid.Cid, 0, len(entries))
+	for _, e := range entries {
 		out = append(out, e.Cid)
 	}
 	return out

exchange/bitswap/decision/engine.go

@@ -201,7 +201,8 @@ func (e *Engine) Peers() []peer.ID {
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
-	response := make([]peer.ID, 0)
+	response := make([]peer.ID, 0, len(e.ledgerMap))
+
 	for _, ledger := range e.ledgerMap {
 		response = append(response, ledger.Partner)
 	}
@@ -297,15 +298,15 @@ func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage) error {
 
 func (e *Engine) PeerConnected(p peer.ID) {
 	e.lock.Lock()
+	defer e.lock.Unlock()
 	l, ok := e.ledgerMap[p]
 	if !ok {
 		l = newLedger(p)
 		e.ledgerMap[p] = l
 	}
 	l.lk.Lock()
+	defer l.lk.Unlock()
 	l.ref++
-	l.lk.Unlock()
-	e.lock.Unlock()
 }
 
 func (e *Engine) PeerDisconnected(p peer.ID) {
@@ -316,11 +317,11 @@ func (e *Engine) PeerDisconnected(p peer.ID) {
 		return
 	}
 	l.lk.Lock()
+	defer l.lk.Unlock()
 	l.ref--
 	if l.ref <= 0 {
 		delete(e.ledgerMap, p)
 	}
-	l.lk.Unlock()
 }
 
 func (e *Engine) numBytesSentTo(p peer.ID) uint64 {
@@ -336,12 +337,12 @@ func (e *Engine) numBytesReceivedFrom(p peer.ID) uint64 {
 // ledger lazily instantiates a ledger
 func (e *Engine) findOrCreate(p peer.ID) *ledger {
 	e.lock.Lock()
+	defer e.lock.Unlock()
 	l, ok := e.ledgerMap[p]
 	if !ok {
 		l = newLedger(p)
 		e.ledgerMap[p] = l
 	}
-	e.lock.Unlock()
 	return l
 }
 

exchange/bitswap/message/message.go

@@ -120,7 +120,7 @@ func (m *impl) Empty() bool {
 }
 
 func (m *impl) Wantlist() []Entry {
-	var out []Entry
+	out := make([]Entry, 0, len(m.wantlist))
 	for _, e := range m.wantlist {
 		out = append(out, e)
 	}
@@ -182,6 +182,7 @@ func FromPBReader(pbr ggio.Reader) (BitSwapMessage, error) {
 func (m *impl) ToProtoV0() *pb.Message {
 	pbm := new(pb.Message)
 	pbm.Wantlist = new(pb.Message_Wantlist)
+	pbm.Wantlist.Entries = make([]*pb.Message_Wantlist_Entry, 0, len(m.wantlist))
 	for _, e := range m.wantlist {
 		pbm.Wantlist.Entries = append(pbm.Wantlist.Entries, &pb.Message_Wantlist_Entry{
 			Block:    proto.String(e.Cid.KeyString()),
@@ -190,7 +191,10 @@ func (m *impl) ToProtoV0() *pb.Message {
 		})
 	}
 	pbm.Wantlist.Full = proto.Bool(m.full)
-	for _, b := range m.Blocks() {
+
+	blocks := m.Blocks()
+	pbm.Blocks = make([][]byte, 0, len(blocks))
+	for _, b := range blocks {
 		pbm.Blocks = append(pbm.Blocks, b.RawData())
 	}
 	return pbm
@@ -199,6 +203,7 @@ func (m *impl) ToProtoV0() *pb.Message {
 func (m *impl) ToProtoV1() *pb.Message {
 	pbm := new(pb.Message)
 	pbm.Wantlist = new(pb.Message_Wantlist)
+	pbm.Wantlist.Entries = make([]*pb.Message_Wantlist_Entry, 0, len(m.wantlist))
 	for _, e := range m.wantlist {
 		pbm.Wantlist.Entries = append(pbm.Wantlist.Entries, &pb.Message_Wantlist_Entry{
 			Block:    proto.String(e.Cid.KeyString()),
@@ -207,7 +212,10 @@ func (m *impl) ToProtoV1() *pb.Message {
 		})
 	}
 	pbm.Wantlist.Full = proto.Bool(m.full)
-	for _, b := range m.Blocks() {
+
+	blocks := m.Blocks()
+	pbm.Payload = make([]*pb.Message_Block, 0, len(blocks))
+	for _, b := range blocks {
 		blk := &pb.Message_Block{
 			Data:   b.RawData(),
 			Prefix: b.Cid().Prefix().Bytes(),
@@ -230,7 +238,7 @@ func (m *impl) ToNetV1(w io.Writer) error {
 }
 
 func (m *impl) Loggable() map[string]interface{} {
-	var blocks []string
+	blocks := make([]string, 0, len(m.blocks))
 	for _, v := range m.blocks {
 		blocks = append(blocks, v.Cid().String())
 	}

exchange/bitswap/notifications/notifications.go

@@ -73,7 +73,7 @@ func (ps *impl) Subscribe(ctx context.Context, keys ...*cid.Cid) <-chan blocks.B
 }
 
 func toStrings(keys []*cid.Cid) []string {
-	strs := make([]string, 0)
+	strs := make([]string, 0, len(keys))
 	for _, key := range keys {
 		strs = append(strs, key.KeyString())
 	}

exchange/bitswap/session.go

@@ -199,11 +199,12 @@ func (s *Session) run(ctx context.Context) {
 			s.cancel(keys)
 
 		case <-s.tick.C:
-			var live []*cid.Cid
+			live := make([]*cid.Cid, 0, len(s.liveWants))
+			now := time.Now()
 			for c := range s.liveWants {
 				cs, _ := cid.Cast([]byte(c))
 				live = append(live, cs)
-				s.liveWants[c] = time.Now()
+				s.liveWants[c] = now
 			}
 
 			// Broadcast these keys to everyone we're connected to

exchange/bitswap/stat.go

@@ -32,7 +32,10 @@ func (bs *Bitswap) Stat() (*Stat, error) {
 	st.DataReceived = c.dataRecvd
 	bs.counterLk.Unlock()
 
-	for _, p := range bs.engine.Peers() {
+	peers := bs.engine.Peers()
+	st.Peers = make([]string, 0, len(peers))
+
+	for _, p := range peers {
 		st.Peers = append(st.Peers, p.Pretty())
 	}
 	sort.Strings(st.Peers)

exchange/bitswap/wantlist/wantlist.go

@@ -126,7 +126,7 @@ func (w *ThreadSafe) Contains(k *cid.Cid) (*Entry, bool) {
 func (w *ThreadSafe) Entries() []*Entry {
 	w.lk.RLock()
 	defer w.lk.RUnlock()
-	var es entrySlice
+	es := make([]*Entry, 0, len(w.set))
 	for _, e := range w.set {
 		es = append(es, e)
 	}
@@ -134,13 +134,8 @@ func (w *ThreadSafe) Entries() []*Entry {
 }
 
 func (w *ThreadSafe) SortedEntries() []*Entry {
-	w.lk.RLock()
-	defer w.lk.RUnlock()
-	var es entrySlice
-	for _, e := range w.set {
-		es = append(es, e)
-	}
-	sort.Sort(es)
+	es := w.Entries()
+	sort.Sort(entrySlice(es))
 	return es
 }
 
@@ -194,7 +189,7 @@ func (w *Wantlist) Contains(k *cid.Cid) (*Entry, bool) {
 }
 
 func (w *Wantlist) Entries() []*Entry {
-	var es entrySlice
+	es := make([]*Entry, 0, len(w.set))
 	for _, e := range w.set {
 		es = append(es, e)
 	}
@@ -202,10 +197,7 @@ func (w *Wantlist) Entries() []*Entry {
 }
 
 func (w *Wantlist) SortedEntries() []*Entry {
-	var es entrySlice
-	for _, e := range w.set {
-		es = append(es, e)
-	}
-	sort.Sort(es)
+	es := w.Entries()
+	sort.Sort(entrySlice(es))
 	return es
 }

exchange/bitswap/wantmanager.go

@@ -94,7 +94,7 @@ type wantSet struct {
 }
 
 func (pm *WantManager) addEntries(ctx context.Context, ks []*cid.Cid, targets []peer.ID, cancel bool, ses uint64) {
-	var entries []*bsmsg.Entry
+	entries := make([]*bsmsg.Entry, 0, len(ks))
 	for i, k := range ks {
 		entries = append(entries, &bsmsg.Entry{
 			Cancel: cancel,
@@ -340,7 +340,7 @@ func (pm *WantManager) Run() {
 				pm.stopPeerHandler(p.peer)
 			}
 		case req := <-pm.peerReqs:
-			var peers []peer.ID
+			peers := make([]peer.ID, 0, len(pm.peers))
 			for p := range pm.peers {
 				peers = append(peers, p)
 			}