Merge branch 'miner-broadcast' into core-optimisations-2

Conflicts:
	core/chain_manager.go
	miner/worker.go

Author: obscuren

core/canary.go

@@ -0,0 +1,28 @@
+package core
+
+import (
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/state"
+)
+
+var (
+	jeff      = common.HexToAddress("9d38997c624a71b21278389ea2fdc460d000e4b2")
+	vitalik   = common.HexToAddress("b1e570be07eaa673e4fd0c8265b64ef739385709")
+	christoph = common.HexToAddress("529bc43a5d93789fa28de1961db6a07e752204ae")
+	gav       = common.HexToAddress("e3e942b2aa524293c84ff6c7f87a6635790ad5e4")
+)
+
+// Canary will check the 0'd address of the 4 contracts above.
+// If two or more are set to anything other than a 0 the canary
+// dies a horrible death.
+func Canary(statedb *state.StateDB) bool {
+	r := new(big.Int)
+	r.Add(r, statedb.GetState(jeff, common.Hash{}).Big())
+	r.Add(r, statedb.GetState(vitalik, common.Hash{}).Big())
+	r.Add(r, statedb.GetState(christoph, common.Hash{}).Big())
+	r.Add(r, statedb.GetState(gav, common.Hash{}).Big())
+
+	return r.Cmp(big.NewInt(1)) > 0
+}

core/chain_manager.go

@@ -541,6 +541,58 @@ func (self *ChainManager) flushQueuedBlocks() {
 	}
 }
 
+type writeStatus byte
+
+const (
+	nonStatTy writeStatus = iota
+	canonStatTy
+	splitStatTy
+	sideStatTy
+)
+
+func (self *ChainManager) WriteBlock(block *types.Block) (status writeStatus, err error) {
+	self.wg.Add(1)
+	defer self.wg.Done()
+
+	cblock := self.currentBlock
+	// Compare the TD of the last known block in the canonical chain to make sure it's greater.
+	// At this point it's possible that a different chain (fork) becomes the new canonical chain.
+	if block.Td.Cmp(self.Td()) > 0 {
+		// chain fork
+		if block.ParentHash() != cblock.Hash() {
+			// during split we merge two different chains and create the new canonical chain
+			err := self.merge(cblock, block)
+			if err != nil {
+				return nonStatTy, err
+			}
+
+			status = splitStatTy
+		}
+
+		self.mu.Lock()
+		self.setTotalDifficulty(block.Td)
+		self.insert(block)
+		self.mu.Unlock()
+
+		self.setTransState(state.New(block.Root(), self.stateDb))
+		self.txState.SetState(state.New(block.Root(), self.stateDb))
+
+		status = canonStatTy
+	} else {
+		status = sideStatTy
+	}
+
+	// Write block to database. Eventually we'll have to improve on this and throw away blocks that are
+	// not in the canonical chain.
+	self.mu.Lock()
+	self.enqueueForWrite(block)
+	self.mu.Unlock()
+	// Delete from future blocks
+	self.futureBlocks.Delete(block.Hash())
+
+	return
+}
+
 // InsertChain will attempt to insert the given chain in to the canonical chain or, otherwise, create a fork. It an error is returned
 // it will return the index number of the failing block as well an error describing what went wrong (for possible errors see core/errors.go).
 func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
@@ -635,57 +687,29 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
 
 		txcount += len(block.Transactions())
 
-		cblock := self.currentBlock
-		// Compare the TD of the last known block in the canonical chain to make sure it's greater.
-		// At this point it's possible that a different chain (fork) becomes the new canonical chain.
-		if block.Td.Cmp(self.Td()) > 0 {
-			// chain fork
-			if block.ParentHash() != cblock.Hash() {
-				// during split we merge two different chains and create the new canonical chain
-				err := self.merge(cblock, block)
-				if err != nil {
-					return i, err
-				}
-
-				queue[i] = ChainSplitEvent{block, logs}
-				queueEvent.splitCount++
-			}
-
-			self.mu.Lock()
-			self.setTotalDifficulty(block.Td)
-			self.insert(block)
-			self.mu.Unlock()
-
-			jsonlogger.LogJson(&logger.EthChainNewHead{
-				BlockHash:     block.Hash().Hex(),
-				BlockNumber:   block.Number(),
-				ChainHeadHash: cblock.Hash().Hex(),
-				BlockPrevHash: block.ParentHash().Hex(),
-			})
-
-			self.setTransState(state.New(block.Root(), self.stateDb))
-			self.txState.SetState(state.New(block.Root(), self.stateDb))
-
-			queue[i] = ChainEvent{block, block.Hash(), logs}
-			queueEvent.canonicalCount++
-
+		// write the block to the chain and get the status
+		status, err := self.WriteBlock(block)
+		if err != nil {
+			return i, err
+		}
+		switch status {
+		case canonStatTy:
 			if glog.V(logger.Debug) {
 				glog.Infof("[%v] inserted block #%d (%d TXs %d UNCs) (%x...). Took %v\n", time.Now().UnixNano(), block.Number(), len(block.Transactions()), len(block.Uncles()), block.Hash().Bytes()[0:4], time.Since(bstart))
 			}
-		} else {
+			queue[i] = ChainEvent{block, block.Hash(), logs}
+			queueEvent.canonicalCount++
+		case sideStatTy:
 			if glog.V(logger.Detail) {
 				glog.Infof("inserted forked block #%d (TD=%v) (%d TXs %d UNCs) (%x...). Took %v\n", block.Number(), block.Difficulty(), len(block.Transactions()), len(block.Uncles()), block.Hash().Bytes()[0:4], time.Since(bstart))
 			}
-
 			queue[i] = ChainSideEvent{block, logs}
 			queueEvent.sideCount++
+		case splitStatTy:
+			queue[i] = ChainSplitEvent{block, logs}
+			queueEvent.splitCount++
 		}
-		self.enqueueForWrite(block)
-		// Delete from future blocks
-		self.futureBlocks.Delete(block.Hash())
-
 		stats.processed++
-		blockInsertTimer.UpdateSince(bstart)
 	}
 
 	if (stats.queued > 0 || stats.processed > 0 || stats.ignored > 0) && bool(glog.V(logger.Info)) {
@@ -744,9 +768,9 @@ func (self *ChainManager) diff(oldBlock, newBlock *types.Block) (types.Blocks, e
 		}
 	}
 
-	if glog.V(logger.Info) {
+	if glog.V(logger.Debug) {
 		commonHash := commonBlock.Hash()
-		glog.Infof("Fork detected @ %x. Reorganising chain from #%v %x to %x", commonHash[:4], numSplit, oldStart.Hash().Bytes()[:4], newStart.Hash().Bytes()[:4])
+		glog.Infof("Chain split detected @ %x. Reorganising chain from #%v %x to %x", commonHash[:4], numSplit, oldStart.Hash().Bytes()[:4], newStart.Hash().Bytes()[:4])
 	}
 
 	return newChain, nil

miner/worker.go

@@ -233,38 +233,40 @@ func (self *worker) wait() {
 				continue
 			}
 
-			if _, err := self.chain.InsertChain(types.Blocks{block}); err == nil {
-				for _, uncle := range block.Uncles() {
-					delete(self.possibleUncles, uncle.Hash())
-				}
-				self.mux.Post(core.NewMinedBlockEvent{block})
-
-				var stale, confirm string
-				canonBlock := self.chain.GetBlockByNumber(block.NumberU64())
-				if canonBlock != nil && canonBlock.Hash() != block.Hash() {
-					stale = "stale "
-				} else {
-					confirm = "Wait 5 blocks for confirmation"
-					self.current.localMinedBlocks = newLocalMinedBlock(block.Number().Uint64(), self.current.localMinedBlocks)
-				}
-
-				glog.V(logger.Info).Infof("🔨  Mined %sblock (#%v / %x). %s", stale, block.Number(), block.Hash().Bytes()[:4], confirm)
+			_, err := self.chain.WriteBlock(block)
+			if err != nil {
+				glog.V(logger.Error).Infoln("error writing block to chain", err)
+				continue
+			}
 
-				jsonlogger.LogJson(&logger.EthMinerNewBlock{
-					BlockHash:     block.Hash().Hex(),
-					BlockNumber:   block.Number(),
-					ChainHeadHash: block.ParentHash().Hex(),
-					BlockPrevHash: block.ParentHash().Hex(),
-				})
+			// check staleness and display confirmation
+			var stale, confirm string
+			canonBlock := self.chain.GetBlockByNumber(block.NumberU64())
+			if canonBlock != nil && canonBlock.Hash() != block.Hash() {
+				stale = "stale "
 			} else {
-				self.commitNewWork()
+				confirm = "Wait 5 blocks for confirmation"
+				self.current.localMinedBlocks = newLocalMinedBlock(block.Number().Uint64(), self.current.localMinedBlocks)
 			}
+
+			glog.V(logger.Info).Infof("🔨  Mined %sblock (#%v / %x). %s", stale, block.Number(), block.Hash().Bytes()[:4], confirm)
+
+			// broadcast before waiting for validation
+			go self.mux.Post(core.NewMinedBlockEvent{block})
+
+			self.commitNewWork()
 		}
 	}
 }
 
 func (self *worker) push() {
 	if atomic.LoadInt32(&self.mining) == 1 {
+		if core.Canary(self.current.state) {
+			glog.Infoln("Toxicity levels rising to deadly levels. Your canary has died. You can go back or continue down the mineshaft --more--")
+			glog.Infoln("You turn back and abort mining")
+			return
+		}
+
 		// push new work to agents
 		for _, agent := range self.agents {
 			atomic.AddInt32(&self.atWork, 1)
@@ -369,6 +371,13 @@ func (self *worker) commitNewWork() {
 	if tstamp <= parent.Time() {
 		tstamp = parent.Time() + 1
 	}
+	// this will ensure we're not going off too far in the future
+	if now := time.Now().Unix(); tstamp > now+4 {
+		wait := time.Duration(tstamp-now) * time.Second
+		glog.V(logger.Info).Infoln("We are too far in the future. Waiting for", wait)
+		time.Sleep(wait)
+	}
+
 	num := parent.Number()
 	header := &types.Header{
 		ParentHash: parent.Hash(),
@@ -420,11 +429,13 @@ func (self *worker) commitNewWork() {
 		// commit state root after all state transitions.
 		core.AccumulateRewards(self.current.state, header, uncles)
 		current.state.Update()
+		self.current.state.Sync()
 		header.Root = current.state.Root()
 	}
 
 	// create the new block whose nonce will be mined.
 	current.block = types.NewBlock(header, current.txs, uncles, current.receipts)
+	self.current.block.Td = new(big.Int).Set(core.CalcTD(self.current.block, self.chain.GetBlock(self.current.block.ParentHash())))
 
 	// We only care about logging if we're actually mining.
 	if atomic.LoadInt32(&self.mining) == 1 {