use separate channel for blockSyncMsg (iotexproject#2411)

* use separate channel for blockSyncMsg

Author: dustinxie

action/protocol/poll/blockmeta.go

@@ -58,7 +58,7 @@ func (bm *BlockMeta) Proto() (*blockmetapb.BlockMeta, error) {
 func (bm *BlockMeta) Deserialize(buf []byte) error {
 	epochMetapb := &blockmetapb.BlockMeta{}
 	if err := proto.Unmarshal(buf, epochMetapb); err != nil {
-		return errors.Wrap(err, "failed to unmarshal blacklist")
+		return errors.Wrap(err, "failed to unmarshal blocklist")
 	}
 	return bm.LoadProto(epochMetapb)
 }

blocksync/blocksync.go

@@ -8,6 +8,7 @@ package blocksync
 
 import (
 	"context"
+	"time"
 
 	"github.com/golang/protobuf/proto"
 	peerstore "github.com/libp2p/go-libp2p-peerstore"
@@ -71,13 +72,14 @@ type BlockSync interface {
 
 // blockSyncer implements BlockSync interface
 type blockSyncer struct {
-	commitHeight     uint64 // last commit block height
-	buf              *blockBuffer
-	worker           *syncWorker
-	bc               blockchain.Blockchain
-	dao              BlockDAO
-	unicastHandler   UnicastOutbound
-	neighborsHandler Neighbors
+	commitHeight          uint64 // last commit block height
+	processSyncRequestTTL time.Duration
+	buf                   *blockBuffer
+	worker                *syncWorker
+	bc                    blockchain.Blockchain
+	dao                   BlockDAO
+	unicastHandler        UnicastOutbound
+	neighborsHandler      Neighbors
 }
 
 // NewBlockSyncer returns a new block syncer instance
@@ -102,12 +104,13 @@ func NewBlockSyncer(
 		}
 	}
 	bs := &blockSyncer{
-		bc:               chain,
-		dao:              dao,
-		buf:              buf,
-		unicastHandler:   bsCfg.unicastHandler,
-		neighborsHandler: bsCfg.neighborsHandler,
-		worker:           newSyncWorker(chain.ChainID(), cfg, bsCfg.unicastHandler, bsCfg.neighborsHandler, buf),
+		bc:                    chain,
+		dao:                   dao,
+		buf:                   buf,
+		unicastHandler:        bsCfg.unicastHandler,
+		neighborsHandler:      bsCfg.neighborsHandler,
+		worker:                newSyncWorker(chain.ChainID(), cfg, bsCfg.unicastHandler, bsCfg.neighborsHandler, buf),
+		processSyncRequestTTL: cfg.BlockSync.ProcessSyncRequestTTL,
 	}
 	return bs, nil
 }
@@ -184,9 +187,9 @@ func (bs *blockSyncer) ProcessSyncRequest(ctx context.Context, peer peerstore.Pe
 			return err
 		}
 		// TODO: send back multiple blocks in one shot
-		if err := bs.unicastHandler(context.Background(), peer,
-			blk.ConvertToBlockPb(),
-		); err != nil {
+		syncCtx, cancel := context.WithTimeout(ctx, bs.processSyncRequestTTL)
+		defer cancel()
+		if err := bs.unicastHandler(syncCtx, peer, blk.ConvertToBlockPb()); err != nil {
 			log.L().Debug("Failed to response to ProcessSyncRequest.", zap.Error(err))
 		}
 	}

config/config.go

@@ -155,11 +155,12 @@ var (
 			},
 		},
 		BlockSync: BlockSync{
-			Interval:        10 * time.Second,
-			BufferSize:      200,
-			IntervalSize:    20,
-			MaxRepeat:       3,
-			RepeatDecayStep: 1,
+			Interval:              10 * time.Second,
+			ProcessSyncRequestTTL: 10 * time.Second,
+			BufferSize:            200,
+			IntervalSize:          20,
+			MaxRepeat:             3,
+			RepeatDecayStep:       1,
 		},
 		Dispatcher: Dispatcher{
 			EventChanSize: 10000,
@@ -274,9 +275,10 @@ type (
 
 	// BlockSync is the config struct for the BlockSync
 	BlockSync struct {
-		Interval     time.Duration `yaml:"interval"` // update duration
-		BufferSize   uint64        `yaml:"bufferSize"`
-		IntervalSize uint64        `yaml:"intervalSize"`
+		Interval              time.Duration `yaml:"interval"` // update duration
+		ProcessSyncRequestTTL time.Duration `yaml:"processSyncRequestTTL"`
+		BufferSize            uint64        `yaml:"bufferSize"`
+		IntervalSize          uint64        `yaml:"intervalSize"`
 		// MaxRepeat is the maximal number of repeat of a block sync request
 		MaxRepeat int `yaml:"maxRepeat"`
 		// RepeatDecayStep is the step for repeat number decreasing by 1

dispatcher/dispatcher.go

@@ -100,19 +100,20 @@ type IotxDispatcher struct {
 	started        int32
 	shutdown       int32
 	eventChan      chan interface{}
+	syncChan       chan *blockSyncMsg
 	eventAudit     map[iotexrpc.MessageType]int
 	eventAuditLock sync.RWMutex
 	wg             sync.WaitGroup
 	quit           chan struct{}
-
-	subscribers   map[uint32]Subscriber
-	subscribersMU sync.RWMutex
+	subscribers    map[uint32]Subscriber
+	subscribersMU  sync.RWMutex
 }
 
 // NewDispatcher creates a new Dispatcher
 func NewDispatcher(cfg config.Config) (Dispatcher, error) {
 	d := &IotxDispatcher{
 		eventChan:   make(chan interface{}, cfg.Dispatcher.EventChanSize),
+		syncChan:    make(chan *blockSyncMsg, cfg.Dispatcher.EventChanSize),
 		eventAudit:  make(map[iotexrpc.MessageType]int),
 		quit:        make(chan struct{}),
 		subscribers: make(map[uint32]Subscriber),
@@ -136,8 +137,10 @@ func (d *IotxDispatcher) Start(ctx context.Context) error {
 		return errors.New("Dispatcher already started")
 	}
 	log.L().Info("Starting dispatcher.")
-	d.wg.Add(1)
+	d.wg.Add(2)
 	go d.newsHandler()
+	go d.syncHandler()
+
 	return nil
 }
 
@@ -153,9 +156,11 @@ func (d *IotxDispatcher) Stop(ctx context.Context) error {
 	return nil
 }
 
-// EventChan returns the event chan
-func (d *IotxDispatcher) EventChan() *chan interface{} {
-	return &d.eventChan
+// EventQueueSize returns the event queue size
+func (d *IotxDispatcher) EventQueueSize() int {
+	d.eventAuditLock.RLock()
+	defer d.eventAuditLock.RUnlock()
+	return len(d.eventChan) + len(d.syncChan)
 }
 
 // EventAudit returns the event audit map
@@ -180,26 +185,43 @@ loop:
 				d.handleActionMsg(msg)
 			case *blockMsg:
 				d.handleBlockMsg(msg)
-			case *blockSyncMsg:
-				d.handleBlockSyncMsg(msg)
-
 			default:
 				log.L().Warn("Invalid message type in block handler.", zap.Any("msg", msg))
 			}
+		case <-d.quit:
+			break loop
+		}
+	}
+
+	d.wg.Done()
+	log.L().Info("news handler done.")
+}
 
+// syncHandler handles incoming block sync requests
+func (d *IotxDispatcher) syncHandler() {
+loop:
+	for {
+		select {
+		case m := <-d.syncChan:
+			d.handleBlockSyncMsg(m)
 		case <-d.quit:
 			break loop
 		}
 	}
 
 	d.wg.Done()
-	log.L().Info("News handler done.")
+	log.L().Info("block sync handler done.")
 }
 
 // handleActionMsg handles actionMsg from all peers.
 func (d *IotxDispatcher) handleActionMsg(m *actionMsg) {
-	d.updateEventAudit(iotexrpc.MessageType_ACTION)
-	if subscriber, ok := d.subscribers[m.ChainID()]; ok {
+	log.L().Debug("receive actionMsg.")
+
+	d.subscribersMU.RLock()
+	subscriber, ok := d.subscribers[m.ChainID()]
+	d.subscribersMU.RUnlock()
+	if ok {
+		d.updateEventAudit(iotexrpc.MessageType_ACTION)
 		if err := subscriber.HandleAction(m.ctx, m.action); err != nil {
 			requestMtc.WithLabelValues("AddAction", "false").Inc()
 			log.L().Debug("Handle action request error.", zap.Error(err))
@@ -214,8 +236,9 @@ func (d *IotxDispatcher) handleBlockMsg(m *blockMsg) {
 	log.L().Debug("receive blockMsg.", zap.Uint64("height", m.block.GetHeader().GetCore().GetHeight()))
 
 	d.subscribersMU.RLock()
-	defer d.subscribersMU.RUnlock()
-	if subscriber, ok := d.subscribers[m.ChainID()]; ok {
+	subscriber, ok := d.subscribers[m.ChainID()]
+	d.subscribersMU.RUnlock()
+	if ok {
 		d.updateEventAudit(iotexrpc.MessageType_BLOCK)
 		if err := subscriber.HandleBlock(m.ctx, m.block); err != nil {
 			log.L().Error("Fail to handle the block.", zap.Error(err))
@@ -232,8 +255,11 @@ func (d *IotxDispatcher) handleBlockSyncMsg(m *blockSyncMsg) {
 		zap.Uint64("start", m.sync.Start),
 		zap.Uint64("end", m.sync.End))
 
-	d.updateEventAudit(iotexrpc.MessageType_BLOCK_REQUEST)
-	if subscriber, ok := d.subscribers[m.ChainID()]; ok {
+	d.subscribersMU.RLock()
+	subscriber, ok := d.subscribers[m.ChainID()]
+	d.subscribersMU.RUnlock()
+	if ok {
+		d.updateEventAudit(iotexrpc.MessageType_BLOCK_REQUEST)
 		// dispatch to block sync
 		if err := subscriber.HandleSyncRequest(m.ctx, m.peer, m.sync); err != nil {
 			log.L().Error("Failed to handle sync request.", zap.Error(err))
@@ -272,12 +298,17 @@ func (d *IotxDispatcher) dispatchBlockSyncReq(ctx context.Context, chainID uint3
 	if atomic.LoadInt32(&d.shutdown) != 0 {
 		return
 	}
-	d.enqueueEvent(&blockSyncMsg{
+
+	if len(d.syncChan) == cap(d.syncChan) {
+		log.L().Warn("dispatcher sync chan is full, drop an event.")
+		return
+	}
+	d.syncChan <- &blockSyncMsg{
 		ctx:     ctx,
 		chainID: chainID,
 		peer:    peer,
 		sync:    (msg).(*iotexrpc.BlockSync),
-	})
+	}
 }
 
 // HandleBroadcast handles incoming broadcast message
@@ -288,12 +319,11 @@ func (d *IotxDispatcher) HandleBroadcast(ctx context.Context, chainID uint32, me
 	}
 	d.subscribersMU.RLock()
 	subscriber, ok := d.subscribers[chainID]
+	d.subscribersMU.RUnlock()
 	if !ok {
 		log.L().Warn("chainID has not been registered in dispatcher.", zap.Uint32("chainID", chainID))
-		d.subscribersMU.RUnlock()
 		return
 	}
-	d.subscribersMU.RUnlock()
 
 	switch msgType {
 	case iotexrpc.MessageType_CONSENSUS:
@@ -326,13 +356,11 @@ func (d *IotxDispatcher) HandleTell(ctx context.Context, chainID uint32, peer pe
 }
 
 func (d *IotxDispatcher) enqueueEvent(event interface{}) {
-	go func() {
-		if len(d.eventChan) == cap(d.eventChan) {
-			log.L().Debug("dispatcher event chan is full, drop an event.")
-			return
-		}
-		d.eventChan <- event
-	}()
+	if len(d.eventChan) == cap(d.eventChan) {
+		log.L().Warn("dispatcher event chan is full, drop an event.")
+		return
+	}
+	d.eventChan <- event
 }
 
 func (d *IotxDispatcher) updateEventAudit(t iotexrpc.MessageType) {

p2p/agent.go

@@ -26,7 +26,6 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/iotexproject/iotex-core/config"
-	"github.com/iotexproject/iotex-core/pkg/cache"
 	"github.com/iotexproject/iotex-core/pkg/log"
 	goproto "github.com/iotexproject/iotex-proto/golang"
 	"github.com/iotexproject/iotex-proto/golang/iotexrpc"
@@ -66,8 +65,6 @@ const (
 	unicastTopic      = "unicast"
 	numDialRetries    = 8
 	dialRetryInterval = 2 * time.Second
-	blackListLen      = 1000
-	blackListTTL      = 15 * time.Minute
 )
 
 type (
@@ -85,7 +82,7 @@ type Agent struct {
 	broadcastInboundHandler    HandleBroadcastInbound
 	unicastInboundAsyncHandler HandleUnicastInboundAsync
 	host                       *p2p.Host
-	unicastBlacklist           *cache.ThreadSafeLruCache
+	unicastBlocklist           *BlockList
 }
 
 // NewAgent instantiates a local P2P agent instance
@@ -97,7 +94,7 @@ func NewAgent(cfg config.Config, broadcastHandler HandleBroadcastInbound, unicas
 		topicSuffix:                hex.EncodeToString(gh[22:]), // last 10 bytes of genesis hash
 		broadcastInboundHandler:    broadcastHandler,
 		unicastInboundAsyncHandler: unicastHandler,
-		unicastBlacklist:           cache.NewThreadSafeLruCache(blackListLen),
+		unicastBlocklist:           NewBlockList(blockListLen),
 	}
 }
 
@@ -338,15 +335,24 @@ func (p *Agent) BroadcastOutbound(ctx context.Context, msg proto.Message) (err e
 
 // UnicastOutbound sends a unicast message to the given address
 func (p *Agent) UnicastOutbound(ctx context.Context, peer peerstore.PeerInfo, msg proto.Message) (err error) {
-	var msgType iotexrpc.MessageType
-	var msgBody []byte
+	var (
+		peerName = peer.ID.Pretty()
+		msgType  iotexrpc.MessageType
+		msgBody  []byte
+	)
 	defer func() {
 		status := successStr
 		if err != nil {
 			status = failureStr
 		}
 		p2pMsgCounter.WithLabelValues("unicast", strconv.Itoa(int(msgType)), "out", peer.ID.Pretty(), status).Inc()
 	}()
+
+	if p.unicastBlocklist.Blocked(peerName, time.Now()) {
+		err = errors.New("peer is in blocklist at this moment")
+		return
+	}
+
 	msgType, msgBody, err = convertAppMsg(msg)
 	if err != nil {
 		return
@@ -366,14 +372,18 @@ func (p *Agent) UnicastOutbound(ctx context.Context, peer peerstore.PeerInfo, ms
 	data, err := proto.Marshal(&unicast)
 	if err != nil {
 		err = errors.Wrap(err, "error when marshaling unicast message")
-		return err
+		return
 	}
+
 	if err = p.host.Unicast(ctx, peer, unicastTopic+p.topicSuffix, data); err != nil {
 		err = errors.Wrap(err, "error when sending unicast message")
-		p.unicastBlacklist.Add(peer.ID.Pretty(), time.Now().Add(blackListTTL))
-		return err
+		p.unicastBlocklist.Add(peerName, time.Now())
+		return
 	}
-	return err
+
+	// remove peer from blocklist upon success
+	p.unicastBlocklist.Remove(peerName)
+	return
 }
 
 // Info returns agents' peer info.
@@ -391,12 +401,7 @@ func (p *Agent) Neighbors(ctx context.Context) ([]peerstore.PeerInfo, error) {
 	}
 
 	for i, nb := range nbs {
-		if v, ok := p.unicastBlacklist.Get(nb.ID.Pretty()); ok {
-			if t, isT := v.(time.Time); isT {
-				if time.Now().After(t) {
-					p.unicastBlacklist.Remove(nb.ID.Pretty())
-				}
-			}
+		if p.unicastBlocklist.Blocked(nb.ID.Pretty(), time.Now()) {
 			continue
 		}
 		res = append(res, nbs[i])