Merge pull request #681 from harmony-one/rj_branch

Separate functionality and genesis block between beacon and shard chain
pull/692/head
Rongjian Lan 6 years ago committed by GitHub
commit 25dbb3fa58
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 3
      api/service/syncing/downloader/server.go
  2. 14
      cmd/harmony/main.go
  3. 7
      consensus/consensus_leader.go
  4. 19
      node/node.go
  5. 6
      node/node_genesis.go
  6. 4
      node/node_handler.go
  7. 3
      node/node_newblock.go
  8. 4
      node/node_syncing.go
  9. 2
      node/service_setup.go
  10. 4
      test/deploy.sh

@ -17,6 +17,7 @@ const (
// Server is the Server struct for downloader package. // Server is the Server struct for downloader package.
type Server struct { type Server struct {
downloadInterface DownloadInterface downloadInterface DownloadInterface
GrpcServer *grpc.Server
} }
// Query returns the feature at the given point. // Query returns the feature at the given point.
@ -39,6 +40,8 @@ func (s *Server) Start(ip, port string) (*grpc.Server, error) {
grpcServer := grpc.NewServer(opts...) grpcServer := grpc.NewServer(opts...)
pb.RegisterDownloaderServer(grpcServer, s) pb.RegisterDownloaderServer(grpcServer, s)
go grpcServer.Serve(lis) go grpcServer.Serve(lis)
s.GrpcServer = grpcServer
return grpcServer, nil return grpcServer, nil
} }

@ -167,7 +167,7 @@ func createGlobalConfig() *nodeconfig.ConfigType {
if nodeConfig.MainDB, err = InitLDBDatabase(*ip, *port, *freshDB, false); err != nil { if nodeConfig.MainDB, err = InitLDBDatabase(*ip, *port, *freshDB, false); err != nil {
panic(err) panic(err)
} }
if !*isGenesis { if shardID != 0 {
if nodeConfig.BeaconDB, err = InitLDBDatabase(*ip, *port, *freshDB, true); err != nil { if nodeConfig.BeaconDB, err = InitLDBDatabase(*ip, *port, *freshDB, true); err != nil {
panic(err) panic(err)
} }
@ -238,8 +238,6 @@ func setUpConsensusAndNode(nodeConfig *nodeconfig.ConfigType) (*consensus.Consen
currentNode.NodeConfig.SetShardGroupID(p2p.NewGroupIDByShardID(p2p.ShardID(nodeConfig.ShardID))) currentNode.NodeConfig.SetShardGroupID(p2p.NewGroupIDByShardID(p2p.ShardID(nodeConfig.ShardID)))
} }
} else { } else {
currentNode.AddBeaconChainDatabase(nodeConfig.BeaconDB)
if *isNewNode { if *isNewNode {
currentNode.NodeConfig.SetRole(nodeconfig.NewNode) currentNode.NodeConfig.SetRole(nodeconfig.NewNode)
// TODO: fix the roles as it's unknown before resharding. // TODO: fix the roles as it's unknown before resharding.
@ -262,6 +260,12 @@ func setUpConsensusAndNode(nodeConfig *nodeconfig.ConfigType) (*consensus.Consen
currentNode.Consensus.RegisterRndChannel(dRand.RndChannel) currentNode.Consensus.RegisterRndChannel(dRand.RndChannel)
currentNode.DRand = dRand currentNode.DRand = dRand
if consensus.ShardID != 0 {
currentNode.AddBeaconChainDatabase(nodeConfig.BeaconDB)
}
// This needs to be executed after consensus and drand are setup
currentNode.InitGenesisShardState()
// Assign closure functions to the consensus object // Assign closure functions to the consensus object
consensus.BlockVerifier = currentNode.VerifyNewBlock consensus.BlockVerifier = currentNode.VerifyNewBlock
consensus.OnConsensusDone = currentNode.PostConsensusProcessing consensus.OnConsensusDone = currentNode.PostConsensusProcessing
@ -293,6 +297,10 @@ func main() {
if consensus.IsLeader { if consensus.IsLeader {
go currentNode.SendPongMessage() go currentNode.SendPongMessage()
} }
// TODO: enable beacon chain sync
//if consensus.ShardID != 0 {
// go currentNode.SupportBeaconSyncing()
//}
go currentNode.SupportSyncing() go currentNode.SupportSyncing()
utils.GetLogInstance().Info("New Harmony Node ====", "Role", currentNode.NodeConfig.Role(), "multiaddress", fmt.Sprintf("/ip4/%s/tcp/%s/p2p/%s", *ip, *port, nodeConfig.Host.GetID().Pretty())) utils.GetLogInstance().Info("New Harmony Node ====", "Role", currentNode.NodeConfig.Role(), "multiaddress", fmt.Sprintf("/ip4/%s/tcp/%s/p2p/%s", *ip, *port, nodeConfig.Host.GetID().Pretty()))
currentNode.ServiceManagerSetup() currentNode.ServiceManagerSetup()

@ -55,7 +55,8 @@ func (consensus *Consensus) WaitForNewBlock(blockChannel chan *types.Block, stop
utils.GetLogInstance().Debug("WaitForNewBlock", "removed peers", c) utils.GetLogInstance().Debug("WaitForNewBlock", "removed peers", c)
} }
if core.IsEpochBlock(newBlock) { if consensus.ShardID == 0 {
if core.IsEpochBlock(newBlock) { // Only beacon chain do randomness generation
// Receive pRnd from DRG protocol // Receive pRnd from DRG protocol
utils.GetLogInstance().Debug("[DRG] Waiting for pRnd") utils.GetLogInstance().Debug("[DRG] Waiting for pRnd")
pRndAndBitmap := <-consensus.PRndChannel pRndAndBitmap := <-consensus.PRndChannel
@ -69,6 +70,7 @@ func (consensus *Consensus) WaitForNewBlock(blockChannel chan *types.Block, stop
// TODO: check validity of pRnd // TODO: check validity of pRnd
newBlock.AddRandPreimage(pRnd) newBlock.AddRandPreimage(pRnd)
} }
rnd, blockHash, err := consensus.GetNextRnd() rnd, blockHash, err := consensus.GetNextRnd()
if err == nil { if err == nil {
// Verify the randomness // Verify the randomness
@ -78,6 +80,7 @@ func (consensus *Consensus) WaitForNewBlock(blockChannel chan *types.Block, stop
} else { } else {
utils.GetLogInstance().Info("Failed to get randomness", "error", err) utils.GetLogInstance().Info("Failed to get randomness", "error", err)
} }
}
startTime = time.Now() startTime = time.Now()
utils.GetLogInstance().Debug("STARTING CONSENSUS", "numTxs", len(newBlock.Transactions()), "consensus", consensus, "startTime", startTime, "publicKeys", len(consensus.PublicKeys)) utils.GetLogInstance().Debug("STARTING CONSENSUS", "numTxs", len(newBlock.Transactions()), "consensus", consensus, "startTime", startTime, "publicKeys", len(consensus.PublicKeys))
for { // Wait until last consensus is finished for { // Wait until last consensus is finished
@ -312,7 +315,7 @@ func (consensus *Consensus) processCommitMessage(message *msg_pb.Message) {
select { select {
case consensus.VerifiedNewBlock <- &blockObj: case consensus.VerifiedNewBlock <- &blockObj:
default: default:
utils.GetLogInstance().Info("[SYNC] consensus verified block send to chan failed", "blockHash", blockObj.Hash()) utils.GetLogInstance().Info("[SYNC] Failed to send consensus verified block for state sync", "blockHash", blockObj.Hash())
} }
consensus.reportMetrics(blockObj) consensus.reportMetrics(blockObj)

@ -243,13 +243,13 @@ func New(host p2p.Host, consensusObj *consensus.Consensus, db ethdb.Database, is
database := db database := db
if database == nil { if database == nil {
database = ethdb.NewMemDatabase() database = ethdb.NewMemDatabase()
chain, err = node.GenesisBlockSetup(database, false) chain, err = node.GenesisBlockSetup(database, consensusObj.ShardID, false)
isFirstTime = true isFirstTime = true
} else { } else {
chain, err = node.InitBlockChainFromDB(db, node.Consensus, isArchival) chain, err = node.InitBlockChainFromDB(db, node.Consensus, isArchival)
isFirstTime = false isFirstTime = false
if err != nil || chain == nil || chain.CurrentBlock().NumberU64() <= 0 { if err != nil || chain == nil || chain.CurrentBlock().NumberU64() <= 0 {
chain, err = node.GenesisBlockSetup(database, isArchival) chain, err = node.GenesisBlockSetup(database, consensusObj.ShardID, isArchival)
isFirstTime = true isFirstTime = true
} }
} }
@ -267,6 +267,8 @@ func New(host p2p.Host, consensusObj *consensus.Consensus, db ethdb.Database, is
node.Consensus.VerifiedNewBlock = make(chan *types.Block) node.Consensus.VerifiedNewBlock = make(chan *types.Block)
if node.Consensus.ShardID == 0 {
// Contracts only exist in beacon chain
if isFirstTime { if isFirstTime {
// Setup one time smart contracts // Setup one time smart contracts
node.AddFaucetContractToPendingTransactions() node.AddFaucetContractToPendingTransactions()
@ -278,7 +280,7 @@ func New(host p2p.Host, consensusObj *consensus.Consensus, db ethdb.Database, is
} else { } else {
node.AddContractKeyAndAddress() node.AddContractKeyAndAddress()
} }
}
} }
node.ContractCaller = contracts.NewContractCaller(&db, node.blockchain, params.TestChainConfig) node.ContractCaller = contracts.NewContractCaller(&db, node.blockchain, params.TestChainConfig)
@ -315,7 +317,14 @@ func New(host p2p.Host, consensusObj *consensus.Consensus, db ethdb.Database, is
// InitGenesisShardState initialize genesis shard state and update committee pub keys for consensus and drand // InitGenesisShardState initialize genesis shard state and update committee pub keys for consensus and drand
func (node *Node) InitGenesisShardState() { func (node *Node) InitGenesisShardState() {
// Store the genesis shard state into db. // Store the genesis shard state into db.
shardState := node.blockchain.StoreNewShardState(node.blockchain.CurrentBlock(), nil) shardState := types.ShardState{}
if node.Consensus != nil {
if node.Consensus.ShardID == 0 {
shardState = node.blockchain.StoreNewShardState(node.blockchain.CurrentBlock(), nil)
} else {
shardState = node.beaconChain.StoreNewShardState(node.beaconChain.CurrentBlock(), nil)
}
}
// Update validator public keys // Update validator public keys
for _, shard := range shardState { for _, shard := range shardState {
if shard.ShardID == node.Consensus.ShardID { if shard.ShardID == node.Consensus.ShardID {
@ -428,7 +437,7 @@ func (node *Node) AddBeaconChainDatabase(db ethdb.Database) {
database = ethdb.NewMemDatabase() database = ethdb.NewMemDatabase()
} }
// TODO (chao) currently we use the same genesis block as normal shard // TODO (chao) currently we use the same genesis block as normal shard
chain, err := node.GenesisBlockSetup(database, false) chain, err := node.GenesisBlockSetup(database, 0, false)
if err != nil { if err != nil {
utils.GetLogInstance().Error("Error when doing genesis setup") utils.GetLogInstance().Error("Error when doing genesis setup")
os.Exit(1) os.Exit(1)

@ -26,7 +26,7 @@ const (
) )
// GenesisBlockSetup setups a genesis blockchain. // GenesisBlockSetup setups a genesis blockchain.
func (node *Node) GenesisBlockSetup(db ethdb.Database, isArchival bool) (*core.BlockChain, error) { func (node *Node) GenesisBlockSetup(db ethdb.Database, shardID uint32, isArchival bool) (*core.BlockChain, error) {
// Initialize genesis block and blockchain // Initialize genesis block and blockchain
// Tests account for txgen to use // Tests account for txgen to use
genesisAlloc := node.CreateGenesisAllocWithTestingAddresses(FakeAddressNumber) genesisAlloc := node.CreateGenesisAllocWithTestingAddresses(FakeAddressNumber)
@ -39,11 +39,13 @@ func (node *Node) GenesisBlockSetup(db ethdb.Database, isArchival bool) (*core.B
genesisAlloc[contractDeployerAddress] = core.GenesisAccount{Balance: contractDeployerFunds} genesisAlloc[contractDeployerAddress] = core.GenesisAccount{Balance: contractDeployerFunds}
node.ContractDeployerKey = contractDeployerKey node.ContractDeployerKey = contractDeployerKey
if shardID == 0 {
// Accounts used by validator/nodes to stake and participate in the network. // Accounts used by validator/nodes to stake and participate in the network.
AddNodeAddressesToGenesisAlloc(genesisAlloc) AddNodeAddressesToGenesisAlloc(genesisAlloc)
}
chainConfig := params.TestChainConfig chainConfig := params.TestChainConfig
chainConfig.ChainID = big.NewInt(int64(node.Consensus.ShardID)) // Use ChainID as piggybacked ShardID chainConfig.ChainID = big.NewInt(int64(shardID)) // Use ChainID as piggybacked ShardID
gspec := core.Genesis{ gspec := core.Genesis{
Config: chainConfig, Config: chainConfig,
Alloc: genesisAlloc, Alloc: genesisAlloc,

@ -288,7 +288,8 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block) {
node.AddNewBlock(newBlock) node.AddNewBlock(newBlock)
// TODO: enable drand only for beacon chain if node.Consensus.ShardID == 0 {
// ConfirmedBlockChannel which is listened by drand leader who will initiate DRG if its a epoch block (first block of a epoch) // ConfirmedBlockChannel which is listened by drand leader who will initiate DRG if its a epoch block (first block of a epoch)
if node.DRand != nil { if node.DRand != nil {
go func() { go func() {
@ -337,6 +338,7 @@ func (node *Node) PostConsensusProcessing(newBlock *types.Block) {
} }
} }
} }
}
} }
// AddNewBlock is usedd to add new block into the blockchain. // AddNewBlock is usedd to add new block into the blockchain.

@ -60,9 +60,10 @@ func (node *Node) WaitForConsensusReady(readySignal chan struct{}, stopChan chan
if err != nil { if err != nil {
utils.GetLogInstance().Debug("Failed commiting new block", "Error", err) utils.GetLogInstance().Debug("Failed commiting new block", "Error", err)
} else { } else {
if node.Consensus.ShardID == 0 {
// add new shard state if it's epoch block // add new shard state if it's epoch block
// TODO(minhdoan): only happens for beaconchain
node.addNewShardStateHash(block) node.addNewShardStateHash(block)
}
newBlock = block newBlock = block
utils.GetLogInstance().Debug("Successfully proposed new block", "blockNum", block.NumberU64(), "numTxs", block.Transactions().Len()) utils.GetLogInstance().Debug("Successfully proposed new block", "blockNum", block.NumberU64(), "numTxs", block.Transactions().Len())
break break

@ -111,13 +111,17 @@ func (node *Node) SupportSyncing() {
// InitSyncingServer starts downloader server. // InitSyncingServer starts downloader server.
func (node *Node) InitSyncingServer() { func (node *Node) InitSyncingServer() {
if node.downloaderServer == nil {
node.downloaderServer = downloader.NewServer(node) node.downloaderServer = downloader.NewServer(node)
}
} }
// StartSyncingServer starts syncing server. // StartSyncingServer starts syncing server.
func (node *Node) StartSyncingServer() { func (node *Node) StartSyncingServer() {
utils.GetLogInstance().Info("support_sycning: StartSyncingServer") utils.GetLogInstance().Info("support_sycning: StartSyncingServer")
if node.downloaderServer.GrpcServer == nil {
node.downloaderServer.Start(node.SelfPeer.IP, syncing.GetSyncingPort(node.SelfPeer.Port)) node.downloaderServer.Start(node.SelfPeer.IP, syncing.GetSyncingPort(node.SelfPeer.Port))
}
} }
// SendNewBlockToUnsync send latest verified block to unsync, registered nodes // SendNewBlockToUnsync send latest verified block to unsync, registered nodes

@ -33,8 +33,6 @@ func (node *Node) setupForShardLeader() {
node.serviceManager.RegisterService(service.BlockProposal, blockproposal.New(node.Consensus.ReadySignal, node.WaitForConsensusReady)) node.serviceManager.RegisterService(service.BlockProposal, blockproposal.New(node.Consensus.ReadySignal, node.WaitForConsensusReady))
// Register client support service. // Register client support service.
node.serviceManager.RegisterService(service.ClientSupport, clientsupport.New(node.blockchain.State, node.CallFaucetContract, node.getDeployedStakingContract, node.SelfPeer.IP, node.SelfPeer.Port)) node.serviceManager.RegisterService(service.ClientSupport, clientsupport.New(node.blockchain.State, node.CallFaucetContract, node.getDeployedStakingContract, node.SelfPeer.IP, node.SelfPeer.Port))
// Register randomness service
node.serviceManager.RegisterService(service.Randomness, randomness.New(node.DRand))
} }
func (node *Node) setupForShardValidator() { func (node *Node) setupForShardValidator() {

@ -70,7 +70,7 @@ EOU
DB= DB=
TXGEN=true TXGEN=true
DURATION=90 DURATION=60
MIN=5 MIN=5
SHARDS=2 SHARDS=2
KILLPORT=9004 KILLPORT=9004
@ -99,7 +99,7 @@ if [ -z "$config" ]; then
fi fi
if [ "$SYNC" == "true" ]; then if [ "$SYNC" == "true" ]; then
DURATION=300 DURATION=120
fi fi
# Kill nodes if any # Kill nodes if any

Loading…
Cancel
Save