p2p, network, bzzeth: p2p protocol handlers to async (#2018)

p2p, network, bzzeth, pss: use async run from p2p/protocol

Author: pradovic

bzzeth/bzzeth.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/hex"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -102,11 +103,11 @@ func (b *BzzEth) handleMsg(p *Peer) func(context.Context, interface{}) error {
 		p.logger.Trace("bzzeth.handleMsg")
 		switch msg := msg.(type) {
 		case *NewBlockHeaders:
-			go b.handleNewBlockHeaders(ctx, p, msg)
+			return b.handleNewBlockHeaders(ctx, p, msg)
 		case *BlockHeaders:
-			go b.handleBlockHeaders(ctx, p, msg)
+			return b.handleBlockHeaders(ctx, p, msg)
 		case *GetBlockHeaders:
-			go b.handleGetBlockHeaders(ctx, p, msg)
+			return b.handleGetBlockHeaders(ctx, p, msg)
 		}
 		return nil
 	}
@@ -123,7 +124,7 @@ func (b *BzzEth) handleMsgFromSwarmNode(p *Peer) func(context.Context, interface
 
 // handleNewBlockHeaders handles new header hashes
 // only request headers that are in Kad Nearest Neighbourhood
-func (b *BzzEth) handleNewBlockHeaders(ctx context.Context, p *Peer, msg *NewBlockHeaders) {
+func (b *BzzEth) handleNewBlockHeaders(ctx context.Context, p *Peer, msg *NewBlockHeaders) error {
 	p.logger.Trace("bzzeth.handleNewBlockHeaders")
 
 	// collect the addresses of blocks that are not in our localstore
@@ -135,7 +136,7 @@ func (b *BzzEth) handleNewBlockHeaders(ctx context.Context, p *Peer, msg *NewBlo
 	yes, err := b.netStore.Store.HasMulti(ctx, addresses...)
 	if err != nil {
 		log.Error("Error checking hashesh in store", "Reason", err)
-		return
+		return nil
 	}
 
 	// collect the hashes of block headers we want
@@ -160,7 +161,7 @@ func (b *BzzEth) handleNewBlockHeaders(ctx context.Context, p *Peer, msg *NewBlo
 	req, err := p.getBlockHeaders(ctx, hashes, deliveries)
 	if err != nil {
 		p.logger.Error("Error sending GetBlockHeader message", "Reason", err)
-		return
+		return nil
 	}
 	defer req.cancel()
 
@@ -172,19 +173,24 @@ func (b *BzzEth) handleNewBlockHeaders(ctx context.Context, p *Peer, msg *NewBlo
 		case hdr, ok := <-deliveries:
 			if !ok {
 				p.logger.Debug("bzzeth.handleNewBlockHeaders", "delivered", deliveredCnt)
-				return
+				// todo: introduce better errors
+				return nil
 			}
 			ch := newChunk(hdr)
 			deliveredCnt++
 			p.logger.Trace("bzzeth.handleNewBlockHeaders", "hash", ch.Address().Hex(), "delivered", deliveredCnt)
+
+			req.lock.RLock()
 			if deliveredCnt == len(req.hashes) {
 				p.logger.Debug("all headers delivered", "count", deliveredCnt)
 				finishDeliveryFunc(req.hashes)
-				return
+				req.lock.RUnlock()
+				return nil
 			}
+			req.lock.RUnlock()
 		case <-ctx.Done():
 			p.logger.Debug("bzzeth.handleNewBlockHeaders", "delivered", deliveredCnt, "err", err)
-			return
+			return nil
 		}
 	}
 }
@@ -220,15 +226,14 @@ func finishDelivery(hashes map[string]bool) {
 }
 
 // handleBlockHeaders handles block headers message
-func (b *BzzEth) handleBlockHeaders(ctx context.Context, p *Peer, msg *BlockHeaders) {
+func (b *BzzEth) handleBlockHeaders(ctx context.Context, p *Peer, msg *BlockHeaders) error {
 	p.logger.Debug("bzzeth.handleBlockHeaders", "id", msg.Rid)
 
 	// retrieve the request for this id
 	req, ok := p.requests.get(msg.Rid)
 	if !ok {
-		p.logger.Warn("bzzeth.handleBlockHeaders: nonexisting request id", "id", msg.Rid)
-		p.Drop("nonexisting request id")
-		return
+		return fmt.Errorf("bzzeth.handleBlockHeaders: nonexisting request id %d", msg.Rid)
+
 	}
 
 	// convert rlp.RawValue to bytes
@@ -237,11 +242,7 @@ func (b *BzzEth) handleBlockHeaders(ctx context.Context, p *Peer, msg *BlockHead
 		headers[i] = h
 	}
 
-	err := b.deliverAndStoreAll(ctx, req, headers)
-	if err != nil {
-		p.logger.Warn("bzzeth.handleBlockHeaders: fatal dropping peer", "id", msg.Rid, "err", err)
-		p.Drop("error on deliverAndStoreAll")
-	}
+	return b.deliverAndStoreAll(ctx, req, headers)
 }
 
 // Validates and headers asynchronously and stores the valid chunks in one go
@@ -263,13 +264,14 @@ func (b *BzzEth) deliverAndStoreAll(ctx context.Context, req *request, headers [
 			return nil
 		})
 	}
-	// finish storage is used mostly in testing
-	// in normal scenario.. it just logs Trace
-	defer finishStorageFunc(chunks)
 
 	// wait for all validations to get over and close the channels
 	err := wg.Wait()
 
+	// finish storage is used mostly in testing
+	// in normal scenario.. it just logs Trace
+	defer finishStorageFunc(chunks)
+
 	// We want to store even if there is any validation error.
 	// since some headers may be valid in the batch.
 	// Store all the valid header chunks in one shot
@@ -354,7 +356,7 @@ func arrangeHeader(hashes []chunk.Address, headers []chunk.Address) []chunk.Addr
 
 // handles GetBlockHeader requests, in the protocol handler this call is asynchronous
 // so it is safe to have it run until delivery is finished
-func (b *BzzEth) handleGetBlockHeaders(ctx context.Context, p *Peer, msg *GetBlockHeaders) {
+func (b *BzzEth) handleGetBlockHeaders(ctx context.Context, p *Peer, msg *GetBlockHeaders) error {
 	p.logger.Debug("bzzeth.handleGetBlockHeaders", "id", msg.Rid)
 	total := len(msg.Hashes)
 	ctx, osp := spancontext.StartSpan(ctx, "bzzeth.handleGetBlockHeaders")
@@ -420,6 +422,8 @@ DELIVERY:
 		}
 	}
 	p.logger.Debug("bzzeth.handleGetBlockHeaders: sent all headers", "id", msg.Rid)
+
+	return nil
 }
 
 var batchWait = 100 * time.Millisecond // time to wait for collecting headers in a batch

network/hive.go

@@ -390,24 +390,30 @@ func (h *Hive) handlePeersMsg(d *Peer, msg *peersMsg) error {
 func (h *Hive) handleSubPeersMsg(ctx context.Context, d *Peer, msg *subPeersMsg) error {
 	d.setDepth(msg.Depth)
 	// only send peers after the initial subPeersMsg
-	if !d.sentPeers {
-		var peers []*BzzAddr
-		// iterate connection in ascending order of disctance from the remote address
-		h.EachConn(d.Over(), 255, func(p *Peer, po int) bool {
-			// terminate if we are beyond the radius
-			if uint8(po) < msg.Depth {
-				return false
-			}
-			if !d.seen(p.BzzAddr) { // here just records the peer sent
-				peers = append(peers, p.BzzAddr)
-			}
-			return true
-		})
-		// if useful  peers are found, send them over
-		if len(peers) > 0 {
-			go d.Send(ctx, &peersMsg{Peers: sortPeers(peers)})
-		}
+	h.lock.Lock()
+	if d.sentPeers {
+		h.lock.Unlock()
+		return nil
 	}
+
 	d.sentPeers = true
+	h.lock.Unlock()
+
+	var peers []*BzzAddr
+	// iterate connection in ascending order of distance from the remote address
+	h.EachConn(d.Over(), 255, func(p *Peer, po int) bool {
+		// terminate if we are beyond the radius
+		if uint8(po) < msg.Depth {
+			return false
+		}
+		if !d.seen(p.BzzAddr) { // here just records the peer sent
+			peers = append(peers, p.BzzAddr)
+		}
+		return true
+	})
+	// if useful  peers are found, send them over
+	if len(peers) > 0 {
+		go d.Send(ctx, &peersMsg{Peers: sortPeers(peers)})
+	}
 	return nil
 }

network/retrieval/retrieve.go

@@ -157,11 +157,9 @@ func (r *Retrieval) handleMsg(p *Peer) func(context.Context, interface{}) error
 	return func(ctx context.Context, msg interface{}) error {
 		switch msg := msg.(type) {
 		case *RetrieveRequest:
-			// we must handle them in a different goroutine otherwise parallel requests
-			// for other chunks from the same peer will get stuck in the queue
-			go r.handleRetrieveRequest(ctx, p, msg)
+			return r.handleRetrieveRequest(ctx, p, msg)
 		case *ChunkDelivery:
-			go r.handleChunkDelivery(ctx, p, msg)
+			return r.handleChunkDelivery(ctx, p, msg)
 		}
 		return nil
 	}
@@ -296,7 +294,7 @@ func (r *Retrieval) findPeerLB(ctx context.Context, req *storage.Request) (retPe
 // handleRetrieveRequest handles an incoming retrieve request from a certain Peer
 // if the chunk is found in the localstore it is served immediately, otherwise
 // it results in a new retrieve request to candidate peers in our kademlia
-func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *RetrieveRequest) {
+func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *RetrieveRequest) error {
 	p.logger.Debug("retrieval.handleRetrieveRequest", "ref", msg.Addr)
 	handleRetrieveRequestMsgCount.Inc(1)
 
@@ -319,7 +317,8 @@ func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *Ret
 	if err != nil {
 		retrieveChunkFail.Inc(1)
 		p.logger.Trace("netstore.Get can not retrieve chunk", "ref", msg.Addr, "err", err)
-		return
+		// continue in event loop
+		return nil
 	}
 
 	p.logger.Trace("retrieval.handleRetrieveRequest - delivery", "ref", msg.Addr)
@@ -334,22 +333,23 @@ func (r *Retrieval) handleRetrieveRequest(ctx context.Context, p *Peer, msg *Ret
 	if err != nil {
 		p.logger.Error("retrieval.handleRetrieveRequest - peer delivery failed", "ref", msg.Addr, "err", err)
 		osp.LogFields(olog.Bool("delivered", false))
-		return
+		// continue in event loop
+		return nil
 	}
 	osp.LogFields(olog.Bool("delivered", true))
+
+	return nil
 }
 
 // handleChunkDelivery handles a ChunkDelivery message from a certain peer
 // if the chunk proximity order in relation to our base address is within depth
 // we treat the chunk as a chunk received in syncing
-func (r *Retrieval) handleChunkDelivery(ctx context.Context, p *Peer, msg *ChunkDelivery) {
+func (r *Retrieval) handleChunkDelivery(ctx context.Context, p *Peer, msg *ChunkDelivery) error {
 	p.logger.Debug("retrieval.handleChunkDelivery", "ref", msg.Addr)
 	err := p.checkRequest(msg.Ruid, msg.Addr)
 	if err != nil {
 		unsolicitedChunkDelivery.Inc(1)
-		p.logger.Error("unsolicited chunk delivery from peer", "ruid", msg.Ruid, "addr", msg.Addr, "err", err)
-		p.Drop("unsolicited chunk delivery")
-		return
+		return fmt.Errorf("unsolicited chunk delivery from peer: %s", err)
 	}
 	var osp opentracing.Span
 	ctx, osp = spancontext.StartSpan(
@@ -380,9 +380,11 @@ func (r *Retrieval) handleChunkDelivery(ctx context.Context, p *Peer, msg *Chunk
 	if err != nil {
 		p.logger.Error("netstore error putting chunk to localstore", "err", err)
 		if err == storage.ErrChunkInvalid {
-			p.Drop("invalid chunk in netstore put")
+			return fmt.Errorf("netstore error putting chunk to localstore: %s", err)
 		}
 	}
+
+	return nil
 }
 
 // RequestFromPeers sends a chunk retrieve request to the next found peer

network/retrieval/retrieve_test.go

@@ -172,7 +172,7 @@ func TestUnsolicitedChunkDelivery(t *testing.T) {
 		})
 
 	// expect peer disconnection
-	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("subprotocol error")})
+	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("Message handler error: (msg code 0): unsolicited chunk delivery from peer: cannot find ruid")})
 
 	if err != nil {
 		t.Fatal(err)
@@ -246,7 +246,7 @@ func TestUnsolicitedChunkDeliveryFaultyAddr(t *testing.T) {
 	}
 
 	// expect disconnection
-	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("subprotocol error")})
+	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("Message handler error: (msg code 0): unsolicited chunk delivery from peer: retrieve request found but address does not match")})
 
 	if err != nil {
 		t.Fatal(err)
@@ -334,7 +334,7 @@ func TestUnsolicitedChunkDeliveryDouble(t *testing.T) {
 	}
 
 	// expect disconnection
-	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("subprotocol error")})
+	err = tester.TestDisconnected(&p2ptest.Disconnect{Peer: node.ID(), Error: errors.New("Message handler error: (msg code 0): unsolicited chunk delivery from peer: cannot find ruid")})
 
 	if err != nil {
 		t.Fatal(err)

network/stream/common_test.go

@@ -275,15 +275,15 @@ type syncPauser struct {
 	mu  sync.RWMutex
 }
 
-func (p *syncPauser) pause() {
+func (p *syncPauser) Pause() {
 	p.mu.Lock()
 	if p.c == nil {
 		p.c = sync.NewCond(&p.cMu)
 	}
 	p.mu.Unlock()
 }
 
-func (p *syncPauser) resume() {
+func (p *syncPauser) Resume() {
 	p.c.L.Lock()
 	p.c.Broadcast()
 	p.c.L.Unlock()
@@ -292,7 +292,7 @@ func (p *syncPauser) resume() {
 	p.mu.Unlock()
 }
 
-func (p *syncPauser) wait() {
+func (p *syncPauser) Wait() {
 	p.mu.RLock()
 	if p.c != nil {
 		p.c.L.Lock()