package dex import ( "errors" "fmt" "sync" "sync/atomic" "time" dexCore "github.com/dexon-foundation/dexon-consensus/core" coreEcdsa "github.com/dexon-foundation/dexon-consensus/core/crypto/ecdsa" "github.com/dexon-foundation/dexon-consensus/core/syncer" coreTypes "github.com/dexon-foundation/dexon-consensus/core/types" "github.com/dexon-foundation/dexon/core" "github.com/dexon-foundation/dexon/dex/db" "github.com/dexon-foundation/dexon/log" "github.com/dexon-foundation/dexon/rlp" ) var ( forceSyncTimeout = 20 * time.Second ) type blockProposer struct { mu sync.Mutex running int32 syncing int32 proposing int32 dex *Dexon watchCat *syncer.WatchCat dMoment time.Time wg sync.WaitGroup stopCh chan struct{} } func NewBlockProposer(dex *Dexon, watchCat *syncer.WatchCat, dMoment time.Time) *blockProposer { return &blockProposer{ dex: dex, watchCat: watchCat, dMoment: dMoment, } } func (b *blockProposer) Start() error { b.mu.Lock() defer b.mu.Unlock() if !atomic.CompareAndSwapInt32(&b.running, 0, 1) { return fmt.Errorf("block proposer is already running") } log.Info("Block proposer started") b.stopCh = make(chan struct{}) b.wg.Add(1) go func() { defer b.wg.Done() defer atomic.StoreInt32(&b.running, 0) var err error var c *dexCore.Consensus if b.dMoment.After(time.Now()) { c = b.initConsensus() } else { c, err = b.syncConsensus() } if err != nil { log.Error("Block proposer stopped, before start running", "err", err) return } b.run(c) log.Info("Block proposer successfully stopped") }() return nil } func (b *blockProposer) run(c *dexCore.Consensus) { log.Info("Start running consensus core") go c.Run() atomic.StoreInt32(&b.proposing, 1) <-b.stopCh log.Debug("Block proposer receive stop signal") } func (b *blockProposer) Stop() { log.Info("Stopping block proposer") b.mu.Lock() defer b.mu.Unlock() if atomic.LoadInt32(&b.running) == 1 { atomic.StoreInt32(&b.dex.protocolManager.receiveEnabled, 0) close(b.stopCh) b.wg.Wait() atomic.StoreInt32(&b.proposing, 0) } log.Info("Block proposer stopped") } func (b *blockProposer) IsCoreSyncing() bool { return atomic.LoadInt32(&b.syncing) == 1 } func (b *blockProposer) IsProposing() bool { return atomic.LoadInt32(&b.proposing) == 1 } func (b *blockProposer) initConsensus() *dexCore.Consensus { db := db.NewDatabase(b.dex.chainDb) privkey := coreEcdsa.NewPrivateKeyFromECDSA(b.dex.config.PrivateKey) return dexCore.NewConsensus(b.dMoment, b.dex.app, b.dex.governance, db, b.dex.network, privkey, log.Root()) } func (b *blockProposer) syncConsensus() (*dexCore.Consensus, error) { atomic.StoreInt32(&b.syncing, 1) defer atomic.StoreInt32(&b.syncing, 0) db := db.NewDatabase(b.dex.chainDb) privkey := coreEcdsa.NewPrivateKeyFromECDSA(b.dex.config.PrivateKey) consensusSync := syncer.NewConsensus(b.dMoment, b.dex.app, b.dex.governance, db, b.dex.network, privkey, log.Root()) // Start the watchCat. log.Info("Starting sync watchCat ...") b.watchCat.Start() defer b.watchCat.Stop() // Feed the current block we have in local blockchain. cb := b.dex.blockchain.CurrentBlock() if cb.NumberU64() > 0 { var block coreTypes.Block if err := rlp.DecodeBytes(cb.Header().DexconMeta, &block); err != nil { panic(err) } b.watchCat.Feed(block.Position) } blocksToSync := func(coreHeight, height uint64) []*coreTypes.Block { var blocks []*coreTypes.Block for coreHeight < height { var block coreTypes.Block b := b.dex.blockchain.GetBlockByNumber(coreHeight + 1) if err := rlp.DecodeBytes(b.Header().DexconMeta, &block); err != nil { panic(err) } blocks = append(blocks, &block) coreHeight = coreHeight + 1 } return blocks } // Sync all blocks in compaction chain to core. _, coreHeight := db.GetCompactionChainTipInfo() // Stop receiving block proposer message when syncing. atomic.StoreInt32(&b.dex.protocolManager.receiveEnabled, 0) Loop: for { currentBlock := b.dex.blockchain.CurrentBlock() log.Debug("Syncing compaction chain", "core height", coreHeight, "height", currentBlock.NumberU64()) blocks := blocksToSync(coreHeight, currentBlock.NumberU64()) if len(blocks) == 0 { log.Debug("No new block to sync", "current", currentBlock.NumberU64()) break Loop } b.watchCat.Feed(blocks[len(blocks)-1].Position) log.Debug("Filling compaction chain", "num", len(blocks), "first", blocks[0].Finalization.Height, "last", blocks[len(blocks)-1].Finalization.Height) if _, err := consensusSync.SyncBlocks(blocks, false); err != nil { log.Debug("SyncBlocks fail", "err", err) return nil, err } coreHeight = blocks[len(blocks)-1].Finalization.Height select { case <-b.stopCh: return nil, errors.New("early stop") default: } } ch := make(chan core.ChainHeadEvent) sub := b.dex.blockchain.SubscribeChainHeadEvent(ch) defer sub.Unsubscribe() log.Debug("Listen chain head event until synced") // Listen chain head event until synced. ListenLoop: for { select { case ev := <-ch: blocks := blocksToSync(coreHeight, ev.Block.NumberU64()) b.watchCat.Feed(blocks[len(blocks)-1].Position) if len(blocks) > 0 { log.Debug("Filling compaction chain", "num", len(blocks), "first", blocks[0].Finalization.Height, "last", blocks[len(blocks)-1].Finalization.Height) synced, err := consensusSync.SyncBlocks(blocks, true) if err != nil { log.Error("SyncBlocks fail", "err", err) return nil, err } atomic.CompareAndSwapInt32(&b.dex.protocolManager.receiveEnabled, 0, 1) if synced { log.Debug("Consensus core synced") break ListenLoop } coreHeight = blocks[len(blocks)-1].Finalization.Height } case <-sub.Err(): log.Debug("System stopped when syncing consensus core") return nil, errors.New("system stop") case <-b.stopCh: log.Debug("Early stop, before consensus core can run") return nil, errors.New("early stop") case <-time.After(forceSyncTimeout): log.Debug("no new chain head for a while") if p := b.dex.protocolManager.peers.BestPeer(); p != nil { log.Debug("try force sync with peer", "id", p.id) b.dex.protocolManager.synchronise(p, true) } else { log.Debug("no peer to sync") } case <-b.watchCat.Meow(): log.Info("WatchCat signaled to stop syncing") // Sleep until the next consensus start time slot. // The interval T_i need to meet the following requirement: // // T_i > T_timeout + T_panic + T_restart // // Currently, T_timeout = 120, T_panic = 60, T_restart ~ 60 // // We set T_i = 600 to be safe. interval := int64(600) nextDMoment := (time.Now().Unix()/interval + 1) * interval log.Info("Sleeping until next starting time", "time", nextDMoment) time.Sleep(time.Duration(nextDMoment-time.Now().Unix()) * time.Second) atomic.StoreInt32(&b.dex.protocolManager.receiveEnabled, 1) consensusSync.ForceSync(true) break ListenLoop } } return consensusSync.GetSyncedConsensus() }