simulation: add simulation with scheduler (#71)

- Add new field in test.Event: HistoryIndex
   HistoryIndex allow us to access them by their position in event history.
- Record local time in test.App when receiving events.
- Add statisitics module for slices of test.Event.
- add new command line utility *dexcon-simulation-with-scheduler
   to verify the execution time of core.Consensus.

6 files changed, 348 insertions, 46 deletions

diff --git a/integration_test/latency.go b/integration_test/latency.go
index 383d069..8f06084 100644
--- a/integration_test/latency.go
+++ b/integration_test/latency.go
@@ -28,17 +28,27 @@ type LatencyModel interface {
 	Delay() time.Duration
 }
 
-// normalLatencyModel would return latencies in normal distribution.
-type normalLatencyModel struct {
+// NormalLatencyModel would return latencies in normal distribution.
+type NormalLatencyModel struct {
 	Sigma float64
 	Mean  float64
 }
 
 // Delay implements LatencyModel interface.
-func (m *normalLatencyModel) Delay() time.Duration {
+func (m *NormalLatencyModel) Delay() time.Duration {
 	delay := rand.NormFloat64()*m.Sigma + m.Mean
 	if delay < 0 {
 		delay = m.Sigma / 2
 	}
 	return time.Duration(delay) * time.Millisecond
 }
+
+// FixedLatencyModel return fixed latencies.
+type FixedLatencyModel struct {
+	Latency float64
+}
+
+// Delay implements LatencyModel interface.
+func (m *FixedLatencyModel) Delay() time.Duration {
+	return time.Duration(m.Latency) * time.Millisecond
+}
diff --git a/integration_test/non-byzantine_test.go b/integration_test/non-byzantine_test.go
index 111dcd0..827d2ad 100644
--- a/integration_test/non-byzantine_test.go
+++ b/integration_test/non-byzantine_test.go
@@ -33,11 +33,11 @@ type NonByzantineTestSuite struct {
 
 func (s *NonByzantineTestSuite) TestNonByzantine() {
 	var (
-		networkLatency = &normalLatencyModel{
+		networkLatency = &NormalLatencyModel{
 			Sigma: 20,
 			Mean:  250,
 		}
-		proposingLatency = &normalLatencyModel{
+		proposingLatency = &NormalLatencyModel{
 			Sigma: 30,
 			Mean:  500,
 		}
@@ -46,42 +46,19 @@ func (s *NonByzantineTestSuite) TestNonByzantine() {
 		req  = s.Require()
 	)
 
-	gov, err := test.NewGovernance(25, 700)
+	apps, dbs, validators, err := PrepareValidators(
+		25, networkLatency, proposingLatency)
 	req.Nil(err)
 	now := time.Now().UTC()
-	for vID := range gov.GetValidatorSet() {
-		apps[vID] = test.NewApp()
-
-		db, err := blockdb.NewMemBackedBlockDB()
-		req.Nil(err)
-		dbs[vID] = db
-	}
-	stopper := test.NewStopByConfirmedBlocks(50, apps, dbs)
-	sch := test.NewScheduler(stopper)
-	for vID := range gov.GetValidatorSet() {
-		key, err := gov.GetPrivateKey(vID)
-		req.Nil(err)
-		v := newValidator(
-			apps[vID],
-			gov,
-			dbs[vID],
-			key,
-			vID,
-			networkLatency,
-			proposingLatency)
+	sch := test.NewScheduler(test.NewStopByConfirmedBlocks(50, apps, dbs))
+	for vID, v := range validators {
 		sch.RegisterEventHandler(vID, v)
-		req.Nil(sch.Seed(newProposeBlockEvent(vID, now)))
+		req.Nil(sch.Seed(NewProposeBlockEvent(vID, now)))
 	}
 	sch.Run(10)
 	// Check results by comparing test.App instances.
-	for vFrom := range gov.GetValidatorSet() {
-		req.Nil(apps[vFrom].Verify())
-		for vTo := range gov.GetValidatorSet() {
-			if vFrom == vTo {
-				continue
-			}
-			req.Nil(apps[vFrom].Compare(apps[vTo]))
-		}
+	if err = VerifyApps(apps); err != nil {
+		panic(err)
 	}
 }
 
diff --git a/integration_test/stats.go b/integration_test/stats.go
new file mode 100644
index 0000000..ae8ded4
--- /dev/null
+++ b/integration_test/stats.go
@@ -0,0 +1,176 @@
+package integration
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus-core/core/test"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+)
+
+// Errors when calculating statistics for events.
+var (
+	ErrUnknownEvent              = fmt.Errorf("unknown event")
+	ErrUnknownConsensusEventType = fmt.Errorf("unknown consensus event type")
+)
+
+// StatsSet represents accumulatee result of a group of related events
+// (ex. All events from one validator).
+type StatsSet struct {
+	ProposedBlockCount      int
+	ReceivedBlockCount      int
+	StronglyAckedBlockCount int
+	TotalOrderedBlockCount  int
+	DeliveredBlockCount     int
+	ProposingLatency        time.Duration
+	ReceivingLatency        time.Duration
+	PrepareExecLatency      time.Duration
+	ProcessExecLatency      time.Duration
+}
+
+// newBlockProposeEvent accumulates a block proposing event.
+func (s *StatsSet) newBlockProposeEvent(
+	e *test.Event, payload *consensusEventPayload, history []*test.Event) {
+
+	// Find previous block proposing event.
+	if e.ParentHistoryIndex != -1 {
+		parentEvent := history[e.ParentHistoryIndex]
+		s.ProposingLatency +=
+			e.Time.Sub(parentEvent.Time) - parentEvent.ExecInterval
+	}
+	s.PrepareExecLatency += e.ExecInterval
+	s.ProposedBlockCount++
+}
+
+// newBlockReceiveEvent accumulates a block received event.
+func (s *StatsSet) newBlockReceiveEvent(
+	e *test.Event,
+	payload *consensusEventPayload,
+	history []*test.Event,
+	app *test.App) {
+
+	// Find previous block proposing event.
+	parentEvent := history[e.ParentHistoryIndex]
+	s.ReceivingLatency +=
+		e.Time.Sub(parentEvent.Time) - parentEvent.ExecInterval
+	s.ProcessExecLatency += e.ExecInterval
+	s.ReceivedBlockCount++
+
+	// Find statistics from test.App
+	block := payload.PiggyBack.(*types.Block)
+	app.Check(func(app *test.App) {
+		// Is this block strongly acked?
+		if _, exists := app.Acked[block.Hash]; !exists {
+			return
+		}
+		s.StronglyAckedBlockCount++
+
+		// Is this block total ordered?
+		if _, exists := app.TotalOrderedByHash[block.Hash]; !exists {
+			return
+		}
+		s.TotalOrderedBlockCount++
+
+		// Is this block delivered?
+		if _, exists := app.Delivered[block.Hash]; !exists {
+			return
+		}
+		s.DeliveredBlockCount++
+	})
+}
+
+// done would divide the latencies we cached with related event count. This way
+// to calculate average latency is more accurate.
+func (s *StatsSet) done(validatorCount int) {
+	s.ProposingLatency /= time.Duration(s.ProposedBlockCount - validatorCount)
+	s.ReceivingLatency /= time.Duration(s.ReceivedBlockCount)
+	s.PrepareExecLatency /= time.Duration(s.ProposedBlockCount)
+	s.ProcessExecLatency /= time.Duration(s.ReceivedBlockCount)
+}
+
+// Stats is statistics of a slice of test.Event generated by validators.
+type Stats struct {
+	ByValidator   map[types.ValidatorID]*StatsSet
+	All           *StatsSet
+	BPS           float64
+	ExecutionTime time.Duration
+}
+
+// NewStats constructs an Stats instance by providing a slice of
+// test.Event.
+func NewStats(
+	history []*test.Event, apps map[types.ValidatorID]*test.App) (
+	stats *Stats, err error) {
+
+	stats = &Stats{
+		ByValidator: make(map[types.ValidatorID]*StatsSet),
+		All:         &StatsSet{},
+	}
+	if err = stats.calculate(history, apps); err != nil {
+		stats = nil
+	}
+	stats.summary(history)
+	return
+}
+
+func (stats *Stats) calculate(
+	history []*test.Event, apps map[types.ValidatorID]*test.App) error {
+
+	defer func() {
+		stats.All.done(len(stats.ByValidator))
+		for _, set := range stats.ByValidator {
+			set.done(1)
+		}
+	}()
+
+	for _, e := range history {
+		payload, ok := e.Payload.(*consensusEventPayload)
+		if !ok {
+			return ErrUnknownEvent
+		}
+		switch payload.Type {
+		case evtProposeBlock:
+			stats.All.newBlockProposeEvent(
+				e, payload, history)
+			stats.getStatsSetByValidator(e.ValidatorID).newBlockProposeEvent(
+				e, payload, history)
+		case evtReceiveBlock:
+			stats.All.newBlockReceiveEvent(
+				e, payload, history, apps[e.ValidatorID])
+			stats.getStatsSetByValidator(e.ValidatorID).newBlockReceiveEvent(
+				e, payload, history, apps[e.ValidatorID])
+		default:
+			return ErrUnknownConsensusEventType
+		}
+	}
+	return nil
+}
+
+func (stats *Stats) getStatsSetByValidator(
+	vID types.ValidatorID) (s *StatsSet) {
+
+	s = stats.ByValidator[vID]
+	if s == nil {
+		s = &StatsSet{}
+		stats.ByValidator[vID] = s
+	}
+	return
+}
+
+func (stats *Stats) summary(history []*test.Event) {
+	// Find average delivered block count among all blocks.
+	totalConfirmedBlocks := 0
+	for _, s := range stats.ByValidator {
+		totalConfirmedBlocks += s.DeliveredBlockCount
+	}
+	averageConfirmedBlocks := totalConfirmedBlocks / len(stats.ByValidator)
+
+	// Find execution time.
+	// Note: it's a simplified way to calculate the execution time:
+	//       the latest event might not be at the end of history when
+	//       the number of worker routine is larger than 1.
+	stats.ExecutionTime = history[len(history)-1].Time.Sub(history[0].Time)
+	// Calculate BPS.
+	latencyAsSecond := stats.ExecutionTime.Nanoseconds() / (1000 * 1000 * 1000)
+	stats.BPS = float64(averageConfirmedBlocks) / float64(latencyAsSecond)
+}
diff --git a/integration_test/stats_test.go b/integration_test/stats_test.go
new file mode 100644
index 0000000..e0be126
--- /dev/null
+++ b/integration_test/stats_test.go
@@ -0,0 +1,60 @@
+package integration
+
+import (
+	"testing"
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus-core/core/test"
+	"github.com/stretchr/testify/suite"
+)
+
+type EventStatsTestSuite struct {
+	suite.Suite
+}
+
+func (s *EventStatsTestSuite) TestCalculate() {
+	// Setup a test with fixed latency in proposing and network,
+	// and make sure the calculated statistics is expected.
+	var (
+		networkLatency   = &FixedLatencyModel{Latency: 100}
+		proposingLatency = &FixedLatencyModel{Latency: 300}
+		req              = s.Require()
+	)
+
+	apps, dbs, validators, err := PrepareValidators(
+		7, networkLatency, proposingLatency)
+	req.Nil(err)
+
+	sch := test.NewScheduler(test.NewStopByConfirmedBlocks(50, apps, dbs))
+	now := time.Now().UTC()
+	for vID, v := range validators {
+		sch.RegisterEventHandler(vID, v)
+		req.Nil(sch.Seed(NewProposeBlockEvent(vID, now)))
+	}
+	sch.Run(10)
+	req.Nil(VerifyApps(apps))
+	// Check total statistics result.
+	stats, err := NewStats(sch.CloneExecutionHistory(), apps)
+	req.Nil(err)
+	req.True(stats.All.ProposedBlockCount > 350)
+	req.True(stats.All.ReceivedBlockCount > 350)
+	req.True(stats.All.StronglyAckedBlockCount > 350)
+	req.True(stats.All.TotalOrderedBlockCount >= 350)
+	req.True(stats.All.DeliveredBlockCount >= 350)
+	req.Equal(stats.All.ProposingLatency, 300*time.Millisecond)
+	req.Equal(stats.All.ReceivingLatency, 100*time.Millisecond)
+	// Check statistics for each validator.
+	for _, vStats := range stats.ByValidator {
+		req.True(vStats.ProposedBlockCount > 50)
+		req.True(vStats.ReceivedBlockCount > 50)
+		req.True(vStats.StronglyAckedBlockCount > 50)
+		req.True(vStats.TotalOrderedBlockCount >= 50)
+		req.True(vStats.DeliveredBlockCount >= 50)
+		req.Equal(vStats.ProposingLatency, 300*time.Millisecond)
+		req.Equal(vStats.ReceivingLatency, 100*time.Millisecond)
+	}
+}
+
+func TestEventStats(t *testing.T) {
+	suite.Run(t, new(EventStatsTestSuite))
+}
diff --git a/integration_test/utils.go b/integration_test/utils.go
new file mode 100644
index 0000000..c1eafb7
--- /dev/null
+++ b/integration_test/utils.go
@@ -0,0 +1,72 @@
+package integration
+
+import (
+	"github.com/dexon-foundation/dexon-consensus-core/blockdb"
+	"github.com/dexon-foundation/dexon-consensus-core/core/test"
+	"github.com/dexon-foundation/dexon-consensus-core/core/types"
+	"github.com/dexon-foundation/dexon-consensus-core/crypto"
+)
+
+// PrepareValidators setups validators for testing.
+func PrepareValidators(
+	validatorCount int,
+	networkLatency, proposingLatency LatencyModel) (
+	apps map[types.ValidatorID]*test.App,
+	dbs map[types.ValidatorID]blockdb.BlockDatabase,
+	validators map[types.ValidatorID]*Validator,
+	err error) {
+
+	var (
+		db  blockdb.BlockDatabase
+		key crypto.PrivateKey
+	)
+
+	apps = make(map[types.ValidatorID]*test.App)
+	dbs = make(map[types.ValidatorID]blockdb.BlockDatabase)
+	validators = make(map[types.ValidatorID]*Validator)
+
+	gov, err := test.NewGovernance(validatorCount, 700)
+	if err != nil {
+		return
+	}
+	for vID := range gov.GetValidatorSet() {
+		apps[vID] = test.NewApp()
+
+		if db, err = blockdb.NewMemBackedBlockDB(); err != nil {
+			return
+		}
+		dbs[vID] = db
+	}
+	for vID := range gov.GetValidatorSet() {
+		if key, err = gov.GetPrivateKey(vID); err != nil {
+			return
+		}
+		validators[vID] = NewValidator(
+			apps[vID],
+			gov,
+			dbs[vID],
+			key,
+			vID,
+			networkLatency,
+			proposingLatency)
+	}
+	return
+}
+
+// VerifyApps is a helper to check delivery between test.Apps
+func VerifyApps(apps map[types.ValidatorID]*test.App) (err error) {
+	for vFrom, fromApp := range apps {
+		if err = fromApp.Verify(); err != nil {
+			return
+		}
+		for vTo, toApp := range apps {
+			if vFrom == vTo {
+				continue
+			}
+			if err = fromApp.Compare(toApp); err != nil {
+				return
+			}
+		}
+	}
+	return
+}
diff --git a/integration_test/validator.go b/integration_test/validator.go
index 00ffff2..fd7a7ad 100644
--- a/integration_test/validator.go
+++ b/integration_test/validator.go
@@ -41,13 +41,17 @@ type consensusEventPayload struct {
 	PiggyBack interface{}
 }
 
-func newProposeBlockEvent(vID types.ValidatorID, when time.Time) *test.Event {
+// NewProposeBlockEvent constructs an test.Event that would trigger
+// block proposing.
+func NewProposeBlockEvent(vID types.ValidatorID, when time.Time) *test.Event {
 	return test.NewEvent(vID, when, &consensusEventPayload{
 		Type: evtProposeBlock,
 	})
 }
 
-func newReceiveBlockEvent(
+// NewReceiveBlockEvent constructs an test.Event that would trigger
+// block received.
+func NewReceiveBlockEvent(
 	vID types.ValidatorID, when time.Time, block *types.Block) *test.Event {
 
 	return test.NewEvent(vID, when, &consensusEventPayload{
@@ -56,7 +60,8 @@ func newReceiveBlockEvent(
 	})
 }
 
-type validator struct {
+// Validator is designed to work with test.Scheduler.
+type Validator struct {
 	ID               types.ValidatorID
 	cons             *core.Consensus
 	gov              core.Governance
@@ -64,16 +69,17 @@ type validator struct {
 	proposingLatency LatencyModel
 }
 
-func newValidator(
+// NewValidator constructs an instance of Validator.
+func NewValidator(
 	app core.Application,
 	gov core.Governance,
 	db blockdb.BlockDatabase,
 	privateKey crypto.PrivateKey,
 	vID types.ValidatorID,
 	networkLatency LatencyModel,
-	proposingLatency LatencyModel) *validator {
+	proposingLatency LatencyModel) *Validator {
 
-	return &validator{
+	return &Validator{
 		ID:               vID,
 		gov:              gov,
 		networkLatency:   networkLatency,
@@ -83,7 +89,8 @@ func newValidator(
 	}
 }
 
-func (v *validator) Handle(e *test.Event) (events []*test.Event) {
+// Handle implements test.EventHandler interface.
+func (v *Validator) Handle(e *test.Event) (events []*test.Event) {
 	payload := e.Payload.(*consensusEventPayload)
 	switch payload.Type {
 	case evtProposeBlock:
@@ -96,7 +103,7 @@ func (v *validator) Handle(e *test.Event) (events []*test.Event) {
 	return
 }
 
-func (v *validator) handleProposeBlock(when time.Time, piggyback interface{}) (
+func (v *Validator) handleProposeBlock(when time.Time, piggyback interface{}) (
 	events []*test.Event, err error) {
 
 	b := &types.Block{ProposerID: v.ID}
@@ -111,16 +118,16 @@ func (v *validator) handleProposeBlock(when time.Time, piggyback interface{}) (
 		if vID == v.ID {
 			continue
 		}
-		events = append(events, newReceiveBlockEvent(
+		events = append(events, NewReceiveBlockEvent(
 			vID, when.Add(v.networkLatency.Delay()), b.Clone()))
 	}
 	// Create next 'block proposing' event for this validators.
-	events = append(events, newProposeBlockEvent(
+	events = append(events, NewProposeBlockEvent(
 		v.ID, when.Add(v.proposingLatency.Delay())))
 	return
 }
 
-func (v *validator) handleReceiveBlock(piggyback interface{}) (
+func (v *Validator) handleReceiveBlock(piggyback interface{}) (
 	events []*test.Event, err error) {
 
 	err = v.cons.ProcessBlock(piggyback.(*types.Block))