Merge pull request #4873 from harmony-one/refactor/proto_id

Refactor Stream Protocol ID, Add New Protocol for Epoch Sync to P2P Host

Author: GheisMohammadi

api/service/stagedstreamsync/downloader.go

@@ -42,27 +42,28 @@ type (
 )
 
 // NewDownloader creates a new downloader
-func NewDownloader(host p2p.Host, bc core.BlockChain, nodeConfig *nodeconfig.ConfigType, consensus *consensus.Consensus, dbDir string, isBeaconNode bool, config Config) *Downloader {
-	config.fixValues()
+func NewDownloader(host p2p.Host,
+	bc core.BlockChain,
+	nodeConfig *nodeconfig.ConfigType,
+	consensus *consensus.Consensus,
+	dbDir string,
+	isBeaconNode bool,
+	config Config) *Downloader {
 
-	sp := streamSyncProtocol.NewProtocol(streamSyncProtocol.Config{
-		Chain:                bc,
-		Host:                 host.GetP2PHost(),
-		Discovery:            host.GetDiscovery(),
-		ShardID:              nodeconfig.ShardID(bc.ShardID()),
-		Network:              config.Network,
-		BeaconNode:           isBeaconNode,
-		Validator:            nodeConfig.Role() == nodeconfig.Validator,
-		Explorer:             nodeConfig.Role() == nodeconfig.ExplorerNode,
-		MaxAdvertiseWaitTime: config.MaxAdvertiseWaitTime,
-		SmSoftLowCap:         config.SmSoftLowCap,
-		SmHardLowCap:         config.SmHardLowCap,
-		SmHiCap:              config.SmHiCap,
-		DiscBatch:            config.SmDiscBatch,
-	})
+	config.fixValues()
 
+	protoCfg := protocolConfig(host, bc, nodeConfig, isBeaconNode, config)
+	sp := streamSyncProtocol.NewProtocol(*protoCfg)
 	host.AddStreamProtocol(sp)
 
+	// beacon nodes support epoch chain as well
+	if isBeaconNode {
+		epochProtoCfg := protocolConfig(host, bc, nodeConfig, isBeaconNode, config)
+		epochProtoCfg.EpochChain = true
+		epochChainProtocol := streamSyncProtocol.NewProtocol(*epochProtoCfg)
+		host.AddStreamProtocol(epochChainProtocol)
+	}
+
 	logger := utils.Logger().With().
 		Str("module", "StagedStreamSync").
 		Uint32("ShardID", bc.ShardID()).
@@ -101,6 +102,31 @@ func NewDownloader(host p2p.Host, bc core.BlockChain, nodeConfig *nodeconfig.Con
 	}
 }
 
+// protocolConfig returns protocol config
+func protocolConfig(host p2p.Host,
+	bc core.BlockChain,
+	nodeConfig *nodeconfig.ConfigType,
+	isBeaconNode bool,
+	config Config) *streamSyncProtocol.Config {
+
+	return &streamSyncProtocol.Config{
+		Chain:                bc,
+		Host:                 host.GetP2PHost(),
+		Discovery:            host.GetDiscovery(),
+		ShardID:              nodeconfig.ShardID(bc.ShardID()),
+		Network:              config.Network,
+		BeaconNode:           isBeaconNode,
+		Validator:            nodeConfig.Role() == nodeconfig.Validator,
+		Explorer:             nodeConfig.Role() == nodeconfig.ExplorerNode,
+		EpochChain:           !isBeaconNode && bc.ShardID() == shard.BeaconChainShardID,
+		MaxAdvertiseWaitTime: config.MaxAdvertiseWaitTime,
+		SmSoftLowCap:         config.SmSoftLowCap,
+		SmHardLowCap:         config.SmHardLowCap,
+		SmHiCap:              config.SmHiCap,
+		DiscBatch:            config.SmDiscBatch,
+	}
+}
+
 // Start starts the downloader
 func (d *Downloader) Start() {
 	go func() {

hmy/downloader/downloader.go

@@ -18,6 +18,7 @@ import (
 	"github.com/harmony-one/harmony/p2p"
 	"github.com/harmony-one/harmony/p2p/stream/common/streammanager"
 	"github.com/harmony-one/harmony/p2p/stream/protocols/sync"
+	"github.com/harmony-one/harmony/shard"
 )
 
 type (
@@ -56,6 +57,7 @@ func NewDownloader(host p2p.Host, bc core.BlockChain, nodeConfig *nodeconfig.Con
 		BeaconNode:   isBeaconNode,
 		Validator:    nodeConfig.Role() == nodeconfig.Validator,
 		Explorer:     nodeConfig.Role() == nodeconfig.ExplorerNode,
+		EpochChain:   !isBeaconNode && bc.ShardID() == shard.BeaconChainShardID,
 		SmSoftLowCap: config.SmSoftLowCap,
 		SmHardLowCap: config.SmHardLowCap,
 		SmHiCap:      config.SmHiCap,

p2p/stream/common/streammanager/interface_test.go

@@ -18,7 +18,7 @@ var _ StreamManager = &streamManager{}
 
 var (
 	myPeerID    = makePeerID(0)
-	testProtoID = sttypes.ProtoID("harmony/sync/unitest/0/1.0.0/1")
+	testProtoID = sttypes.ProtoID("harmony/sync/unitest/0/1.0.0")
 )
 
 const (

p2p/stream/common/streammanager/streammanager.go

@@ -10,7 +10,6 @@ import (
 	"github.com/harmony-one/abool"
 	"github.com/harmony-one/harmony/internal/utils"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
-	"github.com/harmony-one/harmony/shard"
 	"github.com/libp2p/go-libp2p/core/network"
 	libp2p_peer "github.com/libp2p/go-libp2p/core/peer"
 	"github.com/libp2p/go-libp2p/core/protocol"
@@ -113,12 +112,6 @@ func newStreamManager(pid sttypes.ProtoID, host host, pf peerFinder, handleStrea
 
 	protoSpec, _ := sttypes.ProtoIDToProtoSpec(pid)
 
-	// if it is a beacon node or shard node, print the peer id and proto id
-	if protoSpec.IsBeaconValidator || protoSpec.ShardID != shard.BeaconChainShardID {
-		fmt.Println("My peer id: ", host.ID().String())
-		fmt.Println("My proto id: ", pid)
-	}
-
 	return &streamManager{
 		myProtoID:       pid,
 		myProtoSpec:     protoSpec,
@@ -476,13 +469,12 @@ func (sm *streamManager) discoverAndSetupStream(discCtx context.Context) (int, e
 func (sm *streamManager) discover(ctx context.Context) (<-chan libp2p_peer.AddrInfo, error) {
 	numStreams := sm.streams.size()
 
-	protoID := sm.targetProtoID()
 	discBatch := sm.config.DiscBatch
 	if sm.config.HiCap-numStreams < sm.config.DiscBatch {
 		discBatch = sm.config.HiCap - numStreams
 	}
 	sm.logger.Debug().
-		Interface("protoID", protoID).
+		Interface("protoID", sm.myProtoID).
 		Int("numStreams", numStreams).
 		Int("discBatch", discBatch).
 		Msg("[StreamManager] discovering")
@@ -495,15 +487,7 @@ func (sm *streamManager) discover(ctx context.Context) (<-chan libp2p_peer.AddrI
 		<-time.After(discTimeout)
 		cancel()
 	}()
-	return sm.pf.FindPeers(ctx2, protoID, discBatch)
-}
-
-func (sm *streamManager) targetProtoID() string {
-	targetSpec := sm.myProtoSpec
-	if targetSpec.ShardID == shard.BeaconChainShardID { // for beacon chain, only connect to beacon nodes
-		targetSpec.IsBeaconValidator = true
-	}
-	return string(targetSpec.ToProtoID())
+	return sm.pf.FindPeers(ctx2, string(sm.myProtoID), discBatch)
 }
 
 func (sm *streamManager) setupStreamWithPeer(ctx context.Context, pid libp2p_peer.ID) error {
@@ -513,7 +497,7 @@ func (sm *streamManager) setupStreamWithPeer(ctx context.Context, pid libp2p_pee
 	nCtx, cancel := context.WithTimeout(ctx, connectTimeout)
 	defer cancel()
 
-	st, err := sm.host.NewStream(nCtx, pid, protocol.ID(sm.targetProtoID()))
+	st, err := sm.host.NewStream(nCtx, pid, protocol.ID(sm.myProtoID))
 	if err != nil {
 		return err
 	}

p2p/stream/common/streammanager/streammanager_test.go

@@ -264,7 +264,7 @@ func TestStreamSet_numStreamsWithMinProtoID(t *testing.T) {
 		pid1    = testProtoID
 		numPid1 = 5
 
-		pid2    = sttypes.ProtoID("harmony/sync/unitest/0/1.0.1/1")
+		pid2    = sttypes.ProtoID("harmony/sync/unitest/0/1.0.1")
 		numPid2 = 10
 	)
 

p2p/stream/protocols/sync/protocol.go

@@ -3,8 +3,8 @@ package sync
 import (
 	"context"
 	"errors"
+	"fmt"
 	"math/rand"
-	"strconv"
 	"time"
 
 	"github.com/ethereum/go-ethereum/event"
@@ -17,7 +17,6 @@ import (
 	"github.com/harmony-one/harmony/p2p/stream/common/requestmanager"
 	"github.com/harmony-one/harmony/p2p/stream/common/streammanager"
 	sttypes "github.com/harmony-one/harmony/p2p/stream/types"
-	"github.com/harmony-one/harmony/shard"
 	"github.com/hashicorp/go-version"
 	libp2p_host "github.com/libp2p/go-libp2p/core/host"
 	libp2p_network "github.com/libp2p/go-libp2p/core/network"
@@ -27,7 +26,8 @@ import (
 
 const (
 	// serviceSpecifier is the specifier for the service.
-	serviceSpecifier = "sync"
+	SyncServiceSpecifier      = "sync"
+	EpochSyncServiceSpecifier = "epochsync"
 )
 
 var (
@@ -71,6 +71,7 @@ type (
 		BeaconNode bool
 		Validator  bool
 		Explorer   bool
+		EpochChain bool
 
 		MaxAdvertiseWaitTime int
 		// stream manager config
@@ -106,6 +107,12 @@ func NewProtocol(config Config) *Protocol {
 
 	sp.rm = requestmanager.NewRequestManager(sp.sm)
 
+	// if it is not epoch chain, print the peer id and proto id
+	if !config.EpochChain {
+		fmt.Println("My peer id: ", config.Host.ID().String())
+		fmt.Println("My proto id: ", sp.ProtoID())
+	}
+
 	sp.logger = utils.Logger().With().Str("Protocol", string(sp.ProtoID())).Logger()
 	return sp
 }
@@ -115,10 +122,7 @@ func (p *Protocol) Start() {
 	p.sm.Start()
 	p.rm.Start()
 	p.rl.Start()
-	// If it's not EpochChain, advertise
-	if p.config.BeaconNode || p.chain.ShardID() != shard.BeaconChainShardID {
-		go p.advertiseLoop()
-	}
+	go p.advertiseLoop()
 }
 
 // Close close the protocol
@@ -130,19 +134,18 @@ func (p *Protocol) Close() {
 	close(p.closeC)
 }
 
-// Specifier return the specifier for the protocol
-func (p *Protocol) Specifier() string {
-	return serviceSpecifier + "/" + strconv.Itoa(int(p.config.ShardID))
-}
-
 // ProtoID return the ProtoID of the sync protocol
 func (p *Protocol) ProtoID() sttypes.ProtoID {
 	return p.protoIDByVersion(MyVersion)
 }
 
-// ShardProtoID returns the ProtoID of the sync protocol for shard nodes
-func (p *Protocol) ShardProtoID() sttypes.ProtoID {
-	return p.protoIDByVersionForShardNodes(MyVersion)
+// ServiceID returns the service ID of the sync protocol
+func (p *Protocol) ServiceID() string {
+	serviceID := SyncServiceSpecifier
+	if p.config.EpochChain {
+		serviceID = EpochSyncServiceSpecifier
+	}
+	return serviceID
 }
 
 // Version returns the sync protocol version
@@ -155,6 +158,11 @@ func (p *Protocol) IsBeaconValidator() bool {
 	return p.config.BeaconNode && p.config.Validator
 }
 
+// IsEpochChain returns true if it is a epoch chain
+func (p *Protocol) IsEpochChain() bool {
+	return p.config.EpochChain
+}
+
 // IsValidator returns true if it is a validator node
 func (p *Protocol) IsValidator() bool {
 	return p.config.Validator
@@ -171,7 +179,7 @@ func (p *Protocol) Match(targetID protocol.ID) bool {
 	if err != nil {
 		return false
 	}
-	if target.Service != serviceSpecifier {
+	if target.Service != p.ServiceID() {
 		return false
 	}
 	if target.NetworkType != p.config.Network {
@@ -347,11 +355,6 @@ func (p *Protocol) supportedProtoIDs() []sttypes.ProtoID {
 	pids := make([]sttypes.ProtoID, 0, len(vs))
 	for _, v := range vs {
 		pids = append(pids, p.protoIDByVersion(v))
-		// beacon node needs to inform shard nodes about it supports them as well for EpochChain
-		// basically beacon node can accept connection from shard nodes to share last epoch blocks
-		if p.IsBeaconValidator() {
-			pids = append(pids, p.protoIDByVersionForShardNodes(v))
-		}
 	}
 	return pids
 }
@@ -362,22 +365,10 @@ func (p *Protocol) supportedVersions() []*version.Version {
 
 func (p *Protocol) protoIDByVersion(v *version.Version) sttypes.ProtoID {
 	spec := sttypes.ProtoSpec{
-		Service:           serviceSpecifier,
-		NetworkType:       p.config.Network,
-		ShardID:           p.config.ShardID,
-		Version:           v,
-		IsBeaconValidator: (p.config.Validator || p.config.Explorer) && p.config.BeaconNode,
-	}
-	return spec.ToProtoID()
-}
-
-func (p *Protocol) protoIDByVersionForShardNodes(v *version.Version) sttypes.ProtoID {
-	spec := sttypes.ProtoSpec{
-		Service:           serviceSpecifier,
-		NetworkType:       p.config.Network,
-		ShardID:           p.config.ShardID,
-		Version:           v,
-		IsBeaconValidator: false,
+		Service:     p.ServiceID(),
+		NetworkType: p.config.Network,
+		ShardID:     p.config.ShardID,
+		Version:     v,
 	}
 	return spec.ToProtoID()
 }

p2p/stream/protocols/sync/protocol_test.go

@@ -16,13 +16,14 @@ func TestProtocol_Match(t *testing.T) {
 		targetID protocol.ID
 		exp      bool
 	}{
-		{"harmony/sync/unitest/0/1.0.1/1", true},
-		{"harmony/sync/unitest/0/1.0.1/0", true},
+		{"harmony/sync/unitest/0/1.0", true},
+		{"harmony/sync/unitest/1/1.0.1", false},
+		{"harmony/sync/unitest/0/1.2-alpha", true},
 		{"h123456", false},
-		{"harmony/sync/unitest/0/0.9.9/1", false},
-		{"harmony/epoch/unitest/0/1.0.1/1", false},
-		{"harmony/sync/mainnet/0/1.0.1/1", false},
-		{"harmony/sync/unitest/1/1.0.1/1", false},
+		{"harmony/sync/unitest/0/0.9.9", false},
+		{"harmony/epochsync/unitest/0/1.0", false},
+		{"harmony/sync/mainnet/0/1.0.1", false},
+		{"harmony/sync/unitest/1/1.0.1", false},
 	}
 
 	for i, test := range tests {

p2p/stream/types/interface.go

@@ -11,10 +11,9 @@ import (
 type Protocol interface {
 	p2ptypes.LifeCycle
 
-	Specifier() string
 	Version() *version.Version
 	ProtoID() ProtoID
-	// ShardProtoID() ProtoID
+	ServiceID() string
 	IsBeaconValidator() bool
 	Match(id protocol.ID) bool
 	HandleStream(st libp2p_network.Stream)