core: Add a nonBlockingApplication. (#62)

4 files changed, 238 insertions, 8 deletions

diff --git a/core/consensus.go b/core/consensus.go
index 9b95e56..f591db5 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -87,7 +87,7 @@ func NewConsensus(
 		rbModule: rb,
 		toModule: to,
 		ctModule: newConsensusTimestamp(),
-		app:      app,
+		app:      newNonBlockingApplication(app),
 		gov:      gov,
 		db:       db,
 		prvKey:   prv,
diff --git a/core/consensus_test.go b/core/consensus_test.go
index 522fcbb..2887d66 100644
--- a/core/consensus_test.go
+++ b/core/consensus_test.go
@@ -47,7 +47,7 @@ func (s *ConsensusTestSuite) prepareGenesisBlock(
 }
 
 func (s *ConsensusTestSuite) prepareConsensus(
-	gov *test.Governance, vID types.ValidatorID) (*test.App, *Consensus) {
+	gov *test.Governance, vID types.ValidatorID) (*Application, *Consensus) {
 
 	app := test.NewApp()
 	db, err := blockdb.NewMemBackedBlockDB()
@@ -55,7 +55,7 @@ func (s *ConsensusTestSuite) prepareConsensus(
 	prv, exist := gov.PrivateKeys[vID]
 	s.Require().True(exist)
 	con := NewConsensus(app, gov, db, prv, eth.SigToPub)
-	return app, con
+	return &con.app, con
 }
 
 func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
@@ -84,13 +84,13 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 
 	// Setup core.Consensus and test.App.
 	objs := map[types.ValidatorID]*struct {
-		app *test.App
+		app *Application
 		con *Consensus
 	}{}
 	for _, vID := range validators {
 		app, con := s.prepareConsensus(gov, vID)
 		objs[vID] = &struct {
-			app *test.App
+			app *Application
 			con *Consensus
 		}{app, con}
 	}
@@ -246,7 +246,14 @@ func (s *ConsensusTestSuite) TestSimpleDeliverBlock() {
 		req.Equal(t, app.Delivered[b11.Hash])
 	}
 	for _, obj := range objs {
-		verify(obj.app)
+		app := *obj.app
+		if nbapp, ok := app.(*nonBlockingApplication); ok {
+			nbapp.wait()
+			app = nbapp.app
+		}
+		testApp, ok := app.(*test.App)
+		s.Require().True(ok)
+		verify(testApp)
 	}
 }
 
@@ -269,13 +276,13 @@ func (s *ConsensusTestSuite) TestPrepareBlock() {
 	}
 	// Setup core.Consensus and test.App.
 	objs := map[types.ValidatorID]*struct {
-		app *test.App
+		app *Application
 		con *Consensus
 	}{}
 	for _, vID := range validators {
 		app, con := s.prepareConsensus(gov, vID)
 		objs[vID] = &struct {
-			app *test.App
+			app *Application
 			con *Consensus
 		}{app, con}
 	}
diff --git a/core/nonblocking-application.go b/core/nonblocking-application.go
new file mode 100644
index 0000000..4c06ab9
--- /dev/null
+++ b/core/nonblocking-application.go
@@ -0,0 +1,126 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core 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-core 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-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+)
+
+type stronglyAckedEvent struct {
+	blockHash common.Hash
+}
+
+type totalOrderingDeliverEvent struct {
+	blockHashes common.Hashes
+	early       bool
+}
+
+type deliverBlockEvent struct {
+	blockHash common.Hash
+	timestamp time.Time
+}
+
+// nonBlockingApplication implements Application and is a decorator for
+// Application that makes the methods to be non-blocking.
+type nonBlockingApplication struct {
+	app          Application
+	eventChan    chan interface{}
+	events       []interface{}
+	eventsChange *sync.Cond
+	running      sync.WaitGroup
+}
+
+func newNonBlockingApplication(app Application) *nonBlockingApplication {
+	nonBlockingApp := &nonBlockingApplication{
+		app:          app,
+		eventChan:    make(chan interface{}, 6),
+		events:       make([]interface{}, 0, 100),
+		eventsChange: sync.NewCond(&sync.Mutex{}),
+	}
+	go nonBlockingApp.run()
+	return nonBlockingApp
+}
+
+func (app *nonBlockingApplication) addEvent(event interface{}) {
+	app.eventsChange.L.Lock()
+	defer app.eventsChange.L.Unlock()
+	app.events = append(app.events, event)
+	app.eventsChange.Broadcast()
+}
+
+func (app *nonBlockingApplication) run() {
+	// This go routine consume the first event from events and call the
+	// corresponding method of app.
+	for {
+		var event interface{}
+		func() {
+			app.eventsChange.L.Lock()
+			defer app.eventsChange.L.Unlock()
+			for len(app.events) == 0 {
+				app.eventsChange.Wait()
+			}
+			event = app.events[0]
+			app.events = app.events[1:]
+			app.running.Add(1)
+		}()
+		switch e := event.(type) {
+		case stronglyAckedEvent:
+			app.app.StronglyAcked(e.blockHash)
+		case totalOrderingDeliverEvent:
+			app.app.TotalOrderingDeliver(e.blockHashes, e.early)
+		case deliverBlockEvent:
+			app.app.DeliverBlock(e.blockHash, e.timestamp)
+		default:
+			fmt.Printf("Unknown event %v.", e)
+		}
+		app.running.Done()
+		app.eventsChange.Broadcast()
+	}
+}
+
+// wait will wait for all event in events finishes.
+func (app *nonBlockingApplication) wait() {
+	app.eventsChange.L.Lock()
+	defer app.eventsChange.L.Unlock()
+	for len(app.events) > 0 {
+		app.eventsChange.Wait()
+	}
+	app.running.Wait()
+}
+
+// StronglyAcked is called when a block is strongly acked.
+func (app *nonBlockingApplication) StronglyAcked(blockHash common.Hash) {
+	app.addEvent(stronglyAckedEvent{blockHash})
+}
+
+// TotalOrderingDeliver is called when the total ordering algorithm deliver
+// a set of block.
+func (app *nonBlockingApplication) TotalOrderingDeliver(
+	blockHashes common.Hashes, early bool) {
+	app.addEvent(totalOrderingDeliverEvent{blockHashes, early})
+}
+
+// DeliverBlock is called when a block is add to the compaction chain.
+func (app *nonBlockingApplication) DeliverBlock(
+	blockHash common.Hash, timestamp time.Time) {
+	app.addEvent(deliverBlockEvent{blockHash, timestamp})
+}
diff --git a/core/nonblocking-application_test.go b/core/nonblocking-application_test.go
new file mode 100644
index 0000000..070eae3
--- /dev/null
+++ b/core/nonblocking-application_test.go
@@ -0,0 +1,97 @@
+// Copyright 2018 The dexon-consensus-core Authors
+// This file is part of the dexon-consensus-core library.
+//
+// The dexon-consensus-core 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-core 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-core library. If not, see
+// <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/suite"
+
+	"github.com/dexon-foundation/dexon-consensus-core/common"
+)
+
+type slowApp struct {
+	sleep                time.Duration
+	stronglyAcked        map[common.Hash]struct{}
+	totalOrderingDeliver map[common.Hash]struct{}
+	deliverBlock         map[common.Hash]struct{}
+}
+
+func newSlowApp(sleep time.Duration) *slowApp {
+	return &slowApp{
+		sleep:                sleep,
+		stronglyAcked:        make(map[common.Hash]struct{}),
+		totalOrderingDeliver: make(map[common.Hash]struct{}),
+		deliverBlock:         make(map[common.Hash]struct{}),
+	}
+}
+
+func (app *slowApp) StronglyAcked(blockHash common.Hash) {
+	time.Sleep(app.sleep)
+	app.stronglyAcked[blockHash] = struct{}{}
+}
+
+func (app *slowApp) TotalOrderingDeliver(blockHashes common.Hashes, early bool) {
+	time.Sleep(app.sleep)
+	for _, hash := range blockHashes {
+		app.totalOrderingDeliver[hash] = struct{}{}
+	}
+}
+
+func (app *slowApp) DeliverBlock(blockHash common.Hash, timestamp time.Time) {
+	time.Sleep(app.sleep)
+	app.deliverBlock[blockHash] = struct{}{}
+}
+
+type NonBlockingAppTestSuite struct {
+	suite.Suite
+}
+
+func (s *NonBlockingAppTestSuite) TestNonBlockingApplication() {
+	sleep := 50 * time.Millisecond
+	app := newSlowApp(sleep)
+	nbapp := newNonBlockingApplication(app)
+	hashes := make(common.Hashes, 10)
+	for idx := range hashes {
+		hashes[idx] = common.NewRandomHash()
+	}
+	now := time.Now().UTC()
+	shouldFinish := now.Add(100 * time.Millisecond)
+
+	// Start doing some 'heavy' job.
+	for _, hash := range hashes {
+		nbapp.StronglyAcked(hash)
+		nbapp.DeliverBlock(hash, time.Now().UTC())
+	}
+	nbapp.TotalOrderingDeliver(hashes, true)
+
+	// nonBlockingApplication should be non-blocking.
+	s.True(shouldFinish.After(time.Now().UTC()))
+
+	nbapp.wait()
+	for _, hash := range hashes {
+		s.Contains(app.stronglyAcked, hash)
+		s.Contains(app.totalOrderingDeliver, hash)
+		s.Contains(app.deliverBlock, hash)
+	}
+}
+
+func TestNonBlockingApplication(t *testing.T) {
+	suite.Run(t, new(NonBlockingAppTestSuite))
+}