core: Run DKG stuffs only if the node is in DKG set (#302)

15 files changed, 165 insertions, 40 deletions

diff --git a/cmd/dexcon-simulation-with-scheduler/main.go b/cmd/dexcon-simulation-with-scheduler/main.go
index 6f3e320..8b7c252 100644
--- a/cmd/dexcon-simulation-with-scheduler/main.go
+++ b/cmd/dexcon-simulation-with-scheduler/main.go
@@ -1,3 +1,20 @@
+// Copyright 2018 The dexon-consensus Authors
+// This file is part of the dexon-consensus library.
+//
+// The dexon-consensus library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus library. If not, see
+// <http://www.gnu.org/licenses/>.
+
 package main
 
 import (
@@ -58,7 +75,7 @@ func main() {
 		Mean:  cfg.Node.Legacy.ProposeIntervalMean,
 	}
 	// Setup key pairs.
-	prvKeys, pubKeys, err := test.NewKeys(cfg.Node.Num)
+	prvKeys, pubKeys, err := test.NewKeys(int(cfg.Node.Num))
 	if err != nil {
 		log.Fatal("could not setup key pairs: ", err)
 	}
diff --git a/core/configuration-chain_test.go b/core/configuration-chain_test.go
index 6ddfeb9..11e5eca 100644
--- a/core/configuration-chain_test.go
+++ b/core/configuration-chain_test.go
@@ -233,7 +233,7 @@ func (s *ConfigurationChainTestSuite) preparePartialSignature(
 func (s *ConfigurationChainTestSuite) TestConfigurationChain() {
 	k := 4
 	n := 10
-	round := uint64(1)
+	round := uint64(0)
 	cfgChains := s.runDKG(k, n, round)
 
 	hash := crypto.Keccak256Hash([]byte("🌚🌝"))
@@ -272,7 +272,7 @@ func (s *ConfigurationChainTestSuite) TestConfigurationChain() {
 func (s *ConfigurationChainTestSuite) TestMultipleTSig() {
 	k := 2
 	n := 7
-	round := uint64(1)
+	round := uint64(0)
 	cfgChains := s.runDKG(k, n, round)
 
 	hash1 := crypto.Keccak256Hash([]byte("Hash1"))
diff --git a/core/consensus.go b/core/consensus.go
index 11d9799..6ff67bb 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -417,15 +417,23 @@ func (con *Consensus) Run(initBlock *types.Block) {
 			panic(err)
 		}
 	}
+	round0 := uint64(0)
+	dkgSet, err := con.nodeSetCache.GetDKGSet(round0)
+	if err != nil {
+		panic(err)
+	}
 	con.logger.Debug("Calling Network.ReceiveChan")
 	go con.processMsg(con.network.ReceiveChan())
 	// Sleep until dMoment come.
 	time.Sleep(con.dMoment.Sub(time.Now().UTC()))
-	con.cfgModule.registerDKG(con.round, int(con.currentConfig.DKGSetSize)/3+1)
-	con.event.RegisterTime(con.dMoment.Add(con.currentConfig.RoundInterval/4),
-		func(time.Time) {
-			con.runDKGTSIG(con.round)
-		})
+	if _, exist := dkgSet[con.ID]; exist {
+		con.logger.Info("Selected as DKG set", "round", round0)
+		con.cfgModule.registerDKG(round0, int(con.currentConfig.DKGSetSize)/3+1)
+		con.event.RegisterTime(con.dMoment.Add(con.currentConfig.RoundInterval/4),
+			func(time.Time) {
+				con.runDKGTSIG(round0)
+			})
+	}
 	round1 := uint64(1)
 	con.logger.Debug("Calling Governance.Configuration", "round", round1)
 	con.lattice.AppendConfig(round1, con.gov.Configuration(round1))
@@ -613,11 +621,23 @@ func (con *Consensus) initialRound(startTime time.Time) {
 	}
 	con.logger.Debug("Calling Governance.Configuration", "round", con.round)
 	con.currentConfig = con.gov.Configuration(con.round)
+	curDkgSet, err := con.nodeSetCache.GetDKGSet(con.round)
+	if err != nil {
+		con.logger.Error("Error getting DKG set", "round", con.round, "error", err)
+		curDkgSet = make(map[types.NodeID]struct{})
+	}
+	if _, exist := curDkgSet[con.ID]; exist {
+		con.event.RegisterTime(startTime.Add(con.currentConfig.RoundInterval/2),
+			func(time.Time) {
+				go func() {
+					con.runCRS()
+				}()
+			})
+	}
 
 	con.event.RegisterTime(startTime.Add(con.currentConfig.RoundInterval/2),
 		func(time.Time) {
 			go func() {
-				con.runCRS()
 				ticker := newTicker(con.gov, con.round, TickerDKG)
 				<-ticker.Tick()
 				// Normally, gov.CRS would return non-nil. Use this for in case of
@@ -627,19 +647,30 @@ func (con *Consensus) initialRound(startTime time.Time) {
 						"nodeID", con.ID)
 					time.Sleep(500 * time.Millisecond)
 				}
+				nextDkgSet, err := con.nodeSetCache.GetDKGSet(con.round + 1)
+				if err != nil {
+					con.logger.Error("Error getting DKG set",
+						"round", con.round+1, "error", err)
+					return
+				}
+				if _, exist := nextDkgSet[con.ID]; !exist {
+					return
+				}
+				con.logger.Info("Selected as DKG set", "round", con.round+1)
 				con.cfgModule.registerDKG(
 					con.round+1, int(con.currentConfig.DKGSetSize/3)+1)
+				con.event.RegisterTime(
+					startTime.Add(con.currentConfig.RoundInterval*2/3),
+					func(time.Time) {
+						func() {
+							con.dkgReady.L.Lock()
+							defer con.dkgReady.L.Unlock()
+							con.dkgRunning = 0
+						}()
+						con.runDKGTSIG(con.round + 1)
+					})
 			}()
 		})
-	con.event.RegisterTime(startTime.Add(con.currentConfig.RoundInterval*2/3),
-		func(time.Time) {
-			func() {
-				con.dkgReady.L.Lock()
-				defer con.dkgReady.L.Unlock()
-				con.dkgRunning = 0
-			}()
-			con.runDKGTSIG(con.round + 1)
-		})
 	con.event.RegisterTime(startTime.Add(con.currentConfig.RoundInterval),
 		func(time.Time) {
 			// Change round.
diff --git a/core/nodeset-cache.go b/core/nodeset-cache.go
index bf7b88d..26e3d55 100644
--- a/core/nodeset-cache.go
+++ b/core/nodeset-cache.go
@@ -177,6 +177,11 @@ func (cache *NodeSetCache) update(
 		err = ErrRoundNotReady
 		return
 	}
+	crs := cache.nsIntf.CRS(round)
+	if (crs == common.Hash{}) {
+		err = ErrRoundNotReady
+		return
+	}
 	// Cache new round.
 	nodeSet := types.NewNodeSet()
 	for _, key := range keySet {
@@ -192,7 +197,6 @@ func (cache *NodeSetCache) update(
 		}
 	}
 	cfg := cache.nsIntf.Configuration(round)
-	crs := cache.nsIntf.CRS(round)
 	nIDs = &sets{
 		nodeSet:   nodeSet,
 		notarySet: make([]map[types.NodeID]struct{}, cfg.NumChains),
diff --git a/core/test/fake-transport.go b/core/test/fake-transport.go
index 03acba5..a783ac9 100644
--- a/core/test/fake-transport.go
+++ b/core/test/fake-transport.go
@@ -1,6 +1,23 @@
 // Copyright 2018 The dexon-consensus Authors
 // This file is part of the dexon-consensus library.
 //
+// The dexon-consensus library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+// Copyright 2018 The dexon-consensus Authors
+// This file is part of the dexon-consensus library.
+//
 // The dexon-consensus library is free software: you can redistribute it and/or
 // modify it under the terms of the GNU Lesser General Public License as
 // published by the Free Software Foundation, either version 3 of the License,
@@ -161,7 +178,7 @@ func (t *FakeTransport) Host() (chan *TransportEnvelope, error) {
 }
 
 // WaitForPeers implements TransportServer.WaitForPeers method.
-func (t *FakeTransport) WaitForPeers(numPeers int) (err error) {
+func (t *FakeTransport) WaitForPeers(numPeers uint32) (err error) {
 	t.peers = make(map[types.NodeID]fakePeerRecord)
 	for {
 		envelope := <-t.recvChannel
@@ -172,7 +189,7 @@ func (t *FakeTransport) WaitForPeers(numPeers int) (err error) {
 			sendChannel: newPeer.recvChannel,
 			pubKey:      newPeer.pubKey,
 		}
-		if len(t.peers) == numPeers {
+		if uint32(len(t.peers)) == numPeers {
 			break
 		}
 	}
diff --git a/core/test/interface.go b/core/test/interface.go
index 9ff79db..e7a8f70 100644
--- a/core/test/interface.go
+++ b/core/test/interface.go
@@ -8,6 +8,23 @@
 //
 // The dexon-consensus library is distributed in the hope that it will be
 // useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+// General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the dexon-consensus library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+// Copyright 2018 The dexon-consensus Authors
+// This file is part of the dexon-consensus library.
+//
+// The dexon-consensus library is free software: you can redistribute it
+// and/or modify it under the terms of the GNU Lesser General Public License as
+// published by the Free Software Foundation, either version 3 of the License,
+// or (at your option) any later version.
+//
+// The dexon-consensus library is distributed in the hope that it will be
+// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 // GNU Lesser General Public License for more details.
 //
@@ -76,7 +93,7 @@ type TransportServer interface {
 	// returned channel could be used after 'WaitForPeers' returns.
 	Host() (chan *TransportEnvelope, error)
 	// WaitForPeers waits for all peers to join the network.
-	WaitForPeers(numPeers int) error
+	WaitForPeers(numPeers uint32) error
 }
 
 // TransportClient defines those peers in the network.
diff --git a/core/test/tcp-transport.go b/core/test/tcp-transport.go
index 285b630..a3b9aba 100644
--- a/core/test/tcp-transport.go
+++ b/core/test/tcp-transport.go
@@ -743,7 +743,7 @@ func (t *TCPTransportServer) Host() (chan *TransportEnvelope, error) {
 }
 
 // WaitForPeers implements TransportServer.WaitForPeers method.
-func (t *TCPTransportServer) WaitForPeers(numPeers int) (err error) {
+func (t *TCPTransportServer) WaitForPeers(numPeers uint32) (err error) {
 	// Collect peers info. Packets other than peer info is
 	// unexpected.
 	peersInfo := make(map[types.NodeID]string)
@@ -764,7 +764,7 @@ func (t *TCPTransportServer) WaitForPeers(numPeers int) (err error) {
 		}
 		peersInfo[msg.NodeID] = msg.Info
 		// Check if we already collect enought peers.
-		if len(peersInfo) == numPeers {
+		if uint32(len(peersInfo)) == numPeers {
 			break
 		}
 	}
@@ -790,7 +790,7 @@ func (t *TCPTransportServer) WaitForPeers(numPeers int) (err error) {
 			panic(fmt.Errorf("already report conn-ready message: %v", e))
 		}
 		readies[msg.NodeID] = struct{}{}
-		if len(readies) == numPeers {
+		if uint32(len(readies)) == numPeers {
 			break
 		}
 	}
diff --git a/core/test/transport_test.go b/core/test/transport_test.go
index d5032cd..d5c4260 100644
--- a/core/test/transport_test.go
+++ b/core/test/transport_test.go
@@ -223,7 +223,7 @@ func (s *TransportTestSuite) TestFake() {
 		}()
 	}
 	// Block here until we collect enough peers.
-	server.trans.WaitForPeers(peerCount)
+	server.trans.WaitForPeers(uint32(peerCount))
 	// Make sure all clients are ready.
 	wg.Wait()
 	s.baseTest(server, peers, 300*time.Millisecond)
@@ -270,7 +270,7 @@ func (s *TransportTestSuite) TestTCPLocal() {
 		}()
 	}
 	// Block here until we collect enough peers.
-	server.trans.WaitForPeers(peerCount)
+	server.trans.WaitForPeers(uint32(peerCount))
 	// Make sure all clients are ready.
 	wg.Wait()
 
diff --git a/core/types/nodeset.go b/core/types/nodeset.go
index 3222b3c..89dfef3 100644
--- a/core/types/nodeset.go
+++ b/core/types/nodeset.go
@@ -100,6 +100,9 @@ func (ns *NodeSet) Clone() *NodeSet {
 // GetSubSet returns the subset of given target.
 func (ns *NodeSet) GetSubSet(
 	size int, target SubSetTarget) map[NodeID]struct{} {
+	if size == 0 {
+		return make(map[NodeID]struct{})
+	}
 	h := rankHeap{}
 	idx := 0
 	for nID := range ns.IDs {
diff --git a/core/types/nodeset_test.go b/core/types/nodeset_test.go
index a879122..2780ed2 100644
--- a/core/types/nodeset_test.go
+++ b/core/types/nodeset_test.go
@@ -54,6 +54,17 @@ func (s *NodeSetTestSuite) TestGetSubSet() {
 	}
 }
 
+func (s *NodeSetTestSuite) TestGetSubSetZeroSize() {
+	total := 10
+	nodes := NewNodeSet()
+	for len(nodes.IDs) < total {
+		nodes.IDs[NodeID{common.NewRandomHash()}] = struct{}{}
+	}
+	// Passing nil should not crash.
+	emptySet := nodes.GetSubSet(0, nil)
+	s.Len(emptySet, 0)
+}
+
 func TestNodeSet(t *testing.T) {
 	suite.Run(t, new(NodeSetTestSuite))
 }
diff --git a/simulation/config/config.go b/simulation/config/config.go
index 5a548f1..023c4df 100644
--- a/simulation/config/config.go
+++ b/simulation/config/config.go
@@ -34,6 +34,8 @@ type Consensus struct {
 	LambdaBA      int    `toml:"lambda_ba"`
 	LambdaDKG     int    `toml:"lambda_dkg"`
 	RoundInterval int
+	NotarySetSize uint32
+	DKGSetSize    uint32 `toml:"dkg_set_size"`
 }
 
 // Legacy config.
@@ -46,7 +48,7 @@ type Legacy struct {
 type Node struct {
 	Consensus Consensus
 	Legacy    Legacy
-	Num       int
+	Num       uint32
 	MaxBlock  uint64
 }
 
@@ -92,6 +94,8 @@ func GenerateDefault(path string) error {
 				LambdaBA:      250,
 				LambdaDKG:     1000,
 				RoundInterval: 30 * 1000,
+				NotarySetSize: 7,
+				DKGSetSize:    7,
 			},
 			Legacy: Legacy{
 				ProposeIntervalMean:  500,
diff --git a/simulation/governance.go b/simulation/governance.go
index 3b48248..4fdaadb 100644
--- a/simulation/governance.go
+++ b/simulation/governance.go
@@ -36,7 +36,9 @@ type simGovernance struct {
 	id                 types.NodeID
 	lock               sync.RWMutex
 	nodeSet            map[types.NodeID]crypto.PublicKey
-	expectedNumNodes   int
+	expectedNumNodes   uint32
+	notarySetSize      uint32
+	dkgSetSize         uint32
 	k                  int
 	phiRatio           float32
 	chainNum           uint32
@@ -54,12 +56,17 @@ type simGovernance struct {
 // newSimGovernance returns a new simGovernance instance.
 func newSimGovernance(
 	id types.NodeID,
-	numNodes int, consensusConfig config.Consensus) *simGovernance {
+	numNodes uint32,
+	notarySetSize uint32,
+	dkgSetSize uint32,
+	consensusConfig config.Consensus) *simGovernance {
 	hashCRS := crypto.Keccak256Hash([]byte(consensusConfig.GenesisCRS))
 	return &simGovernance{
 		id:                 id,
 		nodeSet:            make(map[types.NodeID]crypto.PublicKey),
 		expectedNumNodes:   numNodes,
+		notarySetSize:      notarySetSize,
+		dkgSetSize:         dkgSetSize,
 		k:                  consensusConfig.K,
 		phiRatio:           consensusConfig.PhiRatio,
 		chainNum:           consensusConfig.ChainNum,
@@ -100,8 +107,8 @@ func (g *simGovernance) Configuration(round uint64) *types.Config {
 		LambdaDKG:        g.lambdaDKG,
 		K:                g.k,
 		PhiRatio:         g.phiRatio,
-		NotarySetSize:    uint32(len(g.nodeSet)),
-		DKGSetSize:       uint32(len(g.nodeSet)),
+		NotarySetSize:    g.notarySetSize,
+		DKGSetSize:       g.dkgSetSize,
 		MinBlockInterval: g.lambdaBA * 3,
 		RoundInterval:    g.roundInterval,
 	}
@@ -139,7 +146,7 @@ func (g *simGovernance) addNode(pubKey crypto.PublicKey) {
 	if _, exists := g.nodeSet[nID]; exists {
 		return
 	}
-	if len(g.nodeSet) == g.expectedNumNodes {
+	if uint32(len(g.nodeSet)) == g.expectedNumNodes {
 		panic(fmt.Errorf("attempt to add node when ready"))
 	}
 	g.nodeSet[nID] = pubKey
diff --git a/simulation/kubernetes/config.toml.in b/simulation/kubernetes/config.toml.in
index f34ee76..7f02f74 100644
--- a/simulation/kubernetes/config.toml.in
+++ b/simulation/kubernetes/config.toml.in
@@ -2,18 +2,18 @@ title = "DEXON Consensus Simulation Config"
 
 [node]
 num = {{numNode}}
-lambda_ba = 250
-lambda_dkg = 1000
-max_block = 1000
+max_block = 18446744073709551615
 
 [node.consensus]
-phi_ratio = 6.66670024394989e-01
+phi_ratio = 6.666666865348816e-01
 k = 1
 chain_num = 7
 genesis_crs = "In DEXON we trust."
 lambda_ba = 250
-lambda_dkg = 1000
+lambda_dkg = 4000
 round_interval = 31536000000
+notary_set_size = 7
+dkg_set_size = 7
 
 [node.legacy]
 propose_interval_mean = 5e+02
diff --git a/simulation/node.go b/simulation/node.go
index 8907d5a..56c5832 100644
--- a/simulation/node.go
+++ b/simulation/node.go
@@ -91,7 +91,12 @@ func newNode(
 	if err != nil {
 		panic(err)
 	}
-	gov := newSimGovernance(id, config.Node.Num, config.Node.Consensus)
+	gov := newSimGovernance(
+		id,
+		config.Node.Num,
+		config.Node.Consensus.NotarySetSize,
+		config.Node.Consensus.DKGSetSize,
+		config.Node.Consensus)
 	return &node{
 		ID:        id,
 		prvKey:    prvKey,
diff --git a/simulation/simulation.go b/simulation/simulation.go
index 801bb7e..4e97900 100644
--- a/simulation/simulation.go
+++ b/simulation/simulation.go
@@ -18,6 +18,7 @@
 package simulation
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -35,6 +36,14 @@ func Run(cfg *config.Config) {
 		err         error
 	)
 
+	if cfg.Node.Consensus.NotarySetSize > cfg.Node.Num {
+		panic(fmt.Errorf("NotarySetSize should not be larger the node num"))
+	}
+
+	if cfg.Node.Consensus.DKGSetSize > cfg.Node.Num {
+		panic(fmt.Errorf("DKGSetSze should not be larger the node num"))
+	}
+
 	dMoment := time.Now().UTC().Add(1 * time.Second)
 
 	// init is a function to init a node.
@@ -69,7 +78,7 @@ func Run(cfg *config.Config) {
 			server.Run()
 		}()
 		// Initialize all nodes.
-		for i := 0; i < cfg.Node.Num; i++ {
+		for i := uint32(0); i < cfg.Node.Num; i++ {
 			init(serverEndpoint)
 		}
 	}