simulation: support config change (#381)

7 files changed, 257 insertions, 70 deletions

diff --git a/simulation/config/config.go b/simulation/config/config.go
index 023c4df..797145c 100644
--- a/simulation/config/config.go
+++ b/simulation/config/config.go
@@ -18,24 +18,27 @@
 package config
 
 import (
+	"fmt"
 	"math"
 	"os"
 
+	"github.com/dexon-foundation/dexon-consensus/core"
 	"github.com/dexon-foundation/dexon-consensus/core/test"
 	"github.com/naoina/toml"
 )
 
 // Consensus settings.
 type Consensus struct {
-	PhiRatio      float32
-	K             int
-	ChainNum      uint32
-	GenesisCRS    string `toml:"genesis_crs"`
-	LambdaBA      int    `toml:"lambda_ba"`
-	LambdaDKG     int    `toml:"lambda_dkg"`
-	RoundInterval int
-	NotarySetSize uint32
-	DKGSetSize    uint32 `toml:"dkg_set_size"`
+	PhiRatio         float32
+	K                int
+	NumChains        uint32
+	GenesisCRS       string `toml:"genesis_crs"`
+	LambdaBA         int    `toml:"lambda_ba"`
+	LambdaDKG        int    `toml:"lambda_dkg"`
+	RoundInterval    int
+	NotarySetSize    uint32
+	DKGSetSize       uint32 `toml:"dkg_set_size"`
+	MinBlockInterval int
 }
 
 // Legacy config.
@@ -50,6 +53,7 @@ type Node struct {
 	Legacy    Legacy
 	Num       uint32
 	MaxBlock  uint64
+	Changes   []Change
 }
 
 // Networking config.
@@ -67,6 +71,25 @@ type Scheduler struct {
 	WorkerNum int
 }
 
+// Change represent future configuration changes.
+type Change struct {
+	Round uint64
+	Type  string
+	Value string
+}
+
+// RegisterChange reigster this change to a test.Governance instance.
+func (c Change) RegisterChange(gov *test.Governance) error {
+	if c.Round < core.ConfigRoundShift {
+		panic(fmt.Errorf(
+			"attempt to register config change that never be executed: %v",
+			c.Round))
+	}
+	t := StateChangeTypeFromString(c.Type)
+	return gov.RegisterConfigChange(
+		c.Round, t, StateChangeValueFromString(t, c.Value))
+}
+
 // Config represents the configuration for simulation.
 type Config struct {
 	Title      string
@@ -87,15 +110,16 @@ func GenerateDefault(path string) error {
 		Title: "DEXON Consensus Simulation Config",
 		Node: Node{
 			Consensus: Consensus{
-				PhiRatio:      float32(2) / 3,
-				K:             1,
-				ChainNum:      7,
-				GenesisCRS:    "In DEXON we trust.",
-				LambdaBA:      250,
-				LambdaDKG:     1000,
-				RoundInterval: 30 * 1000,
-				NotarySetSize: 7,
-				DKGSetSize:    7,
+				PhiRatio:         float32(2) / 3,
+				K:                1,
+				NumChains:        7,
+				GenesisCRS:       "In DEXON we trust.",
+				LambdaBA:         250,
+				LambdaDKG:        1000,
+				RoundInterval:    30 * 1000,
+				NotarySetSize:    7,
+				DKGSetSize:       7,
+				MinBlockInterval: 750,
 			},
 			Legacy: Legacy{
 				ProposeIntervalMean:  500,
diff --git a/simulation/config/utils.go b/simulation/config/utils.go
new file mode 100644
index 0000000..9d97fbd
--- /dev/null
+++ b/simulation/config/utils.go
@@ -0,0 +1,80 @@
+// 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 config
+
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/dexon-foundation/dexon-consensus/core/test"
+)
+
+// StateChangeTypeFromString convert a string to test.StateChangeType.
+func StateChangeTypeFromString(s string) test.StateChangeType {
+	switch s {
+	case "num_chains":
+		return test.StateChangeNumChains
+	case "lambda_ba":
+		return test.StateChangeLambdaBA
+	case "lambda_dkg":
+		return test.StateChangeLambdaDKG
+	case "round_interval":
+		return test.StateChangeRoundInterval
+	case "min_block_interval":
+		return test.StateChangeMinBlockInterval
+	case "k":
+		return test.StateChangeK
+	case "phi_ratio":
+		return test.StateChangePhiRatio
+	case "notary_set_size":
+		return test.StateChangeNotarySetSize
+	case "dkg_set_size":
+		return test.StateChangeDKGSetSize
+	}
+	panic(fmt.Errorf("unsupported state change type %s", s))
+}
+
+// StateChangeValueFromString converts a string to a value for state change
+// request.
+func StateChangeValueFromString(
+	t test.StateChangeType, v string) interface{} {
+	switch t {
+	case test.StateChangeNumChains, test.StateChangeNotarySetSize,
+		test.StateChangeDKGSetSize:
+		ret, err := strconv.ParseUint(v, 10, 32)
+		if err != nil {
+			panic(err)
+		}
+		return uint32(ret)
+	case test.StateChangeLambdaBA, test.StateChangeLambdaDKG,
+		test.StateChangeRoundInterval, test.StateChangeMinBlockInterval,
+		test.StateChangeK:
+		ret, err := strconv.ParseInt(v, 10, 32)
+		if err != nil {
+			panic(err)
+		}
+		return int(ret)
+	case test.StateChangePhiRatio:
+		ret, err := strconv.ParseFloat(v, 32)
+		if err != nil {
+			panic(err)
+		}
+		return float32(ret)
+	}
+	panic(fmt.Errorf("unsupported state change type %s", t))
+}
diff --git a/simulation/marshaller.go b/simulation/marshaller.go
index 86eab3e..6f8aee4 100644
--- a/simulation/marshaller.go
+++ b/simulation/marshaller.go
@@ -30,12 +30,12 @@ type jsonMarshaller struct{}
 func (m *jsonMarshaller) Unmarshal(
 	msgType string, payload []byte) (msg interface{}, err error) {
 	switch msgType {
-	case "info-status":
-		var status infoStatus
-		if err = json.Unmarshal(payload, &status); err != nil {
+	case "server-notif":
+		var notif serverNotification
+		if err = json.Unmarshal(payload, &notif); err != nil {
 			break
 		}
-		msg = status
+		msg = notif
 	case "blocklist":
 		var blocks BlockList
 		if err = json.Unmarshal(payload, &blocks); err != nil {
@@ -61,8 +61,8 @@ func (m *jsonMarshaller) Unmarshal(
 func (m *jsonMarshaller) Marshal(msg interface{}) (
 	msgType string, payload []byte, err error) {
 	switch msg.(type) {
-	case infoStatus:
-		msgType = "info-status"
+	case serverNotification:
+		msgType = "server-notif"
 	case *BlockList:
 		msgType = "blocklist"
 	case *message:
diff --git a/simulation/node.go b/simulation/node.go
index e766da8..dc4a725 100644
--- a/simulation/node.go
+++ b/simulation/node.go
@@ -31,17 +31,18 @@ import (
 	"github.com/dexon-foundation/dexon-consensus/simulation/config"
 )
 
-type infoStatus string
+type serverNotification string
 
 const (
-	statusInit     infoStatus = "init"
-	statusNormal   infoStatus = "normal"
-	statusShutdown infoStatus = "shutdown"
+	ntfShutdown         serverNotification = "shutdown"
+	ntfSelectedAsMaster serverNotification = "as_master"
+	ntfReady            serverNotification = "ready"
 )
 
 type messageType string
 
 const (
+	setupOK        messageType = "setupOK"
 	shutdownAck    messageType = "shutdownAck"
 	blockTimestamp messageType = "blockTimestamps"
 )
@@ -60,24 +61,25 @@ type node struct {
 	netModule *test.Network
 	ID        types.NodeID
 	prvKey    crypto.PrivateKey
+	logger    common.Logger
 	consensus *core.Consensus
+	cfg       *config.Config
 }
 
 // newNode returns a new empty node.
-func newNode(
-	prvKey crypto.PrivateKey,
-	config config.Config) *node {
+func newNode(prvKey crypto.PrivateKey, logger common.Logger,
+	cfg config.Config) *node {
 	pubKey := prvKey.PublicKey()
 	netModule := test.NewNetwork(
 		pubKey,
 		&test.NormalLatencyModel{
-			Mean:  config.Networking.Mean,
-			Sigma: config.Networking.Sigma,
+			Mean:  cfg.Networking.Mean,
+			Sigma: cfg.Networking.Sigma,
 		},
 		test.NewDefaultMarshaller(&jsonMarshaller{}),
 		test.NetworkConfig{
-			Type:       config.Networking.Type,
-			PeerServer: config.Networking.PeerServer,
+			Type:       cfg.Networking.Type,
+			PeerServer: cfg.Networking.PeerServer,
 			PeerPort:   peerPort,
 		})
 	id := types.NewNodeID(pubKey)
@@ -86,40 +88,22 @@ func newNode(
 		panic(err)
 	}
 	// Sync config to state in governance.
-	cConfig := config.Node.Consensus
 	gov, err := test.NewGovernance(
 		test.NewState(
-			[]crypto.PublicKey{pubKey},
-			time.Millisecond,
-			&common.NullLogger{},
-			true),
+			[]crypto.PublicKey{pubKey}, time.Millisecond, logger, true),
 		core.ConfigRoundShift)
 	if err != nil {
 		panic(err)
 	}
-	gov.State().RequestChange(test.StateChangeK, cConfig.K)
-	gov.State().RequestChange(test.StateChangePhiRatio, cConfig.PhiRatio)
-	gov.State().RequestChange(test.StateChangeNumChains, cConfig.ChainNum)
-	gov.State().RequestChange(
-		test.StateChangeNotarySetSize, cConfig.NotarySetSize)
-	gov.State().RequestChange(test.StateChangeDKGSetSize, cConfig.DKGSetSize)
-	gov.State().RequestChange(test.StateChangeLambdaBA, time.Duration(
-		cConfig.LambdaBA)*time.Millisecond)
-	gov.State().RequestChange(test.StateChangeLambdaDKG, time.Duration(
-		cConfig.LambdaDKG)*time.Millisecond)
-	gov.State().RequestChange(test.StateChangeRoundInterval, time.Duration(
-		cConfig.RoundInterval)*time.Millisecond)
-	gov.State().RequestChange(
-		test.StateChangeMinBlockInterval,
-		3*time.Duration(cConfig.LambdaBA)*time.Millisecond)
-	gov.State().ProposeCRS(0, crypto.Keccak256Hash([]byte(cConfig.GenesisCRS)))
 	return &node{
 		ID:        id,
 		prvKey:    prvKey,
+		logger:    logger,
 		app:       newSimApp(id, netModule, gov.State()),
 		gov:       gov,
 		db:        dbInst,
 		netModule: netModule,
+		cfg:       &cfg,
 	}
 }
 
@@ -130,7 +114,7 @@ func (n *node) GetID() types.NodeID {
 
 // run starts the node.
 func (n *node) run(
-	serverEndpoint interface{}, dMoment time.Time, logger common.Logger) {
+	serverEndpoint interface{}, dMoment time.Time) {
 	// Run network.
 	if err := n.netModule.Setup(serverEndpoint); err != nil {
 		panic(err)
@@ -145,10 +129,32 @@ func (n *node) run(
 		n.gov.State().RequestChange(test.StateAddNode, pubKey)
 		hashes = append(hashes, nID.Hash)
 	}
-	// This notification is implictly called in full node.
-	n.gov.NotifyRoundHeight(0, 0)
-	// Setup of governance is ready, can be switched to remote mode.
-	n.gov.SwitchToRemoteMode(n.netModule)
+	n.prepareConfigs()
+	if err := n.netModule.Report(&message{Type: setupOK}); err != nil {
+		panic(err)
+	}
+	// Wait for a "ready" server notification.
+readyLoop:
+	for {
+		msg := <-msgChannel
+		ntf := msg.(serverNotification)
+		switch ntf {
+		case ntfReady:
+			break readyLoop
+		case ntfSelectedAsMaster:
+			n.logger.Info(
+				"receive 'selected-as-master' notification from server")
+			for _, c := range n.cfg.Node.Changes {
+				if c.Round <= core.ConfigRoundShift+1 {
+					continue
+				}
+				n.logger.Info("register config change", "change", c)
+				c.RegisterChange(n.gov)
+			}
+		default:
+			panic(fmt.Errorf("receive unexpected server notification: %v", ntf))
+		}
+	}
 	// Setup Consensus.
 	n.consensus = core.NewConsensusForSimulation(
 		dMoment,
@@ -157,7 +163,7 @@ func (n *node) run(
 		n.db,
 		n.netModule,
 		n.prvKey,
-		logger)
+		n.logger)
 	go n.consensus.Run()
 
 	// Blocks forever.
@@ -165,8 +171,9 @@ MainLoop:
 	for {
 		msg := <-msgChannel
 		switch val := msg.(type) {
-		case infoStatus:
-			if val == statusShutdown {
+		case serverNotification:
+			if val == ntfShutdown {
+				n.logger.Info("receive shutdown notification from server")
 				break MainLoop
 			}
 		default:
@@ -178,10 +185,40 @@ MainLoop:
 	if err := n.db.Close(); err != nil {
 		fmt.Println(err)
 	}
-	n.netModule.Report(&message{
-		Type: shutdownAck,
-	})
+	if err := n.netModule.Report(&message{Type: shutdownAck}); err != nil {
+		panic(err)
+	}
 	// TODO(mission): once we have a way to know if consensus is stopped, stop
 	//                the network module.
 	return
 }
+
+func (n *node) prepareConfigs() {
+	// Prepare configurations.
+	cConfig := n.cfg.Node.Consensus
+	n.gov.State().RequestChange(test.StateChangeK, cConfig.K)
+	n.gov.State().RequestChange(test.StateChangePhiRatio, cConfig.PhiRatio)
+	n.gov.State().RequestChange(test.StateChangeNumChains, cConfig.NumChains)
+	n.gov.State().RequestChange(
+		test.StateChangeNotarySetSize, cConfig.NotarySetSize)
+	n.gov.State().RequestChange(test.StateChangeDKGSetSize, cConfig.DKGSetSize)
+	n.gov.State().RequestChange(test.StateChangeLambdaBA, time.Duration(
+		cConfig.LambdaBA)*time.Millisecond)
+	n.gov.State().RequestChange(test.StateChangeLambdaDKG, time.Duration(
+		cConfig.LambdaDKG)*time.Millisecond)
+	n.gov.State().RequestChange(test.StateChangeRoundInterval, time.Duration(
+		cConfig.RoundInterval)*time.Millisecond)
+	n.gov.State().RequestChange(test.StateChangeMinBlockInterval, time.Duration(
+		cConfig.MinBlockInterval)*time.Millisecond)
+	n.gov.State().ProposeCRS(0, crypto.Keccak256Hash([]byte(cConfig.GenesisCRS)))
+	// These rounds are not safe to be registered as pending state change
+	// requests.
+	for i := uint64(0); i <= core.ConfigRoundShift+1; i++ {
+		n.logger.Info("prepare config", "round", i)
+		prepareConfigs(i, n.cfg.Node.Changes, n.gov)
+	}
+	// This notification is implictly called in full node.
+	n.gov.NotifyRoundHeight(0, 0)
+	// Setup of configuration is ready, can be switched to remote mode.
+	n.gov.SwitchToRemoteMode(n.netModule)
+}
diff --git a/simulation/peer-server.go b/simulation/peer-server.go
index 6d1121f..14a825a 100644
--- a/simulation/peer-server.go
+++ b/simulation/peer-server.go
@@ -90,7 +90,7 @@ func (p *PeerServer) handleBlockList(id types.NodeID, blocks *BlockList) {
 		}
 		p.verifiedLen += uint64(length)
 		if p.verifiedLen >= p.cfg.Node.MaxBlock {
-			if err := p.trans.Broadcast(statusShutdown); err != nil {
+			if err := p.trans.Broadcast(ntfShutdown); err != nil {
 				panic(err)
 			}
 		}
@@ -201,6 +201,34 @@ func (p *PeerServer) Run() {
 	for _, pubKey := range p.trans.Peers() {
 		p.peers[types.NewNodeID(pubKey)] = struct{}{}
 	}
+	// Pick a mater node to execute pending config changes.
+	for nID := range p.peers {
+		if err := p.trans.Send(nID, ntfSelectedAsMaster); err != nil {
+			panic(err)
+		}
+		break
+	}
+	// Wait for peers to report 'setupOK' message.
+	readyPeers := make(map[types.NodeID]struct{})
+	for {
+		e := <-p.msgChannel
+		if !p.isNode(e.From) {
+			break
+		}
+		msg := e.Msg.(*message)
+		if msg.Type != setupOK {
+			panic(fmt.Errorf("receive an unexpected peer report: %v", msg))
+		}
+		log.Println("receive setupOK message from", e.From)
+		readyPeers[e.From] = struct{}{}
+		if len(readyPeers) == len(p.peers) {
+			break
+		}
+	}
+	if err := p.trans.Broadcast(ntfReady); err != nil {
+		panic(err)
+	}
+	log.Println("Simulation is ready to go with", len(p.peers), "nodes")
 	// Initialize total order result cache.
 	for id := range p.peers {
 		p.peerTotalOrder[id] = NewTotalOrderResult(id)
diff --git a/simulation/simulation.go b/simulation/simulation.go
index 3a0e9bb..10ccb4c 100644
--- a/simulation/simulation.go
+++ b/simulation/simulation.go
@@ -71,11 +71,11 @@ func Run(cfg *config.Config, logPrefix string) {
 		if err != nil {
 			panic(err)
 		}
-		v := newNode(prv, *cfg)
+		v := newNode(prv, logger, *cfg)
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			v.run(serverEndpoint, dMoment, logger)
+			v.run(serverEndpoint, dMoment)
 		}()
 	}
 
diff --git a/simulation/utils.go b/simulation/utils.go
index a18e8a4..a305d2b 100644
--- a/simulation/utils.go
+++ b/simulation/utils.go
@@ -20,6 +20,9 @@ package simulation
 import (
 	"math"
 	"sort"
+
+	"github.com/dexon-foundation/dexon-consensus/core/test"
+	"github.com/dexon-foundation/dexon-consensus/simulation/config"
 )
 
 func calculateMeanStdDeviationFloat64s(a []float64) (float64, float64) {
@@ -57,3 +60,18 @@ func getMinMedianMaxFloat64s(a []float64) (float64, float64, float64) {
 	sort.Float64s(aCopied)
 	return aCopied[0], aCopied[len(aCopied)/2], aCopied[len(aCopied)-1]
 }
+
+func prepareConfigs(
+	round uint64, cfgs []config.Change, gov *test.Governance) {
+	for _, c := range cfgs {
+		if c.Round != round {
+			continue
+		}
+		t := config.StateChangeTypeFromString(c.Type)
+		if err := gov.State().RequestChange(
+			t, config.StateChangeValueFromString(t, c.Value)); err != nil {
+			panic(err)
+		}
+	}
+	gov.CatchUpWithRound(round)
+}