Fixed.

Dockerfile.node

@@ -27,6 +27,7 @@ ENV NETWORK_ID '89'
 ENV WS_SECRET ''
 ENV NETSTATS_HOST 'netstats-server'
 ENV NETSTATS_PORT '3000'
+ENV ANNOUNCE_TXS ''
 
 RUN apk add --no-cache ca-certificates
 

cmd/tomo/main.go

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

cmd/utils/flags.go

@@ -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) {

consensus/consensus.go

@@ -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

consensus/posv/posv.go

@@ -46,8 +46,6 @@ import (
 
 const (
 	inmemorySnapshots  = 128                    // Number of recent vote snapshots to keep in memory
-	inmemorySignatures = 4096                   // Number of recent block signatures to keep in memory
-	wiggleTime         = 500 * time.Millisecond // Random delay (per signer) to allow concurrent signers
 	M2ByteLength       = 4
 )
 
@@ -58,8 +56,7 @@ type Masternode struct {
 
 // Posv proof-of-stake-voting protocol constants.
 var (
-	epochLength = uint64(30000) // Default number of blocks after which to checkpoint and reset the pending votes
+	epochLength = uint64(900) // Default number of blocks after which to checkpoint and reset the pending votes
-	blockPeriod = uint64(15)    // Default minimum difference between two consecutive block's timestamps
 
 	extraVanity = 32 // Fixed number of extra-data prefix bytes reserved for signer vanity
 	extraSeal   = 65 // Fixed number of extra-data suffix bytes reserved for signer seal
@@ -226,11 +223,11 @@ type Posv struct {
 
 	HookReward    func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error
 	HookPenalty   func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error)
-	HookValidator func(header *types.Header, signers []common.Address) error
+	HookValidator func(header *types.Header, signers []common.Address) ([]byte, error)
 	HookVerifyMNs func(header *types.Header, signers []common.Address) error
 }
 
-// New creates a Posv proof-of-stake-voting consensus engine with the initial
+// New creates a PoSV proof-of-stake-voting consensus engine with the initial
 // signers set to the ones provided by the user.
 func New(config *params.PosvConfig, db ethdb.Database) *Posv {
 	// Set any missing consensus parameters to their defaults
@@ -261,20 +258,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 {
+func (c *Posv) VerifyHeader(chain consensus.ChainReader, header *types.Header, fullVerify bool) error {
-	return c.verifyHeaderWithCache(chain, header, nil)
+	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 +283,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 +299,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{}) {
@@ -344,29 +345,23 @@ func (c *Posv) verifyHeader(chain consensus.ChainReader, header *types.Header, p
 	if header.MixDigest != (common.Hash{}) {
 		return errInvalidMixDigest
 	}
-	// Ensure that the block doesn't contain any uncles which are meaningless in PoA
+	// Ensure that the block doesn't contain any uncles which are meaningless in PoSV
 	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 +380,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 +421,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) {
@@ -471,7 +463,7 @@ func (c *Posv) GetMasternodes(chain consensus.ChainReader, header *types.Header)
 
 func (c *Posv) GetPeriod() uint64 { return c.config.Period }
 
-func WhoIsCreator(snap *Snapshot, header *types.Header) (common.Address, error) {
+func whoIsCreator(snap *Snapshot, header *types.Header) (common.Address, error) {
 	if header.Number.Uint64() == 0 {
 		return common.Address{}, errors.New("Don't take block 0")
 	}
@@ -482,31 +474,38 @@ 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)
 	masternodes = masternodes[:len(masternodes)/2+1]
+	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 {
+	if parent.Number.Uint64() != 0 {
-		pre, err = WhoIsCreator(snap, header)
+		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)
+	curIndex := position(masternodes, signer)
-	log.Info("Masternodes cycle info", "number of masternodes", len(masternodes), "previous", pre, "position", preIndex, "current", cur, "position", curIndex)
+	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.
@@ -534,7 +533,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)
@@ -599,7 +598,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
@@ -608,7 +607,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 {
@@ -625,6 +624,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 {
@@ -664,7 +677,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
@@ -742,38 +755,49 @@ 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))...)
 	}
 	header.Extra = header.Extra[:extraVanity]
-	signers := snap.GetSigners()
+	masternodes := snap.GetSigners()
-	if number%c.config.Epoch == 0 {
+	if number > 0 && number%c.config.Epoch == 0 {
 		if c.HookPenalty != nil {
-			penSigners, err := c.HookPenalty(chain, number)
+			penMasternodes, err := c.HookPenalty(chain, number)
 			if err != nil {
 				return err
 			}
-			if len(penSigners) > 0 {
+			if len(penMasternodes) > 0 {
-				// Keep remove penalty signer out of signer list.
+				// penalize bad masternode(s)
-				signers = common.RemoveItemFromArray(signers, penSigners)
+				masternodes = common.RemoveItemFromArray(masternodes, penMasternodes)
-				for _, address := range penSigners {
+				for _, address := range penMasternodes {
-					log.Debug("Penalty Info", "address", address, "number", number)
+					log.Debug("Penalty status", "address", address, "block number", number)
 				}
-				header.Penalties = common.ExtractAddressToBytes(penSigners)
+				header.Penalties = common.ExtractAddressToBytes(penMasternodes)
 			}
 		}
-		// Prevent penaltied signer in 4 epocs ago jump into signer list.
+		// Prevent penalized masternode(s) within 4 recent epochs
 		for i := 1; i <= common.LimitPenaltyEpoch; i++ {
 			if number > uint64(i)*c.config.Epoch {
-				signers = RemovePenaltiesFromBlock(chain, signers, number-uint64(i)*c.config.Epoch)
+				masternodes = RemovePenaltiesFromBlock(chain, masternodes, number-uint64(i)*c.config.Epoch)
 			}
 		}
-		for _, signer := range signers {
+		for _, masternode := range masternodes {
-			header.Extra = append(header.Extra, signer[:]...)
+			header.Extra = append(header.Extra, masternode[:]...)
+		}
+		if c.HookValidator != nil {
+			validators, err := c.HookValidator(header, masternodes)
+			if err != nil {
+				return err
+			}
+			header.Validators = validators
 		}
 	}
 	header.Extra = append(header.Extra, make([]byte, extraSeal)...)
@@ -782,20 +806,11 @@ 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())
 	}
-	if c.HookValidator != nil {
-		c.HookValidator(header, signers)
-		if err != nil {
-			return err
-		}
-	}
 	return nil
 }
 
@@ -807,23 +822,14 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head
 	if err != nil {
 		return err
 	}
-	currentSigners := snap.GetSigners()
+	newMasternodes := make(map[common.Address]struct{})
-	proposedSigners := make(map[common.Address]struct{})
-	// count all addresses in ms to be masternode
 	for _, m := range ms {
-		proposedSigners[m.Address] = struct{}{}
+		newMasternodes[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)
-		}
 	}
+	snap.Signers = newMasternodes
 	nm := []string{}
-	newSigners := snap.GetSigners()
+	for _, n := range ms {
-	for _, n := range newSigners {
+		nm = append(nm, n.Address.String())
-		nm = append(nm, n.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)
@@ -834,7 +840,6 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head
 // rewards given, and returns the final block.
 func (c *Posv) Finalize(chain consensus.ChainReader, header *types.Header, state *state.StateDB, txs []*types.Transaction, uncles []*types.Header, receipts []*types.Receipt) (*types.Block, error) {
 	// set block reward
-	// FIXME: unit Ether could be too plump
 	number := header.Number.Uint64()
 	rCheckpoint := chain.Config().Posv.RewardCheckpoint
 
@@ -844,7 +849,7 @@ func (c *Posv) Finalize(chain consensus.ChainReader, header *types.Header, state
 		}
 	}
 
-	// No block rewards in PoA, so the state remains as is and uncles are dropped
+	// the state remains as is and uncles are dropped
 	header.Root = state.IntermediateRoot(chain.Config().IsEIP158(header.Number))
 	header.UncleHash = types.CalcUncleHash(nil)
 
@@ -909,7 +914,7 @@ func (c *Posv) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan
 				if limit := uint64(2); number < limit || seen > number-limit {
 					// Only take into account the non-epoch blocks
 					if number%c.config.Epoch != 0 {
-						log.Info("Debugging", "len(masternodes)", len(masternodes), "number", number, "limit", limit, "seen", seen, "recent", recent.String(), "snap.Recents", snap.Recents)
+						log.Info("len(masternodes)", len(masternodes), "number", number, "limit", limit, "seen", seen, "recent", recent.String(), "snap.Recents", snap.Recents)
 						log.Info("Signed recently, must wait for others")
 						<-stop
 						return nil, nil
@@ -937,21 +942,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)
+	return c.calcDifficulty(chain, parent, c.signer)
-	if err != nil {
-		return nil
-	}
-	return CalcDifficulty(snap, c.signer)
 }
 
-// CalcDifficulty is the difficulty adjustment algorithm. It returns the difficulty
+func (c *Posv) calcDifficulty(chain consensus.ChainReader, parent *types.Header, signer common.Address) *big.Int {
-// that a new block should have based on the previous blocks in the chain and the
+	len, preIndex, curIndex, _, err := c.YourTurn(chain, parent, signer)
-// current signer.
+	if err != nil {
-func CalcDifficulty(snap *Snapshot, signer common.Address) *big.Int {
+		return big.NewInt(int64(len + curIndex - preIndex))
-	if snap.inturn(snap.Number+1, signer) {
-		return new(big.Int).Set(diffInTurn)
 	}
-	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
@@ -1006,18 +1005,18 @@ func (c *Posv) GetMasternodesFromCheckpointHeader(preCheckpointHeader *types.Hea
 }
 
 // Extract validators from byte array.
-func RemovePenaltiesFromBlock(chain consensus.ChainReader, signers []common.Address, epochNumber uint64) []common.Address {
+func RemovePenaltiesFromBlock(chain consensus.ChainReader, masternodes []common.Address, epochNumber uint64) []common.Address {
 	if epochNumber <= 0 {
-		return signers
+		return masternodes
 	}
 	header := chain.GetHeaderByNumber(epochNumber)
 	block := chain.GetBlock(header.Hash(), epochNumber)
 	penalties := block.Penalties()
 	if penalties != nil {
 		prevPenalties := common.ExtractAddressFromBytes(penalties)
-		signers = common.RemoveItemFromArray(signers, prevPenalties)
+		masternodes = common.RemoveItemFromArray(masternodes, prevPenalties)
 	}
-	return signers
+	return masternodes
 }
 
 // Get masternodes address from checkpoint Header.
@@ -1066,3 +1065,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
+	}
+}

contracts/blocksigner/contract/blocksigner.go

@@ -377,7 +377,7 @@ func (_BlockSigner *BlockSignerFilterer) WatchSign(opts *bind.WatchOpts, sink ch
 const SafeMathABI = "[]"
 
 // SafeMathBin is the compiled bytecode used for deploying new contracts.
-const SafeMathBin = `0x604c602c600b82828239805160001a60731460008114601c57601e565bfe5b5030600052607381538281f30073000000000000000000000000000000000000000030146080604052600080fd00a165627a7a72305820a3f63b465e1cf25f306b1eb1efefc8dac3c38993a7340f69d8b470c3bf599ff30029`
+const SafeMathBin = `0x604c602c600b82828239805160001a60731460008114601c57601e565bfe5b5030600052607381538281f30073000000000000000000000000000000000000000030146060604052600080fd00a165627a7a72305820b9407d48ebc7efee5c9f08b3b3a957df2939281f5913225e8c1291f069b900490029`
 
 // DeploySafeMath deploys a new Ethereum contract, binding an instance of SafeMath to it.
 func DeploySafeMath(auth *bind.TransactOpts, backend bind.ContractBackend) (common.Address, *types.Transaction, *SafeMath, error) {

contracts/randomize/contract/randomize.go

@@ -16,7 +16,7 @@ import (
 const SafeMathABI = "[]"
 
 // SafeMathBin is the compiled bytecode used for deploying new contracts.
-const SafeMathBin = `0x604c602c600b82828239805160001a60731460008114601c57601e565bfe5b5030600052607381538281f30073000000000000000000000000000000000000000030146080604052600080fd00a165627a7a72305820ea3f7b50706b29b7324f2b7bc6c5eec464df491d2c006d76decd3a344ef24b5b0029`
+const SafeMathBin = `0x604c602c600b82828239805160001a60731460008114601c57601e565bfe5b5030600052607381538281f30073000000000000000000000000000000000000000030146060604052600080fd00a165627a7a72305820b9407d48ebc7efee5c9f08b3b3a957df2939281f5913225e8c1291f069b900490029`
 
 // DeploySafeMath deploys a new Ethereum contract, binding an instance of SafeMath to it.
 func DeploySafeMath(auth *bind.TransactOpts, backend bind.ContractBackend) (common.Address, *types.Transaction, *SafeMath, error) {
@@ -177,7 +177,7 @@ func (_SafeMath *SafeMathTransactorRaw) Transact(opts *bind.TransactOpts, method
 const TomoRandomizeABI = "[{\"constant\":true,\"inputs\":[{\"name\":\"_validator\",\"type\":\"address\"}],\"name\":\"getSecret\",\"outputs\":[{\"name\":\"\",\"type\":\"bytes32[]\"}],\"payable\":false,\"stateMutability\":\"view\",\"type\":\"function\"},{\"constant\":false,\"inputs\":[{\"name\":\"_secret\",\"type\":\"bytes32[]\"}],\"name\":\"setSecret\",\"outputs\":[],\"payable\":false,\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"constant\":true,\"inputs\":[{\"name\":\"_validator\",\"type\":\"address\"}],\"name\":\"getOpening\",\"outputs\":[{\"name\":\"\",\"type\":\"bytes32\"}],\"payable\":false,\"stateMutability\":\"view\",\"type\":\"function\"},{\"constant\":false,\"inputs\":[{\"name\":\"_opening\",\"type\":\"bytes32\"}],\"name\":\"setOpening\",\"outputs\":[],\"payable\":false,\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"inputs\":[],\"payable\":false,\"stateMutability\":\"nonpayable\",\"type\":\"constructor\"}]"
 
 // TomoRandomizeBin is the compiled bytecode used for deploying new contracts.
-const TomoRandomizeBin = `0x608060405234801561001057600080fd5b50610335806100206000396000f3006080604052600436106100615763ffffffff7c0100000000000000000000000000000000000000000000000000000000600035041663284180fc811461006657806334d38600146100e4578063d442d6cc1461013b578063e11f5ba21461017b575b600080fd5b34801561007257600080fd5b5061009473ffffffffffffffffffffffffffffffffffffffff60043516610193565b60408051602080825283518183015283519192839290830191858101910280838360005b838110156100d05781810151838201526020016100b8565b505050509050019250505060405180910390f35b3480156100f057600080fd5b50604080516020600480358082013583810280860185019096528085526101399536959394602494938501929182918501908490808284375094975061020b9650505050505050565b005b34801561014757600080fd5b5061016973ffffffffffffffffffffffffffffffffffffffff60043516610250565b60408051918252519081900360200190f35b34801561018757600080fd5b50610139600435610278565b73ffffffffffffffffffffffffffffffffffffffff8116600090815260208181526040918290208054835181840281018401909452808452606093928301828280156101ff57602002820191906000526020600020905b815481526001909101906020018083116101ea575b50505050509050919050565b610384430661032081101561021f57600080fd5b610352811061022d57600080fd5b33600090815260208181526040909120835161024b9285019061029f565b505050565b73ffffffffffffffffffffffffffffffffffffffff1660009081526001602052604090205490565b610384430661035281101561028c57600080fd5b5033600090815260016020526040902055565b8280548282559060005260206000209081019282156102dc579160200282015b828111156102dc57825182556020909201916001909101906102bf565b506102e89291506102ec565b5090565b61030691905b808211156102e857600081556001016102f2565b905600a165627a7a72305820524cb2eeb0cc4214180425f822fd315cd15fd8352a830bbd9846b46133730a100029`
+const TomoRandomizeBin = `0x6060604052341561000f57600080fd5b6103368061001e6000396000f3006060604052600436106100615763ffffffff7c0100000000000000000000000000000000000000000000000000000000600035041663284180fc811461006657806334d38600146100d8578063d442d6cc14610129578063e11f5ba21461015a575b600080fd5b341561007157600080fd5b610085600160a060020a0360043516610170565b60405160208082528190810183818151815260200191508051906020019060200280838360005b838110156100c45780820151838201526020016100ac565b505050509050019250505060405180910390f35b34156100e357600080fd5b61012760046024813581810190830135806020818102016040519081016040528093929190818152602001838360200280828437509496506101f395505050505050565b005b341561013457600080fd5b610148600160a060020a0360043516610243565b60405190815260200160405180910390f35b341561016557600080fd5b61012760043561025e565b61017861028e565b60008083600160a060020a0316600160a060020a031681526020019081526020016000208054806020026020016040519081016040528092919081815260200182805480156101e757602002820191906000526020600020905b815481526001909101906020018083116101d2575b50505050509050919050565b610384430661032081101561020757600080fd5b610352811061021557600080fd5b600160a060020a033316600090815260208190526040902082805161023e9291602001906102a0565b505050565b600160a060020a031660009081526001602052604090205490565b610384430661035281101561027257600080fd5b50600160a060020a033316600090815260016020526040902055565b60206040519081016040526000815290565b8280548282559060005260206000209081019282156102dd579160200282015b828111156102dd57825182556020909201916001909101906102c0565b506102e99291506102ed565b5090565b61030791905b808211156102e957600081556001016102f3565b905600a165627a7a7230582034991c8dc4001fc254f3ba2811c05d2e7d29bee3908946ca56d1545b2c852de20029`
 
 // DeployTomoRandomize deploys a new Ethereum contract, binding an instance of TomoRandomize to it.
 func DeployTomoRandomize(auth *bind.TransactOpts, backend bind.ContractBackend) (common.Address, *types.Transaction, *TomoRandomize, error) {

contracts/validator/contract/TomoValidator.sol

@@ -98,7 +98,7 @@ contract TomoValidator {
         maxValidatorNumber = _maxValidatorNumber;
         candidateWithdrawDelay = _candidateWithdrawDelay;
         voterWithdrawDelay = _voterWithdrawDelay;
-        candidateCount = candidateCount.add(_candidates.length);
+        candidateCount = _candidates.length;
 
         for (uint256 i = 0; i < _candidates.length; i++) {
             candidates.push(_candidates[i]);
@@ -108,7 +108,7 @@ contract TomoValidator {
                 cap: _caps[i]
             });
             voters[_candidates[i]].push(_firstOwner);
-            validatorsState[candidates[i]].voters[_firstOwner] = minCandidateCap;
+            validatorsState[_candidates[i]].voters[_firstOwner] = minCandidateCap;
         }
     }
 
@@ -120,7 +120,7 @@ contract TomoValidator {
             isCandidate: true,
             cap: cap
         });
-        validatorsState[_candidate].voters[msg.sender] = msg.value;
+        validatorsState[_candidate].voters[msg.sender] = validatorsState[_candidate].voters[msg.sender].add(msg.value);
         candidateCount = candidateCount.add(1);
         voters[_candidate].push(msg.sender);
         emit Propose(msg.sender, _candidate, msg.value);

core/blockchain.go

@@ -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 := externTd.Cmp(localTd) > 0
-	reorg := block.NumberU64() > currentBlock.NumberU64()
+	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

core/error.go

@@ -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")
 )

core/state_processor.go

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

core/tx_pool.go

@@ -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{}
+	go pool.txFeed.Send(TxPreEvent{tx})
-	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())
-		}
-	}()
 	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()
 

core/types.go

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

core/types/block.go

@@ -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.

docker/tomochain/entrypoint.sh

@@ -28,7 +28,7 @@ accountsCount=$(
 
 # file to env
 for env in IDENTITY PASSWORD PRIVATE_KEY BOOTNODES WS_SECRET NETSTATS_HOST \
-           NETSTATS_PORT EXTIP SYNC_MODE NETWORK_ID; do
+           NETSTATS_PORT EXTIP SYNC_MODE NETWORK_ID ANNOUNCE_TXS; do
   file=$(eval echo "\$${env}_FILE")
   if [[ -f $file ]] && [[ ! -z $file ]]; then
     echo "Replacing $env by $file"
@@ -139,6 +139,11 @@ else
   echo "WS_SECRET not set, will not report to netstats server."
 fi
 
+# annonce txs
+if [[ ! -z $ANNOUNCE_TXS ]]; then
+  params="$params --announce-txs"
+fi
+
 # dump
 echo "dump: $IDENTITY $account $BOOTNODES"
 
@@ -151,7 +156,7 @@ exec tomo $params \
   --identity $IDENTITY \
   --password ./password \
   --port 30303 \
-  --maxpeers 50 \
+  --maxpeers 25 \
   --txpool.globalqueue 5000 \
   --txpool.globalslots 5000 \
   --rpc \

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().AnnounceTxs)
 	eth.miner.SetExtra(makeExtraData(config.ExtraData))
 
 	eth.ApiBackend = &EthApiBackend{eth, nil}
@@ -203,47 +203,43 @@ 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, false, nil
-			return block, nil
 		}
 
 		eth.protocolManager.fetcher.SetSignHook(signHook)
 		eth.protocolManager.fetcher.SetAppendM2HeaderHook(appendM2HeaderHook)
 
-		// Hook prepares validators M2 for the current epoch
+		// Hook prepares validators M2 for the current epoch at checkpoint block
-		c.HookValidator = func(header *types.Header, signers []common.Address) error {
+		c.HookValidator = func(header *types.Header, signers []common.Address) ([]byte, error) {
 			start := time.Now()
-			number := header.Number.Int64()
-			if number > 0 && number%common.EpocBlockRandomize == 0 {
 			validators, err := GetValidators(eth.blockchain, signers)
 			if err != nil {
-					return err
+				return []byte{}, err
 			}
 			header.Validators = validators
-			}
 			log.Debug("Time Calculated HookValidator ", "block", header.Number.Uint64(), "time", common.PrettyDuration(time.Since(start)))
-			return nil
+			return validators, nil
 		}
 
 		// Hook scans for bad masternodes and decide to penalty them
@@ -301,8 +297,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 +330,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
 		}
 

eth/downloader/downloader.go

@@ -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
+	ttlScaling       = 2                        // Constant scaling factor for RTT -> TTL conversion
-	ttlLimit         = time.Minute              // Maximum TTL allowance to prevent reaching crazy timeouts
+	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

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

eth/fetcher/fetcher.go

@@ -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.
+// blockInsertFn is a callback type to insert a batch of blocks into the local chain.
-type chainInsertFn func(blocks types.Blocks) (int, error)
+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 {
+				if newBlock, isM2, errM2 = f.appendM2HeaderHook(block); errM2 != nil {
-					log.Error("Append m2 to block header fail", "err", err)
+					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
 }

eth/fetcher/fetcher_test.go

@@ -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) {
+	tester.fetcher.insertBlock = func(block *types.Block) error {
-		atomic.AddUint32(&counter, uint32(len(blocks)))
+		atomic.AddUint32(&counter, uint32(1))
-		return tester.insertChain(blocks)
+		return tester.insertBlock(block)
 	}
 	// Instrument the fetching and imported events
 	fetching := make(chan []common.Hash)

eth/handler.go

@@ -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())
+			log.Warn("Discarded bad propagated block", "number", block.Number(), "hash", block.Hash())
-			return 0, nil
+			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()
+		if _, td := p.Head(); trueTD.Cmp(td) > 0 {
-		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 {
 			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 {

eth/helper_test.go

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

eth/peer.go

@@ -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,82 +158,124 @@ 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,
 // network IDs, difficulties, head and genesis blocks.

eth/protocol.go

@@ -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.

eth/sync.go

@@ -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 {
+	//if head := pm.blockchain.CurrentBlock(); head.NumberU64() > 0 {
-		// We've completed a sync cycle, notify all peers of new state. This path is
+	//	// 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
+	//	// 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
+	//	// 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
+	//	// scenario will most often crop up in private and hackathon networks with
-		// degenerate connectivity, but it should be healthy for the mainnet too to
+	//	// degenerate connectivity, but it should be healthy for the mainnet too to
-		// more reliably update peers or the local TD state.
+	//	// more reliably update peers or the local TD state.
-		go pm.BroadcastBlock(head, false)
+	//	go pm.BroadcastBlock(head, false)
-	}
+	//}
 }

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, 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
 		}
 	}
 }

miner/worker.go

@@ -55,7 +55,7 @@ const (
 	// timeout waiting for M1
 	waitPeriod = 10
 	// timeout for checkpoint.
-	waitPeriodCheckpoint = 60
+	waitPeriodCheckpoint = 30
 )
 
 // Agent can register themself with the worker
@@ -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())
+			len, preIndex, curIndex, ok, err := c.YourTurn(self.chain, parent.Header(),self.coinbase)
-			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)
 			if err != nil {
 				log.Error("Failed when trying to commit new work", "err", err)
 				return
@@ -496,31 +513,22 @@ 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)
 				if uint64(h) >= nearest {
-					gap += waitPeriodCheckpoint
+					gap = waitPeriodCheckpoint * int64(h)
 				}
 				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:
-					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):
+				log.Info("Wait enough. It's my turn", "waited seconds", waitedTime)
-					// wait enough. It's my turn
-					log.Info("Wait enough. It's my turn", "waited seconds", gap)
-					break L
 			}
 		}
 	}
-	}
-
 	tstamp := tstart.Unix()
 	if parent.Time().Cmp(new(big.Int).SetInt64(tstamp)) >= 0 {
 		tstamp = parent.Time().Int64() + 1
@@ -545,7 +553,7 @@ func (self *worker) commitNewWork() {
 		header.Coinbase = self.coinbase
 	}
 	if err := self.engine.Prepare(self.chain, header); err != nil {
-		log.Error("Failed to prepare header for mining", "err", err)
+		log.Error("Failed to prepare header for new block", "err", err)
 		return
 	}
 	// If we are care about TheDAO hard-fork check whether to override the extra-data or not
@@ -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,15 +611,16 @@ 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)
 		return
 	}
-	// We only care about logging if we're actually mining.
 	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)))
+		log.Info("Committing new block", "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 +644,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 +663,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 +677,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 +688,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 +708,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:
@@ -717,7 +740,7 @@ 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 hight nonce", "sender", from, "nonce", tx.Nonce())
+			log.Trace("Skipping account with high nonce", "sender", from, "nonce", tx.Nonce())
 			txs.Pop()
 
 		case nil:
@@ -733,7 +756,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

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"`
+
+	AnnounceTxs bool `toml:",omitempty"`
 }
 
 // IPCEndpoint resolves an IPC endpoint based on a configured value, taking into

params/version.go

@@ -22,7 +22,7 @@ import (
 
 const (
 	VersionMajor = 1        // Major version component of the current release
-	VersionMinor = 0       // Minor version component of the current release
+	VersionMinor = 1        // Minor version component of the current release
 	VersionPatch = 0        // Patch version component of the current release
 	VersionMeta  = "stable" // Version metadata to append to the version string
 )