Merge pull request #3592 from JackyWYX/stream_syncprotocol_reviewfix

[Stream] fix some sync protocol comments

p2p/stream/protocols/sync/client.go

@@ -18,6 +18,9 @@ import (
 // GetBlocksByNumber do getBlocksByNumberRequest through sync stream protocol.
 // Return the block as result, target stream id, and error
 func (p *Protocol) GetBlocksByNumber(ctx context.Context, bns []uint64, opts ...Option) ([]*types.Block, sttypes.StreamID, error) {
+	if len(bns) == 0 {
+		return nil, "", fmt.Errorf("zero block numbers requested")
+	}
 	if len(bns) > GetBlocksByNumAmountCap {
 		return nil, "", fmt.Errorf("number of blocks exceed cap of %v", GetBlocksByNumAmountCap)
 	}
@@ -57,6 +60,9 @@ func (p *Protocol) GetCurrentBlockNumber(ctx context.Context, opts ...Option) (u
 // GetBlockHashes do getBlockHashesRequest through sync stream protocol.
 // Return the hash of the given block number. If a block is unknown, the hash will be emptyHash.
 func (p *Protocol) GetBlockHashes(ctx context.Context, bns []uint64, opts ...Option) ([]common.Hash, sttypes.StreamID, error) {
+	if len(bns) == 0 {
+		return nil, "", fmt.Errorf("zero block numbers requested")
+	}
 	if len(bns) > GetBlockHashesAmountCap {
 		return nil, "", fmt.Errorf("number of requested numbers exceed limit")
 	}
@@ -75,6 +81,9 @@ func (p *Protocol) GetBlockHashes(ctx context.Context, bns []uint64, opts ...Opt
 
 // GetBlocksByHashes do getBlocksByHashesRequest through sync stream protocol.
 func (p *Protocol) GetBlocksByHashes(ctx context.Context, hs []common.Hash, opts ...Option) ([]*types.Block, sttypes.StreamID, error) {
+	if len(hs) == 0 {
+		return nil, "", fmt.Errorf("zero block hashes requested")
+	}
 	if len(hs) > GetBlocksByHashesAmountCap {
 		return nil, "", fmt.Errorf("number of requested hashes exceed limit")
 	}

p2p/stream/protocols/sync/const.go

@@ -3,15 +3,28 @@ package sync
 import "time"
 
 const (
-	// GetBlockHashesAmountCap is the cap of GetBlockHashes reqeust
+	// GetBlockHashesAmountCap is the cap of GetBlockHashes request
 	GetBlockHashesAmountCap = 50
 
-	// GetBlocksByNumAmountCap is the cap of request of a single GetBlocksByNum request
+	// GetBlocksByNumAmountCap is the cap of request of a single GetBlocksByNum request.
+	// This number has an effect on maxMsgBytes as 20MB defined in github.com/harmony-one/harmony/p2p/stream/types.
+	// Since we have an assumption that rlp encoded block size is smaller than 2MB (p2p.node.MaxMessageSize),
+	// so the max size of a stream message is capped at 2MB * 10 = 20MB.
 	GetBlocksByNumAmountCap = 10
 
 	// GetBlocksByHashesAmountCap is the cap of request of single GetBlocksByHashes request
+	// This number has an effect on maxMsgBytes as 20MB defined in github.com/harmony-one/harmony/p2p/stream/types.
+	// See comments for GetBlocksByNumAmountCap.
 	GetBlocksByHashesAmountCap = 10
 
 	// minAdvertiseInterval is the minimum advertise interval
 	minAdvertiseInterval = 1 * time.Minute
+
+	// rateLimiterGlobalRequestPerSecond is the request per second limit for all streams in the sync protocol.
+	// This constant helps prevent the node resource from exhausting for being the stream sync host.
+	rateLimiterGlobalRequestPerSecond = 50
+
+	// rateLimiterSingleRequestsPerSecond is the request per second limit for a single stream in the sync protocol.
+	// This constant helps prevent the node resource from exhausting from a single remote node.
+	rateLimiterSingleRequestsPerSecond = 10
 )

p2p/stream/protocols/sync/protocol.go

@@ -91,7 +91,7 @@ func NewProtocol(config Config) *Protocol {
 	sp.sm = streammanager.NewStreamManager(sp.ProtoID(), config.Host, config.Discovery,
 		sp.HandleStream, smConfig)
 
-	sp.rl = ratelimiter.NewRateLimiter(sp.sm, 50, 10)
+	sp.rl = ratelimiter.NewRateLimiter(sp.sm, rateLimiterGlobalRequestPerSecond, rateLimiterSingleRequestsPerSecond)
 
 	sp.rm = requestmanager.NewRequestManager(sp.sm)