forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 18:32:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #341 from nguyenbatam/devel

Browse source 
final for main net : fix bug and optimize performance
This commit is contained in:
Tuna 2018-12-10 17:16:57 +07:00 committed by GitHub
commit acca52510f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
25 changed files with 707 additions and 344 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/tomo/main.go
View file

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

7
cmd/utils/flags.go
View file

@ -113,6 +113,10 @@ func NewApp(gitCommit, usage string) *cli.App {
var (
// General settings
AnnounceTxsFlag = cli.BoolFlag{
Name: "announce-txs",
Usage: "Always commit transactions",
}
DataDirFlag = DirectoryFlag{
Name: "datadir",
Usage: "Data directory for the databases and keystore",
@ -897,6 +901,9 @@ func SetNodeConfig(ctx *cli.Context, cfg *node.Config) {
if ctx.GlobalIsSet(NoUSBFlag.Name) {
cfg.NoUSB = ctx.GlobalBool(NoUSBFlag.Name)
}
if ctx.GlobalIsSet(AnnounceTxsFlag.Name) {
cfg.AnnounceTxs = ctx.GlobalBool(AnnounceTxsFlag.Name)
}
}
func setGPO(ctx *cli.Context, cfg *gasprice.Config) {

2
consensus/consensus.go
View file

@ -58,7 +58,7 @@ type Engine interface {
// VerifyHeader checks whether a header conforms to the consensus rules of a
// given engine. Verifying the seal may be done optionally here, or explicitly
// via the VerifySeal method.
VerifyHeader(chain ChainReader, header *types.Header, seal bool) error
VerifyHeader(chain ChainReader, header *types.Header, fullVerify bool) error
// VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers
// concurrently. The method returns a quit channel to abort the operations and

145
consensus/posv/posv.go
View file

@ -261,20 +261,20 @@ func (c *Posv) Author(header *types.Header) (common.Address, error) {
}
// VerifyHeader checks whether a header conforms to the consensus rules.
func (c *Posv) VerifyHeader(chain consensus.ChainReader, header *types.Header, seal bool) error {
return c.verifyHeaderWithCache(chain, header, nil)
func (c *Posv) VerifyHeader(chain consensus.ChainReader, header *types.Header, fullVerify bool) error {
return c.verifyHeaderWithCache(chain, header, nil, fullVerify)
}
// VerifyHeaders is similar to VerifyHeader, but verifies a batch of headers. The
// method returns a quit channel to abort the operations and a results channel to
// retrieve the async verifications (the order is that of the input slice).
func (c *Posv) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, seals []bool) (chan<- struct{}, <-chan error) {
func (c *Posv) VerifyHeaders(chain consensus.ChainReader, headers []*types.Header, fullVerifies []bool) (chan<- struct{}, <-chan error) {
abort := make(chan struct{})
results := make(chan error, len(headers))
go func() {
for i, header := range headers {
err := c.verifyHeaderWithCache(chain, header, headers[:i])
err := c.verifyHeaderWithCache(chain, header, headers[:i], fullVerifies[i])
select {
case <-abort:
@ -286,12 +286,12 @@ func (c *Posv) VerifyHeaders(chain consensus.ChainReader, headers []*types.Heade
return abort, results
}
func (c *Posv) verifyHeaderWithCache(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Posv) verifyHeaderWithCache(chain consensus.ChainReader, header *types.Header, parents []*types.Header, fullVerify bool) error {
_, check := c.verifiedHeaders.Get(header.Hash())
if check {
return nil
}
err := c.verifyHeader(chain, header, parents)
err := c.verifyHeader(chain, header, parents, fullVerify)
if err == nil {
c.verifiedHeaders.Add(header.Hash(), true)
}
@ -302,16 +302,20 @@ func (c *Posv) verifyHeaderWithCache(chain consensus.ChainReader, header *types.
// caller may optionally pass in a batch of parents (ascending order) to avoid
// looking those up from the database. This is useful for concurrently verifying
// a batch of new headers.
func (c *Posv) verifyHeader(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Posv) verifyHeader(chain consensus.ChainReader, header *types.Header, parents []*types.Header, fullVerify bool) error {
if header.Number == nil {
return errUnknownBlock
}
number := header.Number.Uint64()
if fullVerify {
if header.Number.Uint64() > c.config.Epoch && len(header.Validator) == 0 {
return consensus.ErrNoValidatorSignature
}
// Don't waste time checking blocks from the future
if header.Time.Cmp(big.NewInt(time.Now().Unix())) > 0 {
return consensus.ErrFutureBlock
}
}
// Checkpoint blocks need to enforce zero beneficiary
checkpoint := (number % c.config.Epoch) == 0
if checkpoint && header.Coinbase != (common.Address{}) {
@ -348,25 +352,19 @@ func (c *Posv) verifyHeader(chain consensus.ChainReader, header *types.Header, p
if header.UncleHash != uncleHash {
return errInvalidUncleHash
}
// Ensure that the block's difficulty is meaningful (may not be correct at this point)
if number > 0 {
if header.Difficulty == nil || (header.Difficulty.Cmp(diffInTurn) != 0 && header.Difficulty.Cmp(diffNoTurn) != 0) {
return errInvalidDifficulty
}
}
// If all checks passed, validate any special fields for hard forks
if err := misc.VerifyForkHashes(chain.Config(), header, false); err != nil {
return err
}
// All basic checks passed, verify cascading fields
return c.verifyCascadingFields(chain, header, parents)
return c.verifyCascadingFields(chain, header, parents, fullVerify)
}
// verifyCascadingFields verifies all the header fields that are not standalone,
// rather depend on a batch of previous headers. The caller may optionally pass
// in a batch of parents (ascending order) to avoid looking those up from the
// database. This is useful for concurrently verifying a batch of new headers.
func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types.Header, parents []*types.Header, fullVerify bool) error {
// The genesis block is the always valid dead-end
number := header.Number.Uint64()
if number == 0 {
@ -385,9 +383,6 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types.
if parent.Time.Uint64()+c.config.Period > header.Time.Uint64() {
return ErrInvalidTimestamp
}
if header.Number.Uint64() > c.config.Epoch && len(header.Validator) == 0 {
return consensus.ErrNoValidatorSignature
}
// Retrieve the snapshot needed to verify this header and cache it
snap, err := c.snapshot(chain, number-1, header.ParentHash, parents)
if err != nil {
@ -429,7 +424,7 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types.
}
}
// All basic checks passed, verify the seal and return
return c.verifySeal(chain, header, parents)
return c.verifySeal(chain, header, parents, fullVerify)
}
func (c *Posv) GetSnapshot(chain consensus.ChainReader, header *types.Header) (*Snapshot, error) {
@ -482,30 +477,37 @@ func WhoIsCreator(snap *Snapshot, header *types.Header) (common.Address, error)
return m, nil
}
func YourTurn(masternodes []common.Address, snap *Snapshot, header *types.Header, cur common.Address) (int, int, bool, error) {
func (c *Posv) YourTurn(chain consensus.ChainReader, parent *types.Header, signer common.Address) (int, int, int, bool, error) {
masternodes := c.GetMasternodes(chain, parent)
snap, err := c.GetSnapshot(chain, parent)
if err != nil {
log.Error("Failed when trying to commit new work", "err", err)
return 0, -1, -1, false, err
}
if len(masternodes) == 0 {
return -1, -1, true, nil
return 0, -1, -1, false, errors.New("Masternodes not found")
}
pre := common.Address{}
// masternode[0] has chance to create block 1
var err error
preIndex := -1
if header.Number.Uint64() != 0 {
pre, err = WhoIsCreator(snap, header)
if parent.Number.Uint64() != 0 {
pre, err = WhoIsCreator(snap, parent)
if err != nil {
return 0, 0, false, err
return 0, 0, 0, false, err
}
preIndex = position(masternodes, pre)
}
curIndex := position(masternodes, cur)
log.Info("Masternodes cycle info", "number of masternodes", len(masternodes), "previous", pre, "position", preIndex, "current", cur, "position", curIndex)
curIndex := position(masternodes, signer)
if signer == c.signer {
log.Info("Masternodes cycle info", "number of masternodes", len(masternodes), "previous", pre, "position", preIndex, "current", signer, "position", curIndex)
}
for i, s := range masternodes {
fmt.Printf("%d - %s\n", i, s.String())
}
if (preIndex+1)%len(masternodes) == curIndex {
return preIndex, curIndex, true, nil
return len(masternodes), preIndex, curIndex, true, nil
}
return preIndex, curIndex, false, nil
return len(masternodes), preIndex, curIndex, false, nil
}
// snapshot retrieves the authorization snapshot at a given point in time.
@ -533,7 +535,7 @@ func (c *Posv) snapshot(chain consensus.ChainReader, number uint64, hash common.
// If we're at block zero, make a snapshot
if number == 0 {
genesis := chain.GetHeaderByNumber(0)
if err := c.VerifyHeader(chain, genesis, false); err != nil {
if err := c.VerifyHeader(chain, genesis, true); err != nil {
return nil, err
}
signers := make([]common.Address, (len(genesis.Extra)-extraVanity-extraSeal)/common.AddressLength)
@ -598,7 +600,7 @@ func (c *Posv) VerifyUncles(chain consensus.ChainReader, block *types.Block) err
// VerifySeal implements consensus.Engine, checking whether the signature contained
// in the header satisfies the consensus protocol requirements.
func (c *Posv) VerifySeal(chain consensus.ChainReader, header *types.Header) error {
return c.verifySeal(chain, header, nil)
return c.verifySeal(chain, header, nil, true)
}
// verifySeal checks whether the signature contained in the header satisfies the
@ -607,7 +609,7 @@ func (c *Posv) VerifySeal(chain consensus.ChainReader, header *types.Header) err
// from.
// verifySeal also checks the pair of creator-validator set in the header satisfies
// the double validation.
func (c *Posv) verifySeal(chain consensus.ChainReader, header *types.Header, parents []*types.Header) error {
func (c *Posv) verifySeal(chain consensus.ChainReader, header *types.Header, parents []*types.Header, fullVerify bool) error {
// Verifying the genesis block is not supported
number := header.Number.Uint64()
if number == 0 {
@ -624,6 +626,20 @@ func (c *Posv) verifySeal(chain consensus.ChainReader, header *types.Header, par
if err != nil {
return err
}
var parent *types.Header
if len(parents) > 0 {
parent = parents[len(parents)-1]
} else {
parent = chain.GetHeader(header.ParentHash, number-1)
}
difficulty := c.calcDifficulty(chain, parent, creator)
log.Debug("verify seal block", "number", header.Number, "hash", header.Hash(), "block difficulty", header.Difficulty, "calc difficulty", difficulty, "creator", creator)
// Ensure that the block's difficulty is meaningful (may not be correct at this point)
if number > 0 {
if header.Difficulty.Int64() != difficulty.Int64() {
return errInvalidDifficulty
}
}
masternodes := c.GetMasternodes(chain, header)
mstring := []string{}
for _, m := range masternodes {
@ -663,7 +679,7 @@ func (c *Posv) verifySeal(chain consensus.ChainReader, header *types.Header, par
// header must contain validator info following double validation design
// start checking from epoch 2nd.
if header.Number.Uint64() > c.config.Epoch {
if header.Number.Uint64() > c.config.Epoch && fullVerify {
validator, err := c.RecoverValidator(header)
if err != nil {
return err
@ -741,9 +757,13 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error
}
c.lock.RUnlock()
}
parent := chain.GetHeader(header.ParentHash, number-1)
if parent == nil {
return consensus.ErrUnknownAncestor
}
// Set the correct difficulty
header.Difficulty = big.NewInt(1)
header.Difficulty = c.calcDifficulty(chain, parent, c.signer)
log.Debug("CalcDifficulty ", "number", header.Number, "difficulty", header.Difficulty)
// Ensure the extra data has all it's components
if len(header.Extra) < extraVanity {
header.Extra = append(header.Extra, bytes.Repeat([]byte{0x00}, extraVanity-len(header.Extra))...)
@ -781,10 +801,7 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error
header.MixDigest = common.Hash{}
// Ensure the timestamp has the correct delay
parent := chain.GetHeader(header.ParentHash, number-1)
if parent == nil {
return consensus.ErrUnknownAncestor
}
header.Time = new(big.Int).Add(parent.Time, new(big.Int).SetUint64(c.config.Period))
if header.Time.Int64() < time.Now().Unix() {
header.Time = big.NewInt(time.Now().Unix())
@ -806,23 +823,14 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head
if err != nil {
return err
}
currentSigners := snap.GetSigners()
proposedSigners := make(map[common.Address]struct{})
// count all addresses in ms to be masternode
newSigners := make(map[common.Address]struct{})
for _, m := range ms {
proposedSigners[m.Address] = struct{}{}
snap.Signers[m.Address] = struct{}{}
}
// deactivate current masternodes which aren't in ms
for _, s := range currentSigners {
if _, ok := proposedSigners[s]; !ok {
delete(snap.Signers, s)
}
newSigners[m.Address] = struct{}{}
}
snap.Signers = newSigners
nm := []string{}
newSigners := snap.GetSigners()
for _, n := range newSigners {
nm = append(nm, n.String())
for _, n := range ms {
nm = append(nm, n.Address.String())
}
c.recents.Add(snap.Hash, snap)
log.Info("New set of masternodes has been updated to snapshot", "number", snap.Number, "hash", snap.Hash, "new masternodes", nm)
@ -936,21 +944,15 @@ func (c *Posv) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan
// that a new block should have based on the previous blocks in the chain and the
// current signer.
func (c *Posv) CalcDifficulty(chain consensus.ChainReader, time uint64, parent *types.Header) *big.Int {
snap, err := c.snapshot(chain, parent.Number.Uint64(), parent.Hash(), nil)
if err != nil {
return nil
}
return CalcDifficulty(snap, c.signer)
return c.calcDifficulty(chain, parent, c.signer)
}
// CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty
// that a new block should have based on the previous blocks in the chain and the
// current signer.
func CalcDifficulty(snap *Snapshot, signer common.Address) *big.Int {
if snap.inturn(snap.Number+1, signer) {
return new(big.Int).Set(diffInTurn)
func (c *Posv) calcDifficulty(chain consensus.ChainReader, parent *types.Header, signer common.Address) *big.Int {
len, preIndex, curIndex, _, err := c.YourTurn(chain, parent, signer)
if err != nil {
return big.NewInt(int64(len + curIndex - preIndex))
}
return new(big.Int).Set(diffNoTurn)
return big.NewInt(int64(len - Hop(len, preIndex, curIndex)))
}
// APIs implements consensus.Engine, returning the user facing RPC API to allow
@ -1065,3 +1067,14 @@ func ExtractValidatorsFromBytes(byteValidators []byte) []int64 {
return validators
}
func Hop(len, pre, cur int) int {
switch {
case pre < cur:
return cur - (pre + 1)
case pre > cur:
return (len - pre) + (cur - 1)
default:
return len - 1
}
}

236
core/blockchain.go
View file

@ -75,6 +75,13 @@ type CacheConfig struct {
TrieNodeLimit int // Memory limit (MB) at which to flush the current in-memory trie to disk
TrieTimeLimit time.Duration // Time limit after which to flush the current in-memory trie to disk
}
type ResultProcessBlock struct {
logs []*types.Log
receipts []*types.Receipt
state *state.StateDB
proctime time.Duration
usedGas uint64
}
// BlockChain represents the canonical chain given a database with a genesis
// block. The Blockchain manages chain imports, reverts, chain reorganisations.
@ -120,7 +127,9 @@ type BlockChain struct {
bodyRLPCache *lru.Cache // Cache for the most recent block bodies in RLP encoded format
blockCache *lru.Cache // Cache for the most recent entire blocks
futureBlocks *lru.Cache // future blocks are blocks added for later processing
resultProcess *lru.Cache // Cache for processed blocks
calculatingBlock *lru.Cache // Cache for processing blocks
downloadingBlock *lru.Cache // Cache for downloading blocks (avoid duplication from fetcher)
quit chan struct{} // blockchain quit channel
running int32 // running must be called atomically
// procInterrupt must be atomically called
@ -152,7 +161,9 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
blockCache, _ := lru.New(blockCacheLimit)
futureBlocks, _ := lru.New(maxFutureBlocks)
badBlocks, _ := lru.New(badBlockLimit)
resultProcess, _ := lru.New(blockCacheLimit)
preparingBlock, _ := lru.New(blockCacheLimit)
downloadingBlock, _ := lru.New(blockCacheLimit)
bc := &BlockChain{
chainConfig: chainConfig,
cacheConfig: cacheConfig,
@ -164,6 +175,9 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
bodyRLPCache: bodyRLPCache,
blockCache: blockCache,
futureBlocks: futureBlocks,
resultProcess: resultProcess,
calculatingBlock: preparingBlock,
downloadingBlock: downloadingBlock,
engine: engine,
vmConfig: vmConfig,
badBlocks: badBlocks,
@ -892,7 +906,7 @@ func (bc *BlockChain) WriteBlockWithState(block *types.Block, receipts []*types.
defer bc.mu.Unlock()
currentBlock := bc.CurrentBlock()
//localTd := bc.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
localTd := bc.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
externTd := new(big.Int).Add(block.Difficulty(), ptd)
// Irrelevant of the canonical status, write the block itself to the database
@ -966,8 +980,12 @@ func (bc *BlockChain) WriteBlockWithState(block *types.Block, receipts []*types.
// If the total difficulty is higher than our known, add it to the canonical chain
// Second clause in the if statement reduces the vulnerability to selfish mining.
// Please refer to http://www.cs.cornell.edu/~ie53/publications/btcProcFC.pdf
reorg := block.NumberU64() > currentBlock.NumberU64()
reorg := externTd.Cmp(localTd) > 0
currentBlock = bc.CurrentBlock()
if !reorg && externTd.Cmp(localTd) == 0 {
// Split same-difficulty blocks by number
reorg = block.NumberU64() > currentBlock.NumberU64()
}
if reorg {
// Reorganise the chain if the parent is not the head block
if block.ParentHash() != currentBlock.Hash() {
@ -1049,6 +1067,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
for i, block := range chain {
headers[i] = block.Header()
seals[i] = true
bc.downloadingBlock.Add(block.Hash(), true)
}
abort, results := bc.engine.VerifyHeaders(bc, headers, seals)
defer close(abort)
@ -1168,7 +1187,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
}
switch status {
case CanonStatTy:
log.Debug("Inserted new block", "number", block.Number(), "hash", block.Hash(), "uncles", len(block.Uncles()),
log.Debug("Inserted new block from downloader", "number", block.Number(), "hash", block.Hash(), "uncles", len(block.Uncles()),
"txs", len(block.Transactions()), "gas", block.GasUsed(), "elapsed", common.PrettyDuration(time.Since(bstart)))
coalescedLogs = append(coalescedLogs, logs...)
@ -1180,7 +1199,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
bc.gcproc += proctime
case SideStatTy:
log.Debug("Inserted forked block", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed",
log.Debug("Inserted forked block from downloader", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed",
common.PrettyDuration(time.Since(bstart)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()))
blockInsertTimer.UpdateSince(bstart)
@ -1189,7 +1208,7 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
stats.processed++
stats.usedGas += usedGas
stats.report(chain, i, bc.stateCache.TrieDB().Size())
if bc.chainConfig.Posv != nil {
if status == CanonStatTy && bc.chainConfig.Posv != nil {
// epoch block
if (chain[i].NumberU64() % bc.chainConfig.Posv.Epoch) == 0 {
CheckpointCh <- 1
@ -1206,11 +1225,212 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
}
// Append a single chain head event if we've progressed the chain
if lastCanon != nil && bc.CurrentBlock().Hash() == lastCanon.Hash() {
log.Debug("New ChainHeadEvent ", "number", lastCanon.NumberU64(), "hash", lastCanon.Hash())
events = append(events, ChainHeadEvent{lastCanon})
}
return 0, events, coalescedLogs, nil
}
func (bc *BlockChain) InsertBlock(block *types.Block) error {
events, logs, err := bc.insertBlock(block)
bc.PostChainEvents(events, logs)
return err
}
func (bc *BlockChain) PrepareBlock(block *types.Block) (err error) {
defer log.Debug("Done prepare block ", "number", block.NumberU64(), "hash", block.Hash(), "validator", block.Header().Validator, "err", err)
if _, check := bc.resultProcess.Get(block.Hash()); check {
log.Debug("Stop prepare a block because the result cached", "number", block.NumberU64(), "hash", block.Hash(), "validator", block.Header().Validator)
return nil
}
if _, check := bc.calculatingBlock.Get(block.Hash()); check {
log.Debug("Stop prepare a block because inserting", "number", block.NumberU64(), "hash", block.Hash(), "validator", block.Header().Validator)
return nil
}
err = bc.engine.VerifyHeader(bc, block.Header(), false)
if err != nil {
return err
}
result, err := bc.getResultBlock(block, false)
if err == nil {
bc.resultProcess.Add(block.Hash(), result)
return nil
} else if err == ErrKnownBlock {
return nil
} else if err == ErrStopPreparingBlock {
log.Debug("Stop prepare a block because calculating", "number", block.NumberU64(), "hash", block.Hash(), "validator", block.Header().Validator)
return nil
}
return err
}
func (bc *BlockChain) getResultBlock(block *types.Block, verifiedM2 bool) (*ResultProcessBlock, error) {
var calculatedBlock *CalculatedBlock
if verifiedM2 {
if result, check := bc.resultProcess.Get(block.HashNoValidator()); check {
log.Debug("Get result block from cache ", "number", block.NumberU64(), "hash", block.Hash(), "hash no validator", block.HashNoValidator())
return result.(*ResultProcessBlock), nil
}
log.Debug("Not found cache prepare block ", "number", block.NumberU64(), "hash", block.Hash(), "validator", block.HashNoValidator())
if calculatedBlock, _ := bc.calculatingBlock.Get(block.HashNoValidator()); calculatedBlock != nil {
calculatedBlock.(*CalculatedBlock).stop = true
}
}
calculatedBlock = &CalculatedBlock{block, false}
bc.calculatingBlock.Add(block.HashNoValidator(), calculatedBlock)
// Start the parallel header verifier
// If the chain is terminating, stop processing blocks
if atomic.LoadInt32(&bc.procInterrupt) == 1 {
log.Debug("Premature abort during blocks processing")
return nil, ErrBlacklistedHash
}
// If the header is a banned one, straight out abort
if BadHashes[block.Hash()] {
bc.reportBlock(block, nil, ErrBlacklistedHash)
return nil, ErrBlacklistedHash
}
// Wait for the block's verification to complete
bstart := time.Now()
err := bc.Validator().ValidateBody(block)
switch {
case err == ErrKnownBlock:
// Block and state both already known. However if the current block is below
// this number we did a rollback and we should reimport it nonetheless.
if bc.CurrentBlock().NumberU64() >= block.NumberU64() {
return nil, ErrKnownBlock
}
case err == consensus.ErrPrunedAncestor:
// Block competing with the canonical chain, store in the db, but don't process
// until the competitor TD goes above the canonical TD
currentBlock := bc.CurrentBlock()
localTd := bc.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
externTd := new(big.Int).Add(bc.GetTd(block.ParentHash(), block.NumberU64()-1), block.Difficulty())
if localTd.Cmp(externTd) > 0 {
return nil, err
}
// Competitor chain beat canonical, gather all blocks from the common ancestor
var winner []*types.Block
parent := bc.GetBlock(block.ParentHash(), block.NumberU64()-1)
for !bc.HasState(parent.Root()) {
winner = append(winner, parent)
parent = bc.GetBlock(parent.ParentHash(), parent.NumberU64()-1)
}
for j := 0; j < len(winner)/2; j++ {
winner[j], winner[len(winner)-1-j] = winner[len(winner)-1-j], winner[j]
}
log.Debug("Number block need calculated again", "number", block.NumberU64(), "hash", block.Hash().Hex(), "winners", len(winner))
// Import all the pruned blocks to make the state available
_, _, _, err := bc.insertChain(winner)
if err != nil {
return nil, err
}
case err != nil:
bc.reportBlock(block, nil, err)
return nil, err
}
// Create a new statedb using the parent block and report an
// error if it fails.
var parent = bc.GetBlock(block.ParentHash(), block.NumberU64()-1)
state, err := state.New(parent.Root(), bc.stateCache)
if err != nil {
return nil, err
}
// Process block using the parent state as reference point.
receipts, logs, usedGas, err := bc.processor.ProcessBlockNoValidator(calculatedBlock, state, bc.vmConfig)
process := time.Since(bstart)
if err != nil {
if err != ErrStopPreparingBlock {
bc.reportBlock(block, receipts, err)
}
return nil, err
}
// Validate the state using the default validator
err = bc.Validator().ValidateState(block, parent, state, receipts, usedGas)
if err != nil {
bc.reportBlock(block, receipts, err)
return nil, err
}
proctime := time.Since(bstart)
log.Debug("Calculate new block", "number", block.Number(), "hash", block.Hash(), "uncles", len(block.Uncles()),
"txs", len(block.Transactions()), "gas", block.GasUsed(), "elapsed", common.PrettyDuration(time.Since(bstart)), "process", process)
return &ResultProcessBlock{receipts: receipts, logs: logs, state: state, proctime: proctime, usedGas: usedGas}, nil
}
// insertChain will execute the actual chain insertion and event aggregation. The
// only reason this method exists as a separate one is to make locking cleaner
// with deferred statements.
func (bc *BlockChain) insertBlock(block *types.Block) ([]interface{}, []*types.Log, error) {
var (
stats = insertStats{startTime: mclock.Now()}
events = make([]interface{}, 0, 1)
coalescedLogs []*types.Log
)
if _, check := bc.downloadingBlock.Get(block.Hash()); check {
log.Debug("Stop fetcher a block because downloading", "number", block.NumberU64(), "hash", block.Hash())
return events, coalescedLogs, nil
}
result, err := bc.getResultBlock(block, true)
if err != nil {
return events, coalescedLogs, err
}
defer bc.resultProcess.Remove(block.HashNoValidator())
bc.wg.Add(1)
defer bc.wg.Done()
// Write the block to the chain and get the status.
bc.chainmu.Lock()
defer bc.chainmu.Unlock()
if bc.HasBlockAndState(block.Hash(), block.NumberU64()) {
return events, coalescedLogs, nil
}
status, err := bc.WriteBlockWithState(block, result.receipts, result.state)
if err != nil {
return events, coalescedLogs, err
}
switch status {
case CanonStatTy:
log.Debug("Inserted new block from fetcher", "number", block.Number(), "hash", block.Hash(), "uncles", len(block.Uncles()),
"txs", len(block.Transactions()), "gas", block.GasUsed(), "elapsed", common.PrettyDuration(time.Since(block.ReceivedAt)))
coalescedLogs = append(coalescedLogs, result.logs...)
events = append(events, ChainEvent{block, block.Hash(), result.logs})
// Only count canonical blocks for GC processing time
bc.gcproc += result.proctime
case SideStatTy:
log.Debug("Inserted forked block from fetcher", "number", block.Number(), "hash", block.Hash(), "diff", block.Difficulty(), "elapsed",
common.PrettyDuration(time.Since(block.ReceivedAt)), "txs", len(block.Transactions()), "gas", block.GasUsed(), "uncles", len(block.Uncles()))
blockInsertTimer.Update(result.proctime)
events = append(events, ChainSideEvent{block})
}
stats.processed++
stats.usedGas += result.usedGas
stats.report(types.Blocks{block}, 0, bc.stateCache.TrieDB().Size())
if status == CanonStatTy && bc.chainConfig.Posv != nil {
// epoch block
if (block.NumberU64() % bc.chainConfig.Posv.Epoch) == 0 {
CheckpointCh <- 1
}
// prepare set of masternodes for the next epoch
if (block.NumberU64() % bc.chainConfig.Posv.Epoch) == (bc.chainConfig.Posv.Epoch - bc.chainConfig.Posv.Gap) {
err := bc.UpdateM1()
if err != nil {
log.Error("Error when update masternodes set. Stopping node", "err", err)
os.Exit(1)
}
}
}
// Append a single chain head event if we've progressed the chain
if status == CanonStatTy && bc.CurrentBlock().Hash() == block.Hash() {
events = append(events, ChainHeadEvent{block})
log.Debug("New ChainHeadEvent from fetcher ", "number", block.NumberU64(), "hash", block.Hash())
}
return events, coalescedLogs, nil
}
// insertStats tracks and reports on block insertion.
type insertStats struct {
queued, processed, ignored int

2
core/error.go
View file

@ -36,4 +36,6 @@ var (
ErrNotPoSV = errors.New("Posv not found in config")
ErrNotFoundM1 = errors.New("list M1 not found ")
ErrStopPreparingBlock = errors.New("stop calculating a block not verified by M2")
)

43
core/state_processor.go
View file

@ -40,6 +40,10 @@ type StateProcessor struct {
bc *BlockChain // Canonical block chain
engine consensus.Engine // Consensus engine used for block rewards
}
type CalculatedBlock struct {
block *types.Block
stop bool
}
// NewStateProcessor initialises a new StateProcessor.
func NewStateProcessor(config *params.ChainConfig, bc *BlockChain, engine consensus.Engine) *StateProcessor {
@ -69,9 +73,7 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg
if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 {
misc.ApplyDAOHardFork(statedb)
}
InitSignerInTransactions(p.config, header, block.Transactions())
// Iterate over and process the individual transactions
for i, tx := range block.Transactions() {
statedb.Prepare(tx.Hash(), block.Hash(), i)
receipt, _, err := ApplyTransaction(p.config, p.bc, nil, gp, statedb, header, tx, usedGas, cfg)
@ -83,7 +85,44 @@ func (p *StateProcessor) Process(block *types.Block, statedb *state.StateDB, cfg
}
// Finalize the block, applying any consensus engine specific extras (e.g. block rewards)
p.engine.Finalize(p.bc, header, statedb, block.Transactions(), block.Uncles(), receipts)
return receipts, allLogs, *usedGas, nil
}
func (p *StateProcessor) ProcessBlockNoValidator(cBlock *CalculatedBlock, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, error) {
block := cBlock.block
var (
receipts types.Receipts
usedGas = new(uint64)
header = block.Header()
allLogs []*types.Log
gp = new(GasPool).AddGas(block.GasLimit())
)
// Mutate the the block and state according to any hard-fork specs
if p.config.DAOForkSupport && p.config.DAOForkBlock != nil && p.config.DAOForkBlock.Cmp(block.Number()) == 0 {
misc.ApplyDAOHardFork(statedb)
}
if cBlock.stop {
return nil, nil, 0, ErrStopPreparingBlock
}
InitSignerInTransactions(p.config, header, block.Transactions())
if cBlock.stop {
return nil, nil, 0, ErrStopPreparingBlock
}
// Iterate over and process the individual transactions
receipts = make([]*types.Receipt, block.Transactions().Len())
for i, tx := range block.Transactions() {
statedb.Prepare(tx.Hash(), block.Hash(), i)
receipt, _, err := ApplyTransaction(p.config, p.bc, nil, gp, statedb, header, tx, usedGas, cfg)
if err != nil {
return nil, nil, 0, err
}
if cBlock.stop {
return nil, nil, 0, ErrStopPreparingBlock
}
receipts[i] = receipt
}
// Finalize the block, applying any consensus engine specific extras (e.g. block rewards)
p.engine.Finalize(p.bc, header, statedb, block.Transactions(), block.Uncles(), receipts)
return receipts, allLogs, *usedGas, nil
}

29
core/tx_pool.go
View file

@ -193,7 +193,6 @@ type TxPool struct {
chain blockChain
gasPrice *big.Int
txFeed event.Feed
specialTxFeed event.Feed
scope event.SubscriptionScope
chainHeadCh chan ChainHeadEvent
chainHeadSub event.Subscription
@ -458,12 +457,6 @@ func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxPreEvent) event.Subscription
return pool.scope.Track(pool.txFeed.Subscribe(ch))
}
// SubscribeSpecialTxPreEvent registers a subscription of TxPreEvent and
// starts sending event to the given channel.
func (pool *TxPool) SubscribeSpecialTxPreEvent(ch chan<- TxPreEvent) event.Subscription {
return pool.scope.Track(pool.specialTxFeed.Subscribe(ch))
}
// GasPrice returns the current gas price enforced by the transaction pool.
func (pool *TxPool) GasPrice() *big.Int {
pool.mu.RLock()
@ -654,11 +647,12 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
return false, err
}
from, _ := types.Sender(pool.signer, tx) // already validated
if tx.IsSpecialTransaction() && pool.IsMasterNode != nil && pool.IsMasterNode(from) {
if tx.IsSpecialTransaction() && pool.IsMasterNode != nil && pool.IsMasterNode(from) && pool.pendingState.GetNonce(from) == tx.Nonce() {
return pool.promoteSpecialTx(from, tx)
}
// If the transaction pool is full, discard underpriced transactions
if uint64(len(pool.all)) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
log.Debug("Add transaction to pool full", "hash", hash, "nonce", tx.Nonce())
// If the new transaction is underpriced, don't accept it
if pool.priced.Underpriced(tx, pool.locals) {
log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
@ -819,21 +813,7 @@ func (pool *TxPool) promoteSpecialTx(addr common.Address, tx *types.Transaction)
// Set the potentially new pending nonce and notify any subsystems of the new tx
pool.beats[addr] = time.Now()
pool.pendingState.SetNonce(addr, tx.Nonce()+1)
broadcastTxs := types.Transactions{}
for i := tx.Nonce() - 1; i > 0; i-- {
before := list.txs.Get(i)
if before == nil || before.IsSpecialTransaction() {
break
}
broadcastTxs = append(broadcastTxs, before)
}
broadcastTxs = append(broadcastTxs, tx)
go func() {
for _, btx := range broadcastTxs {
pool.specialTxFeed.Send(TxPreEvent{btx})
log.Trace("Pooled new special transaction", "hash", tx.Hash(), "from", addr, "to", tx.To(), "nonce", tx.Nonce())
}
}()
go pool.txFeed.Send(TxPreEvent{tx})
return true, nil
}
@ -887,9 +867,6 @@ func (pool *TxPool) addTx(tx *types.Transaction, local bool) error {
// addTxs attempts to queue a batch of transactions if they are valid.
func (pool *TxPool) addTxs(txs []*types.Transaction, local bool) []error {
for _, tx := range txs {
types.CacheSigner(pool.signer, tx)
}
pool.mu.Lock()
defer pool.mu.Unlock()

1
core/types.go
View file

@ -43,4 +43,5 @@ type Validator interface {
// failed.
type Processor interface {
Process(block *types.Block, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, error)
ProcessBlockNoValidator(block *CalculatedBlock, statedb *state.StateDB, cfg vm.Config) (types.Receipts, []*types.Log, uint64, error)
}

27
core/types/block.go
View file

@ -125,6 +125,30 @@ func (h *Header) HashNoNonce() common.Hash {
})
}
// HashNoNonce returns the hash which is used as input for the proof-of-work search.
func (h *Header) HashNoValidator() common.Hash {
return rlpHash([]interface{}{
h.ParentHash,
h.UncleHash,
h.Coinbase,
h.Root,
h.TxHash,
h.ReceiptHash,
h.Bloom,
h.Difficulty,
h.Number,
h.GasLimit,
h.GasUsed,
h.Time,
h.Extra,
h.MixDigest,
h.Nonce,
h.Validators,
[]byte{},
h.Penalties,
})
}
// Size returns the approximate memory used by all internal contents. It is used
// to approximate and limit the memory consumption of various caches.
func (h *Header) Size() common.StorageSize {
@ -337,6 +361,9 @@ func (b *Block) Body() *Body { return &Body{b.transactions, b.uncles} }
func (b *Block) HashNoNonce() common.Hash {
return b.header.HashNoNonce()
}
func (b *Block) HashNoValidator() common.Hash {
return b.header.HashNoValidator()
}
// Size returns the true RLP encoded storage size of the block, either by encoding
// and returning it, or returning a previsouly cached value.

2
core/types/transaction_test.go
View file

@ -144,7 +144,7 @@ func TestTransactionPriceNonceSort(t *testing.T) {
}
}
// Sort the transactions and cross check the nonce ordering
txset, _ := NewTransactionsByPriceAndNonce(signer, groups,nil)
txset, _ := NewTransactionsByPriceAndNonce(signer, groups, nil)
txs := Transactions{}
for tx := txset.Peek(); tx != nil; tx = txset.Peek() {

21
eth/backend.go
View file

@ -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().AnnounceTxs)
eth.miner.SetExtra(makeExtraData(config.ExtraData))
eth.ApiBackend = &EthApiBackend{eth, nil}
@ -203,29 +203,28 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
return nil
}
appendM2HeaderHook := func(block *types.Block) (*types.Block, error) {
appendM2HeaderHook := func(block *types.Block) (*types.Block, bool, error) {
eb, err := eth.Etherbase()
if err != nil {
log.Error("Cannot get etherbase for append m2 header", "err", err)
return block, fmt.Errorf("etherbase missing: %v", err)
return block, false, fmt.Errorf("etherbase missing: %v", err)
}
m1, err := c.RecoverSigner(block.Header())
if err != nil {
return block, fmt.Errorf("can't get block creator: %v", err)
return block, false, fmt.Errorf("can't get block creator: %v", err)
}
m2, err := c.GetValidator(m1, eth.blockchain, block.Header())
if err != nil {
return block, fmt.Errorf("can't get block validator: %v", err)
return block, false, fmt.Errorf("can't get block validator: %v", err)
}
if m2 == eb {
wallet, _ := eth.accountManager.Find(accounts.Account{Address: eb})
header := block.Header()
sighash, _ := wallet.SignHash(accounts.Account{Address: eb}, posv.SigHash(header).Bytes())
header.Validator = sighash
block = types.NewBlockWithHeader(header).WithBody(block.Transactions(), block.Uncles())
return types.NewBlockWithHeader(header).WithBody(block.Transactions(), block.Uncles()), true, nil
}
return block, nil
return block, false, nil
}
eth.protocolManager.fetcher.SetSignHook(signHook)
@ -301,8 +300,8 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
if foudationWalletAddr == (common.Address{}) {
log.Error("Foundation Wallet Address is empty", "error", foudationWalletAddr)
}
start := time.Now()
if number > 0 && number-rCheckpoint > 0 && foudationWalletAddr != (common.Address{}) {
start := time.Now()
// Get signers in blockSigner smartcontract.
addr := common.HexToAddress(common.BlockSigners)
// Get reward inflation.
@ -334,8 +333,8 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
}
}
}
}
log.Debug("Time Calculated HookReward ", "block", header.Number.Uint64(), "time", common.PrettyDuration(time.Since(start)))
}
return nil
}
@ -359,7 +358,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
currentHeader := eth.blockchain.CurrentHeader()
snap, err := c.GetSnapshot(eth.blockchain, currentHeader)
if err != nil {
log.Error("Can't get snap shot with current header ", "number", currentHeader.Number, "hash", currentHeader.Hash().Hex())
log.Error("Can't get snapshot with current header ", "number", currentHeader.Number, "hash", currentHeader.Hash().Hex())
return false
}
if _, ok := snap.Signers[address]; ok {

6
eth/downloader/downloader.go
View file

@ -47,10 +47,10 @@ var (
MaxForkAncestry = 3 * params.EpochDuration // Maximum chain reorganisation
rttMinEstimate = 2 * time.Second // Minimum round-trip time to target for download requests
rttMaxEstimate = 20 * time.Second // Maximum rount-trip time to target for download requests
rttMaxEstimate = 5 * time.Second // Maximum rount-trip time to target for download requests
rttMinConfidence = 0.1 // Worse confidence factor in our estimated RTT value
ttlScaling = 3 // Constant scaling factor for RTT -> TTL conversion
ttlLimit = time.Minute // Maximum TTL allowance to prevent reaching crazy timeouts
ttlScaling = 2 // Constant scaling factor for RTT -> TTL conversion
ttlLimit = 5 * time.Second // Maximum TTL allowance to prevent reaching crazy timeouts
qosTuningPeers = 5 // Number of peers to tune based on (best peers)
qosConfidenceCap = 10 // Number of peers above which not to modify RTT confidence

2
eth/downloader/queue.go
View file

@ -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()
}

53
eth/fetcher/fetcher.go
View file

@ -62,8 +62,10 @@ type blockBroadcasterFn func(block *types.Block, propagate bool)
// chainHeightFn is a callback type to retrieve the current chain height.
type chainHeightFn func() uint64
// chainInsertFn is a callback type to insert a batch of blocks into the local chain.
type chainInsertFn func(blocks types.Blocks) (int, error)
// blockInsertFn is a callback type to insert a batch of blocks into the local chain.
type blockInsertFn func(block *types.Block) error
type blockPrepareFn func(block *types.Block) error
// peerDropFn is a callback type for dropping a peer detected as malicious.
type peerDropFn func(id string)
@ -135,7 +137,8 @@ type Fetcher struct {
verifyHeader headerVerifierFn // Checks if a block's headers have a valid proof of work
broadcastBlock blockBroadcasterFn // Broadcasts a block to connected peers
chainHeight chainHeightFn // Retrieves the current chain's height
insertChain chainInsertFn // Injects a batch of blocks into the chain
insertBlock blockInsertFn // Injects a batch of blocks into the chain
prepareBlock blockPrepareFn
dropPeer peerDropFn // Drops a peer for misbehaving
// Testing hooks
@ -144,11 +147,11 @@ type Fetcher struct {
fetchingHook func([]common.Hash) // Method to call upon starting a block (eth/61) or header (eth/62) fetch
completingHook func([]common.Hash) // Method to call upon starting a block body fetch (eth/62)
signHook func(*types.Block) error
appendM2HeaderHook func(*types.Block) (*types.Block, error)
appendM2HeaderHook func(*types.Block) (*types.Block, bool, error)
}
// New creates a block fetcher to retrieve blocks based on hash announcements.
func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertChain chainInsertFn, dropPeer peerDropFn) *Fetcher {
func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBlock blockBroadcasterFn, chainHeight chainHeightFn, insertBlock blockInsertFn, prepareBlock blockPrepareFn, dropPeer peerDropFn) *Fetcher {
knownBlocks, _ := lru.NewARC(blockLimit)
return &Fetcher{
notify: make(chan *announce),
@ -171,7 +174,8 @@ func New(getBlock blockRetrievalFn, verifyHeader headerVerifierFn, broadcastBloc
verifyHeader: verifyHeader,
broadcastBlock: broadcastBlock,
chainHeight: chainHeight,
insertChain: insertChain,
insertBlock: insertBlock,
prepareBlock: prepareBlock,
dropPeer: dropPeer,
}
}
@ -605,7 +609,7 @@ func (f *Fetcher) rescheduleComplete(complete *time.Timer) {
func (f *Fetcher) enqueue(peer string, block *types.Block) {
hash := block.Hash()
if f.knowns.Contains(hash) {
log.Debug("Discarded propagated block, known block", "peer", peer, "number", block.Number(), "hash", hash, "limit", blockLimit)
log.Trace("Discarded propagated block, known block", "peer", peer, "number", block.Number(), "hash", hash, "limit", blockLimit)
return
}
// Ensure the peer isn't DOSing us
@ -657,40 +661,56 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
log.Debug("Unknown parent of propagated block", "peer", peer, "number", block.Number(), "hash", hash, "parent", block.ParentHash())
return
}
fastBroadCast := true
again:
err := f.verifyHeader(block.Header())
// Quickly validate the header and propagate the block if it passes
switch err := f.verifyHeader(block.Header()); err {
switch err {
case nil:
// All ok, quickly propagate to our peers
propBroadcastOutTimer.UpdateSince(block.ReceivedAt)
if fastBroadCast {
go f.broadcastBlock(block, true)
}
case consensus.ErrFutureBlock:
delay := time.Unix(block.Time().Int64(), 0).Sub(time.Now()) // nolint: gosimple
time.Sleep(delay)
log.Info("Receive future block", "number", block.NumberU64(), "hash", block.Hash().Hex(), "delay", delay)
time.Sleep(delay)
goto again
case consensus.ErrNoValidatorSignature:
newBlock := block
var errM2 error
isM2 := false
if f.appendM2HeaderHook != nil {
if newBlock, err = f.appendM2HeaderHook(block); err != nil {
log.Error("Append m2 to block header fail", "err", err)
if newBlock, isM2, errM2 = f.appendM2HeaderHook(block); errM2 != nil {
log.Error("Append m2 to block header fail", "err", errM2)
return
}
}
if newBlock.Hash() == block.Hash() {
if !isM2 {
go f.broadcastBlock(block, true)
if err := f.prepareBlock(block); err != nil {
log.Debug("Propagated block prepare failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
return
}
return
}
log.Debug("Append M2 to header block", "numer", block.NumberU64(), "hahs", block.Hash())
if err := f.prepareBlock(block); err != nil {
log.Debug("Propagated block prepare failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
return
}
block = newBlock
fastBroadCast = false
goto again
default:
// Something went very wrong, drop the peer
log.Debug("Propagated block verification failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
f.dropPeer(peer)
return
}
// Run the actual import and log any issues
if _, err := f.insertChain(types.Blocks{block}); err != nil {
if err := f.insertBlock(block); err != nil {
log.Debug("Propagated block import failed", "peer", peer, "number", block.Number(), "hash", hash, "err", err)
return
}
@ -703,8 +723,9 @@ func (f *Fetcher) insert(peer string, block *types.Block) {
}
// If import succeeded, broadcast the block
propAnnounceOutTimer.UpdateSince(block.ReceivedAt)
if !fastBroadCast {
go f.broadcastBlock(block, true)
go f.broadcastBlock(block, false)
}
}()
}
@ -768,6 +789,6 @@ func (f *Fetcher) SetSignHook(signHook func(*types.Block) error) {
}
// Bind append m2 to block header hook when imported into chain.
func (f *Fetcher) SetAppendM2HeaderHook(appendM2HeaderHook func(*types.Block) (*types.Block, error)) {
func (f *Fetcher) SetAppendM2HeaderHook(appendM2HeaderHook func(*types.Block) (*types.Block, bool, error)) {
f.appendM2HeaderHook = appendM2HeaderHook
}

35
eth/fetcher/fetcher_test.go
View file

@ -92,7 +92,7 @@ func newTester() *fetcherTester {
blocks: map[common.Hash]*types.Block{genesis.Hash(): genesis},
drops: make(map[string]bool),
}
tester.fetcher = New(tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertChain, tester.dropPeer)
tester.fetcher = New(tester.getBlock, tester.verifyHeader, tester.broadcastBlock, tester.chainHeight, tester.insertBlock, tester.prepareBlock, tester.dropPeer)
tester.fetcher.Start()
return tester
@ -123,7 +123,7 @@ func (f *fetcherTester) chainHeight() uint64 {
return f.blocks[f.hashes[len(f.hashes)-1]].NumberU64()
}
// insertChain injects a new blocks into the simulated chain.
// insertBlock injects a new blocks into the simulated chain.
func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) {
f.lock.Lock()
defer f.lock.Unlock()
@ -144,6 +144,31 @@ func (f *fetcherTester) insertChain(blocks types.Blocks) (int, error) {
return 0, nil
}
// insertBlock injects a new blocks into the simulated chain.
func (f *fetcherTester) insertBlock(block *types.Block) error {
f.lock.Lock()
defer f.lock.Unlock()
// Make sure the parent in known
if _, ok := f.blocks[block.ParentHash()]; !ok {
return errors.New("unknown parent")
}
// Discard any new blocks if the same height already exists
if block.NumberU64() <= f.blocks[f.hashes[len(f.hashes)-1]].NumberU64() {
return nil
}
// Otherwise build our current chain
f.hashes = append(f.hashes, block.Hash())
f.blocks[block.Hash()] = block
return nil
}
// insertBlock injects a new blocks into the simulated chain.
func (f *fetcherTester) prepareBlock(block *types.Block) error {
return nil
}
// dropPeer is an emulator for the peer removal, simply accumulating the various
// peers dropped by the fetcher.
func (f *fetcherTester) dropPeer(peer string) {
@ -512,9 +537,9 @@ func testImportDeduplication(t *testing.T, protocol int) {
bodyFetcher := tester.makeBodyFetcher("valid", blocks, 0)
counter := uint32(0)
tester.fetcher.insertChain = func(blocks types.Blocks) (int, error) {
atomic.AddUint32(&counter, uint32(len(blocks)))
return tester.insertChain(blocks)
tester.fetcher.insertBlock = func(block *types.Block) error {
atomic.AddUint32(&counter, uint32(1))
return tester.insertBlock(block)
}
// Instrument the fetching and imported events
fetching := make(chan []common.Hash)

73
eth/handler.go
View file

@ -20,6 +20,7 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/hashicorp/golang-lru"
"math/big"
"sync"
"sync/atomic"
@ -82,8 +83,6 @@ type ProtocolManager struct {
eventMux *event.TypeMux
txCh chan core.TxPreEvent
txSub event.Subscription
specialTxCh chan core.TxPreEvent
specialTxSub event.Subscription
minedBlockSub *event.TypeMuxSubscription
// channels for fetcher, syncer, txsyncLoop
@ -95,11 +94,13 @@ type ProtocolManager struct {
// wait group is used for graceful shutdowns during downloading
// and processing
wg sync.WaitGroup
knownTxs *lru.Cache
}
// NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
// with the ethereum network.
func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, networkId uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) {
knownTxs, _ := lru.New(maxKnownTxs)
// Create the protocol manager with the base fields
manager := &ProtocolManager{
networkId: networkId,
@ -112,6 +113,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
noMorePeers: make(chan struct{}),
txsyncCh: make(chan *txsync),
quitSync: make(chan struct{}),
knownTxs: knownTxs,
}
// Figure out whether to allow fast sync or not
if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 {
@ -168,16 +170,26 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
heighter := func() uint64 {
return blockchain.CurrentBlock().NumberU64()
}
inserter := func(blocks types.Blocks) (int, error) {
inserter := func(block *types.Block) error {
// If fast sync is running, deny importing weird blocks
if atomic.LoadUint32(&manager.fastSync) == 1 {
log.Warn("Discarded bad propagated block", "number", blocks[0].Number(), "hash", blocks[0].Hash())
return 0, nil
log.Warn("Discarded bad propagated block", "number", block.Number(), "hash", block.Hash())
return nil
}
atomic.StoreUint32(&manager.acceptTxs, 1) // Mark initial sync done on any fetcher import
return manager.blockchain.InsertChain(blocks)
return manager.blockchain.InsertBlock(block)
}
manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
prepare := func(block *types.Block) error {
// If fast sync is running, deny importing weird blocks
if atomic.LoadUint32(&manager.fastSync) == 1 {
log.Warn("Discarded bad propagated block", "number", block.Number(), "hash", block.Hash())
return nil
}
atomic.StoreUint32(&manager.acceptTxs, 1) // Mark initial sync done on any fetcher import
return manager.blockchain.PrepareBlock(block)
}
manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, prepare, manager.removePeer)
return manager, nil
}
@ -209,11 +221,6 @@ func (pm *ProtocolManager) Start(maxPeers int) {
pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh)
go pm.txBroadcastLoop()
// broadcast special transactions
pm.specialTxCh = make(chan core.TxPreEvent, txChanSize)
pm.specialTxSub = pm.txpool.SubscribeSpecialTxPreEvent(pm.specialTxCh)
go pm.specialTxBroadcastLoop()
// broadcast mined blocks
pm.minedBlockSub = pm.eventMux.Subscribe(core.NewMinedBlockEvent{})
go pm.minedBroadcastLoop()
@ -227,7 +234,6 @@ func (pm *ProtocolManager) Stop() {
log.Info("Stopping Ethereum protocol")
pm.txSub.Unsubscribe() // quits txBroadcastLoop
pm.specialTxSub.Unsubscribe() // quits specialTxBroadcastLoop
pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
// Quit the sync loop.
@ -651,17 +657,14 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
trueTD = new(big.Int).Sub(request.TD, request.Block.Difficulty())
)
// Update the peers total difficulty if better than the previous
_, td := p.Head()
currentBlock := pm.blockchain.CurrentBlock()
currentTd := pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
log.Debug("NewBlockMsg", "p", p, "number", request.Block.NumberU64(), "trueTD", trueTD, "td", td, "currentTd", currentTd)
if trueTD.Cmp(td) > 0 {
if _, td := p.Head(); trueTD.Cmp(td) > 0 {
p.SetHead(trueHead, trueTD)
// Schedule a sync if above ours. Note, this will not fire a sync for a gap of
// a singe block (as the true TD is below the propagated block), however this
// scenario should easily be covered by the fetcher.
if trueTD.Cmp(currentTd) > 0 {
currentBlock := pm.blockchain.CurrentBlock()
if trueTD.Cmp(pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())) > 0 {
go pm.synchronise(p)
}
}
@ -676,12 +679,19 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&txs); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
var unkownTxs []*types.Transaction
for i, tx := range txs {
// Validate and mark the remote transaction
if tx == nil {
return errResp(ErrDecode, "transaction %d is nil", i)
}
p.MarkTransaction(tx.Hash())
exist, _ := pm.knownTxs.ContainsOrAdd(tx.Hash(), true)
if !exist {
unkownTxs = append(unkownTxs, tx)
} else {
log.Trace("Discard known tx", "hash", tx.Hash(), "nonce", tx.Nonce(), "to", tx.To())
}
}
pm.txpool.AddRemotes(txs)
@ -735,16 +745,6 @@ func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction)
log.Trace("Broadcast transaction", "hash", hash, "recipients", len(peers))
}
func (pm *ProtocolManager) BroadcastSpecialTx(hash common.Hash, tx *types.Transaction) {
// Broadcast transaction to a batch of peers not knowing about it
peers := pm.peers.PeersWithoutTx(hash)
//FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))]
for _, peer := range peers {
peer.SendSpecialTransactions(tx)
}
log.Trace("Broadcast special transaction", "hash", hash, "recipients", len(peers))
}
// Mined broadcast loop
func (self *ProtocolManager) minedBroadcastLoop() {
// automatically stops if unsubscribe
@ -752,7 +752,7 @@ func (self *ProtocolManager) minedBroadcastLoop() {
switch ev := obj.Data.(type) {
case core.NewMinedBlockEvent:
self.BroadcastBlock(ev.Block, true) // First propagate block to peers
self.BroadcastBlock(ev.Block, false) // Only then announce to the rest
//self.BroadcastBlock(ev.Block, false) // Only then announce to the rest
}
}
}
@ -770,19 +770,6 @@ func (self *ProtocolManager) txBroadcastLoop() {
}
}
func (self *ProtocolManager) specialTxBroadcastLoop() {
for {
select {
case event := <-self.specialTxCh:
self.BroadcastSpecialTx(event.Tx.Hash(), event.Tx)
// Err() channel will be closed when unsubscribing.
case <-self.specialTxSub.Err():
return
}
}
}
// NodeInfo represents a short summary of the Ethereum sub-protocol metadata
// known about the host peer.
type NodeInfo struct {

4
eth/helper_test.go
View file

@ -128,10 +128,6 @@ func (p *testTxPool) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscr
return p.txFeed.Subscribe(ch)
}
func (p *testTxPool) SubscribeSpecialTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
return p.txFeed.Subscribe(ch)
}
// newTestTransaction create a new dummy transaction.
func newTestTransaction(from *ecdsa.PrivateKey, nonce uint64, datasize int) *types.Transaction {
tx := types.NewTransaction(nonce, common.Address{}, big.NewInt(0), 100000, big.NewInt(0), make([]byte, datasize))

56
eth/peer.go
View file

@ -25,7 +25,6 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
"gopkg.in/fatih/set.v0"
@ -141,15 +140,6 @@ func (p *peer) SendTransactions(txs types.Transactions) error {
return p2p.Send(p.rw, TxMsg, txs)
}
func (p *peer) SendSpecialTransactions(tx *types.Transaction) error {
p.knownTxs.Add(tx.Hash())
if p.pairRw != nil {
return p2p.Send(p.pairRw, TxMsg, types.Transactions{tx})
} else {
return p2p.Send(p.rw, TxMsg, types.Transactions{tx})
}
}
// SendNewBlockHashes announces the availability of a number of blocks through
// a hash notification.
func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error {
@ -168,81 +158,123 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error
func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error {
p.knownBlocks.Add(block.Hash())
if p.pairRw != nil {
log.Trace("p2p send new block to the pairRw connection", "p", p, "number", block.NumberU64())
return p2p.Send(p.pairRw, NewBlockMsg, []interface{}{block, td})
} else {
return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td})
}
}
// SendBlockHeaders sends a batch of block headers to the remote peer.
func (p *peer) SendBlockHeaders(headers []*types.Header) error {
if p.pairRw != nil {
return p2p.Send(p.pairRw, BlockHeadersMsg, headers)
} else {
return p2p.Send(p.rw, BlockHeadersMsg, headers)
}
}
// SendBlockBodies sends a batch of block contents to the remote peer.
func (p *peer) SendBlockBodies(bodies []*blockBody) error {
if p.pairRw != nil {
return p2p.Send(p.pairRw, BlockBodiesMsg, blockBodiesData(bodies))
} else {
return p2p.Send(p.rw, BlockBodiesMsg, blockBodiesData(bodies))
}
}
// SendBlockBodiesRLP sends a batch of block contents to the remote peer from
// an already RLP encoded format.
func (p *peer) SendBlockBodiesRLP(bodies []rlp.RawValue) error {
if p.pairRw != nil {
return p2p.Send(p.pairRw, BlockBodiesMsg, bodies)
} else {
return p2p.Send(p.rw, BlockBodiesMsg, bodies)
}
}
// SendNodeDataRLP sends a batch of arbitrary internal data, corresponding to the
// hashes requested.
func (p *peer) SendNodeData(data [][]byte) error {
if p.pairRw != nil {
return p2p.Send(p.pairRw, NodeDataMsg, data)
} else {
return p2p.Send(p.rw, NodeDataMsg, data)
}
}
// SendReceiptsRLP sends a batch of transaction receipts, corresponding to the
// ones requested from an already RLP encoded format.
func (p *peer) SendReceiptsRLP(receipts []rlp.RawValue) error {
if p.pairRw != nil {
return p2p.Send(p.pairRw, ReceiptsMsg, receipts)
} else {
return p2p.Send(p.rw, ReceiptsMsg, receipts)
}
}
// RequestOneHeader is a wrapper around the header query functions to fetch a
// single header. It is used solely by the fetcher.
func (p *peer) RequestOneHeader(hash common.Hash) error {
p.Log().Debug("Fetching single header", "hash", hash)
if p.pairRw != nil {
return p2p.Send(p.pairRw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false})
} else {
return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: hash}, Amount: uint64(1), Skip: uint64(0), Reverse: false})
}
}
// RequestHeadersByHash fetches a batch of blocks' headers corresponding to the
// specified header query, based on the hash of an origin block.
func (p *peer) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromhash", origin, "skip", skip, "reverse", reverse)
if p.pairRw != nil {
return p2p.Send(p.pairRw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
} else {
return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Hash: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
}
}
// RequestHeadersByNumber fetches a batch of blocks' headers corresponding to the
// specified header query, based on the number of an origin block.
func (p *peer) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error {
p.Log().Debug("Fetching batch of headers", "count", amount, "fromnum", origin, "skip", skip, "reverse", reverse)
if p.pairRw != nil {
return p2p.Send(p.pairRw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
} else {
return p2p.Send(p.rw, GetBlockHeadersMsg, &getBlockHeadersData{Origin: hashOrNumber{Number: origin}, Amount: uint64(amount), Skip: uint64(skip), Reverse: reverse})
}
}
// RequestBodies fetches a batch of blocks' bodies corresponding to the hashes
// specified.
func (p *peer) RequestBodies(hashes []common.Hash) error {
p.Log().Debug("Fetching batch of block bodies", "count", len(hashes))
if p.pairRw != nil {
return p2p.Send(p.pairRw, GetBlockBodiesMsg, hashes)
} else {
return p2p.Send(p.rw, GetBlockBodiesMsg, hashes)
}
}
// RequestNodeData fetches a batch of arbitrary data from a node's known state
// data, corresponding to the specified hashes.
func (p *peer) RequestNodeData(hashes []common.Hash) error {
p.Log().Debug("Fetching batch of state data", "count", len(hashes))
if p.pairRw != nil {
return p2p.Send(p.pairRw, GetNodeDataMsg, hashes)
} else {
return p2p.Send(p.rw, GetNodeDataMsg, hashes)
}
}
// RequestReceipts fetches a batch of transaction receipts from a remote node.
func (p *peer) RequestReceipts(hashes []common.Hash) error {
p.Log().Debug("Fetching batch of receipts", "count", len(hashes))
if p.pairRw != nil {
return p2p.Send(p.pairRw, GetReceiptsMsg, hashes)
} else {
return p2p.Send(p.rw, GetReceiptsMsg, hashes)
}
}
// Handshake executes the eth protocol handshake, negotiating version number,

1
eth/protocol.go
View file

@ -106,7 +106,6 @@ type txPool interface {
// SubscribeTxPreEvent should return an event subscription of
// TxPreEvent and send events to the given channel.
SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
SubscribeSpecialTxPreEvent(chan<- core.TxPreEvent) event.Subscription
}
// statusData is the network packet for the status message.

19
eth/sync.go
View file

@ -170,7 +170,6 @@ func (pm *ProtocolManager) synchronise(peer *peer) {
currentBlock := pm.blockchain.CurrentBlock()
td := pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64())
pHead, pTd := peer.Head()
log.Debug("ProtocolManager synchronise ", "p", peer, "pTd", pTd, "currentTd", td)
if pTd.Cmp(td) <= 0 {
return
}
@ -205,13 +204,13 @@ func (pm *ProtocolManager) synchronise(peer *peer) {
atomic.StoreUint32(&pm.fastSync, 0)
}
atomic.StoreUint32(&pm.acceptTxs, 1) // Mark initial sync done
if head := pm.blockchain.CurrentBlock(); head.NumberU64() > 0 {
// We've completed a sync cycle, notify all peers of new state. This path is
// essential in star-topology networks where a gateway node needs to notify
// all its out-of-date peers of the availability of a new block. This failure
// scenario will most often crop up in private and hackathon networks with
// degenerate connectivity, but it should be healthy for the mainnet too to
// more reliably update peers or the local TD state.
go pm.BroadcastBlock(head, false)
}
//if head := pm.blockchain.CurrentBlock(); head.NumberU64() > 0 {
// // We've completed a sync cycle, notify all peers of new state. This path is
// // essential in star-topology networks where a gateway node needs to notify
// // all its out-of-date peers of the availability of a new block. This failure
// // scenario will most often crop up in private and hackathon networks with
// // degenerate connectivity, but it should be healthy for the mainnet too to
// // more reliably update peers or the local TD state.
// go pm.BroadcastBlock(head, false)
//}
}

9
miner/miner.go
View file

@ -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, announceTxs 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, announceTxs),
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
}
}
}

111
miner/worker.go
View file

@ -132,9 +132,11 @@ type worker struct {
// atomic status counters
mining int32
atWork int32
announceTxs 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, announceTxs bool) *worker {
worker := &worker{
config: config,
engine: engine,
@ -151,9 +153,12 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
coinbase: coinbase,
agents: make(map[Agent]struct{}),
unconfirmed: newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth),
announceTxs: announceTxs,
}
if worker.announceTxs {
// 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,23 +253,47 @@ func (self *worker) unregister(agent Agent) {
}
func (self *worker) update() {
if self.announceTxs {
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 {
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:
if self.config.Posv == nil {
self.uncleMu.Lock()
self.possibleUncles[ev.Block.Hash()] = ev.Block
self.uncleMu.Unlock()
}
// Handle TxPreEvent
case ev := <-self.txCh:
// Apply transaction to the pending state if we're not mining
@ -282,9 +311,6 @@ func (self *worker) update() {
self.commitNewWork()
}
}
// System stopped
case <-self.txSub.Err():
return
case <-self.chainHeadSub.Err():
return
case <-self.chainSideSub.Err():
@ -422,6 +448,7 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
createdAt: time.Now(),
}
if self.config.Posv == nil {
// when 08 is processed ancestors contain 07 (quick block)
for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) {
for _, uncle := range ancestor.Uncles() {
@ -430,6 +457,7 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
work.family.Add(ancestor.Hash())
work.ancestors.Add(ancestor.Hash())
}
}
// Keep track of transactions which return errors so they can be removed
work.tcount = 0
@ -444,17 +472,6 @@ func abs(x int64) int64 {
return x
}
func hop(len, pre, cur int) int {
switch {
case pre < cur:
return cur - (pre + 1)
case pre > cur:
return (len - pre) + (cur - 1)
default:
return len - 1
}
}
func (self *worker) commitNewWork() {
self.mu.Lock()
defer self.mu.Unlock()
@ -466,6 +483,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.announceTxs && 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
@ -473,14 +497,7 @@ func (self *worker) commitNewWork() {
if self.config.Posv != nil {
// get masternodes set from latest checkpoint
c := self.engine.(*posv.Posv)
masternodes := c.GetMasternodes(self.chain, parent.Header())
snap, err := c.GetSnapshot(self.chain, parent.Header())
if err != nil {
log.Error("Failed when trying to commit new work", "err", err)
return
}
signers = snap.Signers
preIndex, curIndex, ok, err := posv.YourTurn(masternodes, snap, parent.Header(), self.coinbase)
len, preIndex, curIndex, ok, err := c.YourTurn(self.chain, parent.Header(),self.coinbase)
if err != nil {
log.Error("Failed when trying to commit new work", "err", err)
return
@ -496,7 +513,7 @@ func (self *worker) commitNewWork() {
// you're not allowed to create this block
return
}
h := hop(len(masternodes), preIndex, curIndex)
h := posv.Hop(len, preIndex, curIndex)
gap := waitPeriod * int64(h)
// Check nearest checkpoint block in hop range.
nearest := self.config.Posv.Epoch - (parent.Header().Number.Uint64() % self.config.Posv.Epoch)
@ -504,23 +521,14 @@ func (self *worker) commitNewWork() {
gap = waitPeriodCheckpoint * int64(h)
}
log.Info("Distance from the parent block", "seconds", gap, "hops", h)
L:
select {
case newBlock := <-self.chainHeadCh:
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())
waitedTime := time.Now().Unix() - parent.Header().Time.Int64()
if gap > waitedTime {
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)
}
}
}
}
tstamp := tstart.Unix()
if parent.Time().Cmp(new(big.Int).SetInt64(tstamp)) >= 0 {
tstamp = parent.Time().Int64() + 1
@ -585,6 +593,7 @@ func (self *worker) commitNewWork() {
uncles []*types.Header
badUncles []common.Hash
)
if self.config.Posv == nil {
for hash, uncle := range self.possibleUncles {
if len(uncles) == 2 {
break
@ -602,6 +611,7 @@ func (self *worker) commitNewWork() {
for _, hash := range badUncles {
delete(self.possibleUncles, hash)
}
}
// Create the new block to seal with the consensus engine
if work.Block, err = self.engine.Finalize(self.chain, header, work.state, work.txs, uncles, work.receipts); err != nil {
log.Error("Failed to finalize block for sealing", "err", err)
@ -611,6 +621,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)
}
@ -634,7 +645,6 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
gp := new(core.GasPool).AddGas(env.header.GasLimit)
var coalescedLogs []*types.Log
// first priority for special Txs
for _, tx := range specialTxs {
if gp.Gas() < params.TxGas && tx.Gas() > 0 {
@ -654,6 +664,11 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
}
// Start executing the transaction
env.state.Prepare(tx.Hash(), common.Hash{}, env.tcount)
nonce := env.state.GetNonce(from)
if nonce != tx.Nonce() {
log.Trace("Skipping account with special transaction invalide nonce", "sender", from, "nonce", nonce, "tx nonce ", tx.Nonce(), "to", tx.To())
continue
}
err, logs := env.commitTransaction(tx, bc, coinbase, gp)
switch err {
case core.ErrNonceTooLow:
@ -663,7 +678,6 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
case core.ErrNonceTooHigh:
// Reorg notification data race between the transaction pool and miner, skip account =
log.Trace("Skipping account with special transaction hight nonce", "sender", from, "nonce", tx.Nonce(), "to", tx.To())
case nil:
// Everything ok, collect the logs and shift in the next transaction from the same account
coalescedLogs = append(coalescedLogs, logs...)
@ -675,7 +689,6 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
log.Debug("Add Special Transaction failed, account skipped", "hash", tx.Hash(), "sender", from, "nonce", tx.Nonce(), "to", tx.To(), "err", err)
}
}
for {
// If we don't have enough gas for any further transactions then we're done
if gp.Gas() < params.TxGas {
@ -696,13 +709,24 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
// phase, start ignoring the sender until we do.
if tx.Protected() && !env.config.IsEIP155(env.header.Number) {
log.Trace("Ignoring reply protected transaction", "hash", tx.Hash(), "eip155", env.config.EIP155Block)
txs.Pop()
continue
}
// Start executing the transaction
env.state.Prepare(tx.Hash(), common.Hash{}, env.tcount)
nonce := env.state.GetNonce(from)
if nonce > tx.Nonce() {
// New head notification data race between the transaction pool and miner, shift
log.Trace("Skipping transaction with low nonce", "sender", from, "nonce", tx.Nonce())
txs.Shift()
continue
}
if nonce < tx.Nonce() {
// Reorg notification data race between the transaction pool and miner, skip account =
log.Trace("Skipping account with hight nonce", "sender", from, "nonce", tx.Nonce())
txs.Pop()
continue
}
err, logs := env.commitTransaction(tx, bc, coinbase, gp)
switch err {
case core.ErrGasLimitReached:
@ -733,7 +757,6 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
txs.Shift()
}
}
if len(coalescedLogs) > 0 || env.tcount > 0 {
// make a copy, the state caches the logs and these logs get "upgraded" from pending to mined
// logs by filling in the block hash when the block was mined by the local miner. This can

2
node/config.go
View file

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