fix err download block on masternode

cmd/tomo/main.go

@@ -120,6 +120,7 @@ var (
 		//utils.GpoPercentileFlag,
 		//utils.ExtraDataFlag,
 		configFileFlag,
+		utils.CommitTxWhenNotMiningFlag,
 	}
 
 	rpcFlags = []cli.Flag{

cmd/utils/flags.go

@@ -113,6 +113,11 @@ func NewApp(gitCommit, usage string) *cli.App {
 
 var (
 	// General settings
+	CommitTxWhenNotMiningFlag = DirectoryFlag{
+		Name:  "committxwhennotmining",
+		Usage: "Always commit transactions",
+		Value: DirectoryString{node.DefaultDataDir()},
+	}
 	DataDirFlag = DirectoryFlag{
 		Name:  "datadir",
 		Usage: "Data directory for the databases and keystore",
@@ -897,6 +902,9 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
 	if ctx.GlobalIsSet(NoUSBFlag.Name) {
 		cfg.NoUSB = ctx.GlobalBool(NoUSBFlag.Name)
 	}
+	if ctx.GlobalIsSet(CommitTxWhenNotMiningFlag.Name) {
+		cfg.CommitTxWhenNotMining = ctx.GlobalBool(CommitTxWhenNotMiningFlag.Name)
+	}
 }
 
 func setGPO(ctx *cli.Context, cfg *gasprice.Config) {

eth/backend.go

@@ -173,7 +173,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	if eth.protocolManager, err = NewProtocolManager(eth.chainConfig, config.SyncMode, config.NetworkId, eth.eventMux, eth.txPool, eth.engine, eth.blockchain, chainDb); err != nil {
 		return nil, err
 	}
-	eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine)
+	eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine, ctx.GetConfig().CommitTxWhenNotMining)
 	eth.miner.SetExtra(makeExtraData(config.ExtraData))
 
 	eth.ApiBackend = &EthApiBackend{eth, nil}

eth/downloader/queue.go

@@ -146,9 +146,7 @@ func (q *queue) Reset() {
 // Close marks the end of the sync, unblocking WaitResults.
 // It may be called even if the queue is already closed.
 func (q *queue) Close() {
-	q.lock.Lock()
 	q.closed = true
-	q.lock.Unlock()
 	q.active.Broadcast()
 }
 

miner/miner.go

@@ -57,12 +57,12 @@ type Miner struct {
 	shouldStart int32 // should start indicates whether we should start after sync
 }
 
-func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine consensus.Engine) *Miner {
+func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine consensus.Engine, commitTxWhenNotMining bool) *Miner {
 	miner := &Miner{
 		eth:      eth,
 		mux:      mux,
 		engine:   engine,
-		worker:   newWorker(config, engine, common.Address{}, eth, mux),
+		worker:   newWorker(config, engine, common.Address{}, eth, mux, commitTxWhenNotMining),
 		canStart: 1,
 	}
 	miner.Register(NewCpuAgent(eth.BlockChain(), engine))
@@ -77,7 +77,6 @@ func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine con
 // and halt your mining operation for as long as the DOS continues.
 func (self *Miner) update() {
 	events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
-out:
 	for ev := range events.Chan() {
 		switch ev.Data.(type) {
 		case downloader.StartEvent:
@@ -95,10 +94,6 @@ out:
 			if shouldStart {
 				self.Start(self.coinbase)
 			}
-			// unsubscribe. we're only interested in this event once
-			events.Unsubscribe()
-			// stop immediately and ignore all further pending events
-			break out
 		}
 	}
 }

miner/worker.go

@@ -130,30 +130,35 @@ type worker struct {
 	unconfirmed *unconfirmedBlocks // set of locally mined blocks pending canonicalness confirmations
 
 	// atomic status counters
-	mining int32
+	mining                int32
-	atWork int32
+	atWork                int32
+	commitTxWhenNotMining bool
+	lastParentBlockCommit string
 }
 
-func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase common.Address, eth Backend, mux *event.TypeMux) *worker {
+func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase common.Address, eth Backend, mux *event.TypeMux, commitTxWhenNotMining bool) *worker {
 	worker := &worker{
-		config:         config,
+		config:                config,
-		engine:         engine,
+		engine:                engine,
-		eth:            eth,
+		eth:                   eth,
-		mux:            mux,
+		mux:                   mux,
-		txCh:           make(chan core.TxPreEvent, txChanSize),
+		txCh:                  make(chan core.TxPreEvent, txChanSize),
-		chainHeadCh:    make(chan core.ChainHeadEvent, chainHeadChanSize),
+		chainHeadCh:           make(chan core.ChainHeadEvent, chainHeadChanSize),
-		chainSideCh:    make(chan core.ChainSideEvent, chainSideChanSize),
+		chainSideCh:           make(chan core.ChainSideEvent, chainSideChanSize),
-		chainDb:        eth.ChainDb(),
+		chainDb:               eth.ChainDb(),
-		recv:           make(chan *Result, resultQueueSize),
+		recv:                  make(chan *Result, resultQueueSize),
-		chain:          eth.BlockChain(),
+		chain:                 eth.BlockChain(),
-		proc:           eth.BlockChain().Validator(),
+		proc:                  eth.BlockChain().Validator(),
-		possibleUncles: make(map[common.Hash]*types.Block),
+		possibleUncles:        make(map[common.Hash]*types.Block),
-		coinbase:       coinbase,
+		coinbase:              coinbase,
-		agents:         make(map[Agent]struct{}),
+		agents:                make(map[Agent]struct{}),
-		unconfirmed:    newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth),
+		unconfirmed:           newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth),
+		commitTxWhenNotMining: commitTxWhenNotMining,
+	}
+	if worker.commitTxWhenNotMining {
+		// Subscribe TxPreEvent for tx pool
+		worker.txSub = eth.TxPool().SubscribeTxPreEvent(worker.txCh)
 	}
-	// Subscribe TxPreEvent for tx pool
-	worker.txSub = eth.TxPool().SubscribeTxPreEvent(worker.txCh)
 	// Subscribe events for blockchain
 	worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
 	worker.chainSideSub = eth.BlockChain().SubscribeChainSideEvent(worker.chainSideCh)
@@ -248,16 +253,39 @@ func (self *worker) unregister(agent Agent) {
 }
 
 func (self *worker) update() {
-	defer self.txSub.Unsubscribe()
+	if self.commitTxWhenNotMining {
+		defer self.txSub.Unsubscribe()
+	}
 	defer self.chainHeadSub.Unsubscribe()
 	defer self.chainSideSub.Unsubscribe()
-
+	timeout := time.NewTimer(waitPeriod * time.Second)
+	c := make(chan struct{})
+	finish := make(chan struct{})
+	defer close(finish)
+	defer timeout.Stop()
+	go func() {
+		for {
+			// A real event arrived, process interesting content
+			select {
+			case <-timeout.C:
+				c <- struct{}{}
+			case <-finish:
+				return
+			}
+		}
+	}()
 	for {
 		// A real event arrived, process interesting content
 		select {
-		// Handle ChainHeadEvent
+		case <-c:
+			if atomic.LoadInt32(&self.mining) == 1 {
+				self.commitNewWork()
+			}
+			timeout.Reset(waitPeriod * time.Second)
+			// Handle ChainHeadEvent
 		case <-self.chainHeadCh:
 			self.commitNewWork()
+			timeout.Reset(waitPeriod * time.Second)
 
 			// Handle ChainSideEvent
 		case ev := <-self.chainSideCh:
@@ -283,8 +311,6 @@ func (self *worker) update() {
 				}
 			}
 			// System stopped
-		case <-self.txSub.Err():
-			return
 		case <-self.chainHeadSub.Err():
 			return
 		case <-self.chainSideSub.Err():
@@ -466,6 +492,13 @@ func (self *worker) commitNewWork() {
 	tstart := time.Now()
 	parent := self.chain.CurrentBlock()
 	var signers map[common.Address]struct{}
+	if parent.Hash().Hex() == self.lastParentBlockCommit {
+		return
+	}
+	if !self.commitTxWhenNotMining && atomic.LoadInt32(&self.mining) == 0 {
+		return
+	}
+
 	// Only try to commit new work if we are mining
 	if atomic.LoadInt32(&self.mining) == 1 {
 		// check if we are right after parent's coinbase in the list
@@ -504,19 +537,11 @@ func (self *worker) commitNewWork() {
 					gap += waitPeriodCheckpoint
 				}
 				log.Info("Distance from the parent block", "seconds", gap, "hops", h)
-			L:
+				waitedTime := time.Now().Unix() - parent.Header().Time.Int64()
-				select {
+				if gap > waitedTime {
-				case newBlock := <-self.chainHeadCh:
+					return
-					self.chainHeadCh <- newBlock
-					if newBlock.Block.NumberU64() > parent.NumberU64() {
-						log.Info("New block has came already. Skip this turn", "new block", newBlock.Block.NumberU64(), "current block", parent.NumberU64())
-						return
-					}
-				case <-time.After(time.Duration(gap) * time.Second):
-					// wait enough. It's my turn
-					log.Info("Wait enough. It's my turn", "waited seconds", gap)
-					break L
 				}
+				log.Info("Wait enough. It's my turn", "waited seconds", waitedTime)
 			}
 		}
 	}
@@ -611,6 +636,7 @@ func (self *worker) commitNewWork() {
 	if atomic.LoadInt32(&self.mining) == 1 {
 		log.Info("Commit new mining work", "number", work.Block.Number(), "txs", work.tcount, "special txs", len(specialTxs), "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
 		self.unconfirmed.Shift(work.Block.NumberU64() - 1)
+		self.lastParentBlockCommit = parent.Hash().Hex()
 	}
 	self.push(work)
 }

node/config.go

@@ -147,6 +147,8 @@ type Config struct {
 
 	// Logger is a custom logger to use with the p2p.Server.
 	Logger log.Logger `toml:",omitempty"`
+
+	CommitTxWhenNotMining bool `toml:",omitempty"`
 }
 
 // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into