core: Integrate totalOrdering syncer (#322)

2 files changed, 29 insertions, 21 deletions

diff --git a/core/consensus.go b/core/consensus.go
index 3d46c5c..69131e6 100644
--- a/core/consensus.go
+++ b/core/consensus.go
@@ -284,6 +284,7 @@ type Consensus struct {
 	// Dexon consensus v1's modules.
 	lattice  *Lattice
 	ccModule *compactionChain
+	toSyncer *totalOrderingSyncer
 
 	// Interfaces.
 	db        blockdb.BlockDatabase
@@ -1066,6 +1067,7 @@ func (con *Consensus) processFinalizedBlock(block *types.Block) (err error) {
 				}
 				err = nil
 			}
+			con.lattice.ProcessFinalizedBlock(b)
 			con.deliverBlock(b)
 		}
 	}
diff --git a/core/lattice.go b/core/lattice.go
index 108f288..634ed8b 100644
--- a/core/lattice.go
+++ b/core/lattice.go
@@ -41,6 +41,7 @@ type Lattice struct {
 	pool       blockPool
 	retryAdd   bool
 	data       *latticeData
+	toSyncer   *totalOrderingSyncer
 	toModule   *totalOrdering
 	ctModule   *consensusTimestamp
 	logger     common.Logger
@@ -64,6 +65,7 @@ func NewLattice(
 		debug:      debug,
 		pool:       newBlockPool(cfg.NumChains),
 		data:       newLatticeData(db, dMoment, round, cfg),
+		toSyncer:   newTotalOrderingSyncer(cfg.NumChains),
 		toModule:   newTotalOrdering(dMoment, cfg),
 		ctModule:   newConsensusTimestamp(dMoment, round, cfg.NumChains),
 		logger:     logger,
@@ -236,28 +238,31 @@ func (l *Lattice) ProcessBlock(
 		return
 	}
 
-	// Perform total ordering for each block added to lattice.
-	for _, b = range inLattice {
-		toDelivered, deliveredMode, err = l.toModule.processBlock(b)
-		if err != nil {
-			// All errors from total ordering is serious, should panic.
-			panic(err)
-		}
-		if len(toDelivered) == 0 {
-			continue
-		}
-		hashes := make(common.Hashes, len(toDelivered))
-		for idx := range toDelivered {
-			hashes[idx] = toDelivered[idx].Hash
-		}
-		if l.debug != nil {
-			l.debug.TotalOrderingDelivered(hashes, deliveredMode)
-		}
-		// Perform consensus timestamp module.
-		if err = l.ctModule.processBlocks(toDelivered); err != nil {
-			return
+	for _, blockToSyncer := range inLattice {
+		toTotalOrdering := l.toSyncer.processBlock(blockToSyncer)
+		// Perform total ordering for each block added to lattice.
+		for _, b = range toTotalOrdering {
+			toDelivered, deliveredMode, err = l.toModule.processBlock(b)
+			if err != nil {
+				// All errors from total ordering is serious, should panic.
+				panic(err)
+			}
+			if len(toDelivered) == 0 {
+				continue
+			}
+			hashes := make(common.Hashes, len(toDelivered))
+			for idx := range toDelivered {
+				hashes[idx] = toDelivered[idx].Hash
+			}
+			if l.debug != nil {
+				l.debug.TotalOrderingDelivered(hashes, deliveredMode)
+			}
+			// Perform consensus timestamp module.
+			if err = l.ctModule.processBlocks(toDelivered); err != nil {
+				return
+			}
+			delivered = append(delivered, toDelivered...)
 		}
-		delivered = append(delivered, toDelivered...)
 	}
 	return
 }
@@ -304,4 +309,5 @@ func (l *Lattice) ProcessFinalizedBlock(b *types.Block) {
 		panic(err)
 	}
 	l.pool.purgeBlocks(b.Position.ChainID, b.Position.Height)
+	l.toSyncer.processFinalizedBlock(b)
 }