Merge branch 'master' of https://github.com/tomochain/tomochain

consensus/posv/api.go

@@ -71,7 +71,7 @@ func (api *API) GetSigners(number *rpc.BlockNumber) ([]common.Address, error) {
 	if err != nil {
 		return nil, err
 	}
-	return snap.signers(), nil
+	return snap.GetSigners(), nil
 }
 
 // GetSignersAtHash retrieves the state snapshot at a given block.
@@ -84,7 +84,7 @@ func (api *API) GetSignersAtHash(hash common.Hash) ([]common.Address, error) {
 	if err != nil {
 		return nil, err
 	}
-	return snap.signers(), nil
+	return snap.GetSigners(), nil
 }
 
 // Proposals returns the current proposals the node tries to uphold and vote on.

consensus/posv/posv.go

@@ -136,6 +136,8 @@ var (
 	// on an instant chain (0 second period). It's important to refuse these as the
 	// block reward is zero, so an empty block just bloats the chain... fast.
 	errWaitTransactions = errors.New("waiting for transactions")
+
+	ErrInvalidCheckpointValidators = errors.New("invalid validators list on checkpoint block")
 )
 
 // SignerFn is a signer callback function to request a hash to be signed by a
@@ -213,9 +215,10 @@ type Posv struct {
 	signFn clique.SignerFn // Signer function to authorize hashes with
 	lock   sync.RWMutex    // Protects the signer fields
 
-	HookReward  func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error
+	HookReward    func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error
-	HookPenalty func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error)
+	HookPenalty   func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error)
-	HookPrepare func(header *types.Header, signers []common.Address) error
+	HookValidator func(header *types.Header, signers []common.Address) error
+	HookVerifyMNs func(header *types.Header, signers []common.Address) error
 }
 
 // New creates a Posv proof-of-stake-voting consensus engine with the initial
@@ -379,7 +382,7 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types.
 				return errInvalidCheckpointPenalties
 			}
 		}
-		signers := snap.signers()
+		signers := snap.GetSigners()
 		signers = common.RemoveItemFromArray(signers, penPenalties)
 		for i := 1; i <= common.LimitPenaltyEpoch; i++ {
 			if number > uint64(i)*c.config.Epoch {
@@ -391,6 +394,12 @@ func (c *Posv) verifyCascadingFields(chain consensus.ChainReader, header *types.
 		if !bytes.Equal(header.Extra[extraVanity:extraSuffix], byteMasterNodes) {
 			return errInvalidCheckpointSigners
 		}
+		if c.HookVerifyMNs != nil {
+			err := c.HookVerifyMNs(header, signers)
+			if err != nil {
+				return err
+			}
+		}
 	}
 	// All basic checks passed, verify the seal and return
 	return c.verifySeal(chain, header, parents)
@@ -581,7 +590,7 @@ func (c *Posv) verifySeal(chain consensus.ChainReader, header *types.Header, par
 		mstring = append(mstring, m.String())
 	}
 	nstring := []string{}
-	for _, n := range snap.signers() {
+	for _, n := range snap.GetSigners() {
 		nstring = append(nstring, n.String())
 	}
 	if _, ok := snap.Signers[signer]; !ok {
@@ -656,7 +665,7 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error
 		header.Extra = append(header.Extra, bytes.Repeat([]byte{0x00}, extraVanity-len(header.Extra))...)
 	}
 	header.Extra = header.Extra[:extraVanity]
-	signers := snap.signers()
+	signers := snap.GetSigners()
 	if number%c.config.Epoch == 0 {
 		if c.HookPenalty != nil {
 			penSigners, err := c.HookPenalty(chain, number)
@@ -696,8 +705,11 @@ func (c *Posv) Prepare(chain consensus.ChainReader, header *types.Header) error
 	if header.Time.Int64() < time.Now().Unix() {
 		header.Time = big.NewInt(time.Now().Unix())
 	}
-	if c.HookPrepare != nil {
+	if c.HookValidator != nil {
-		c.HookPrepare(header, signers)
+		c.HookValidator(header, signers)
+		if err != nil {
+			return err
+		}
 	}
 	return nil
 }
@@ -710,7 +722,7 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head
 	if err != nil {
 		return err
 	}
-	currentSigners := snap.signers()
+	currentSigners := snap.GetSigners()
 	proposedSigners := make(map[common.Address]struct{})
 	// count all addresses in ms to be masternode
 	for _, m := range ms {
@@ -724,7 +736,7 @@ func (c *Posv) UpdateMasternodes(chain consensus.ChainReader, header *types.Head
 		}
 	}
 	nm := []string{}
-	newSigners := snap.signers()
+	newSigners := snap.GetSigners()
 	for _, n := range newSigners {
 		nm = append(nm, n.String())
 	}

consensus/posv/snapshot.go

@@ -286,7 +286,7 @@ func (s *Snapshot) apply(headers []*types.Header) (*Snapshot, error) {
 }
 
 // signers retrieves the list of authorized signers in ascending order.
-func (s *Snapshot) signers() []common.Address {
+func (s *Snapshot) GetSigners() []common.Address {
 	signers := make([]common.Address, 0, len(s.Signers))
 	for signer := range s.Signers {
 		signers = append(signers, signer)
@@ -303,7 +303,7 @@ func (s *Snapshot) signers() []common.Address {
 
 // inturn returns if a signer at a given block height is in-turn or not.
 func (s *Snapshot) inturn(number uint64, signer common.Address) bool {
-	signers, offset := s.signers(), 0
+	signers, offset := s.GetSigners(), 0
 	for offset < len(signers) && signers[offset] != signer {
 		offset++
 	}

core/error.go

@@ -34,4 +34,6 @@ var (
 	ErrNonceTooHigh = errors.New("nonce too high")
 
 	ErrNotPoSV = errors.New("Posv not found in config")
+
+	ErrNotFoundM1 = errors.New("list M1 not found ")
 )

core/tx_list.go

@@ -251,19 +251,18 @@ func (l *txList) Overlaps(tx *types.Transaction) bool {
 func (l *txList) Add(tx *types.Transaction, priceBump uint64) (bool, *types.Transaction) {
 	// If there's an older better transaction, abort
 	old := l.txs.Get(tx.Nonce())
-
+	if old != nil && old.IsSpecialTransaction() {
-	if (tx.To() != nil && tx.To().String() != common.RandomizeSMC) || tx.To() == nil {
+		return false, nil
-		if old != nil {
+	}
-			threshold := new(big.Int).Div(new(big.Int).Mul(old.GasPrice(), big.NewInt(100+int64(priceBump))), big.NewInt(100))
+	if old != nil {
-			// Have to ensure that the new gas price is higher than the old gas
+		threshold := new(big.Int).Div(new(big.Int).Mul(old.GasPrice(), big.NewInt(100+int64(priceBump))), big.NewInt(100))
-			// price as well as checking the percentage threshold to ensure that
+		// Have to ensure that the new gas price is higher than the old gas
-			// this is accurate for low (Wei-level) gas price replacements
+		// price as well as checking the percentage threshold to ensure that
-			if old.GasPrice().Cmp(tx.GasPrice()) >= 0 || threshold.Cmp(tx.GasPrice()) > 0 {
+		// this is accurate for low (Wei-level) gas price replacements
-				return false, nil
+		if old.GasPrice().Cmp(tx.GasPrice()) >= 0 || threshold.Cmp(tx.GasPrice()) > 0 {
-			}
+			return false, nil
 		}
 	}
-
 	// Otherwise overwrite the old transaction with the current one
 	l.txs.Put(tx)
 	if cost := tx.Cost(); l.costcap.Cmp(cost) < 0 {

core/tx_pool.go

@@ -80,6 +80,8 @@ var (
 	ErrOversizedData = errors.New("oversized data")
 
 	ErrZeroGasPrice = errors.New("zero gas price")
+
+	ErrDuplicateSpecialTransaction = errors.New("duplicate a specail transaction")
 )
 
 var (
@@ -186,16 +188,17 @@ func (config *TxPoolConfig) sanitize() TxPoolConfig {
 // current state) and future transactions. Transactions move between those
 // two states over time as they are received and processed.
 type TxPool struct {
-	config       TxPoolConfig
+	config        TxPoolConfig
-	chainconfig  *params.ChainConfig
+	chainconfig   *params.ChainConfig
-	chain        blockChain
+	chain         blockChain
-	gasPrice     *big.Int
+	gasPrice      *big.Int
-	txFeed       event.Feed
+	txFeed        event.Feed
-	scope        event.SubscriptionScope
+	specialTxFeed event.Feed
-	chainHeadCh  chan ChainHeadEvent
+	scope         event.SubscriptionScope
-	chainHeadSub event.Subscription
+	chainHeadCh   chan ChainHeadEvent
-	signer       types.Signer
+	chainHeadSub  event.Subscription
-	mu           sync.RWMutex
+	signer        types.Signer
+	mu            sync.RWMutex
 
 	currentState  *state.StateDB      // Current state in the blockchain head
 	pendingState  *state.ManagedState // Pending state tracking virtual nonces
@@ -454,6 +457,12 @@ 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()
@@ -576,7 +585,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
 	}
 	// Drop non-local transactions under our own minimal accepted gas price
 	local = local || pool.locals.contains(from) // account may be local even if the transaction arrived from the network
-	if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
+	if !local && tx.To() != nil && !tx.IsSpecialTransaction() && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
 		return ErrUnderpriced
 	}
 	// Ensure the transaction adheres to nonce ordering
@@ -589,7 +598,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
 		return ErrInsufficientFunds
 	}
 
-	if tx.To() != nil && tx.To().String() != common.BlockSigners && tx.To().String() != common.RandomizeSMC {
+	if tx.To() != nil && !tx.IsSpecialTransaction() {
 		intrGas, err := IntrinsicGas(tx.Data(), tx.To() == nil, pool.homestead)
 		if err != nil {
 			return err
@@ -646,8 +655,11 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 			pool.removeTx(tx.Hash())
 		}
 	}
-	// If the transaction is replacing an already pending one, do directly
 	from, _ := types.Sender(pool.signer, tx) // already validated
+	if tx.IsSpecialTransaction() {
+		return pool.promoteSpecialTx(from, tx)
+	}
+	// If the transaction is replacing an already pending one, do directly
 	if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
 		// Nonce already pending, check if required price bump is met
 		inserted, old := list.Add(tx, pool.config.PriceBump)
@@ -763,6 +775,54 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	go pool.txFeed.Send(TxPreEvent{tx})
 }
 
+func (pool *TxPool) promoteSpecialTx(addr common.Address, tx *types.Transaction) (bool, error) {
+	// Try to insert the transaction into the pending queue
+	if pool.pending[addr] == nil {
+		pool.pending[addr] = newTxList(true)
+	}
+	list := pool.pending[addr]
+	old := list.txs.Get(tx.Nonce())
+	if old != nil && old.IsSpecialTransaction() {
+		return false, ErrDuplicateSpecialTransaction
+	}
+	// Otherwise discard any previous transaction and mark this
+	if old != nil {
+		delete(pool.all, old.Hash())
+		pool.priced.Removed()
+		pendingReplaceCounter.Inc(1)
+	}
+	list.txs.Put(tx)
+	if cost := tx.Cost(); list.costcap.Cmp(cost) < 0 {
+		list.costcap = cost
+	}
+	if gas := tx.Gas(); list.gascap < gas {
+		list.gascap = gas
+	}
+	// Failsafe to work around direct pending inserts (tests)
+	if pool.all[tx.Hash()] == nil {
+		pool.all[tx.Hash()] = tx
+	}
+	// Set the potentially new pending nonce and notify any subsystems of the new tx
+	pool.beats[addr] = time.Now()
+	pool.pendingState.SetNonce(addr, tx.Nonce()+1)
+	broadcastTxs := types.Transactions{}
+	for i := tx.Nonce() - 1; i > 0; i-- {
+		before := list.txs.Get(i)
+		if before == nil || before.IsSpecialTransaction() {
+			break
+		}
+		broadcastTxs = append(broadcastTxs, before)
+	}
+	broadcastTxs = append(broadcastTxs, tx)
+	go func() {
+		for _, btx := range broadcastTxs {
+			pool.specialTxFeed.Send(TxPreEvent{btx})
+			log.Debug("Pooled new special transaction", "hash", tx.Hash(), "from", addr, "to", tx.To(), "nonce", tx.Nonce())
+		}
+	}()
+	return true, nil
+}
+
 // AddLocal enqueues a single transaction into the pool if it is valid, marking
 // the sender as a local one in the mean time, ensuring it goes around the local
 // pricing constraints.

core/types/transaction.go

@@ -267,6 +267,13 @@ func (tx *Transaction) RawSignatureValues() (*big.Int, *big.Int, *big.Int) {
 	return tx.data.V, tx.data.R, tx.data.S
 }
 
+func (tx *Transaction) IsSpecialTransaction() bool {
+	if tx.To() == nil {
+		return false
+	}
+	return tx.To().String() == common.RandomizeSMC || tx.To().String() == common.BlockSigners
+}
+
 func (tx *Transaction) String() string {
 	var from, to string
 	if tx.data.V != nil {
@@ -395,14 +402,34 @@ type TransactionsByPriceAndNonce struct {
 //
 // Note, the input map is reowned so the caller should not interact any more with
 // if after providing it to the constructor.
-func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions) *TransactionsByPriceAndNonce {
+
+// It also classifies special txs and normal txs
+func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transactions) (*TransactionsByPriceAndNonce, Transactions) {
 	// Initialize a price based heap with the head transactions
-	heads := make(TxByPrice, 0, len(txs))
+	heads := TxByPrice{}
+	specialTxs := Transactions{}
 	for _, accTxs := range txs {
-		heads = append(heads, accTxs[0])
+		var normalTxs Transactions
-		// Ensure the sender address is from the signer
+		lastSpecialTx := -1
-		acc, _ := Sender(signer, accTxs[0])
+		for i, tx := range accTxs {
-		txs[acc] = accTxs[1:]
+			if tx.IsSpecialTransaction() {
+				lastSpecialTx = i
+			}
+		}
+		if lastSpecialTx >= 0 {
+			for i := 0; i <= lastSpecialTx; i++ {
+				specialTxs = append(specialTxs, accTxs[i])
+			}
+			normalTxs = accTxs[lastSpecialTx+1:]
+		} else {
+			normalTxs = accTxs
+		}
+		if len(normalTxs) > 0 {
+			acc, _ := Sender(signer, normalTxs[0])
+			heads = append(heads, normalTxs[0])
+			// Ensure the sender address is from the signer
+			txs[acc] = normalTxs[1:]
+		}
 	}
 	heap.Init(&heads)
 
@@ -411,7 +438,7 @@ func NewTransactionsByPriceAndNonce(signer Signer, txs map[common.Address]Transa
 		txs:    txs,
 		heads:  heads,
 		signer: signer,
-	}
+	}, specialTxs
 }
 
 // Peek returns the next transaction by price.

core/types/transaction_test.go

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

eth/backend.go

@@ -25,6 +25,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"bytes"
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
@@ -189,7 +190,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	if eth.chainConfig.Posv != nil {
 		c := eth.engine.(*posv.Posv)
 
-		// Inject hook for send tx sign to smartcontract after insert block into chain.
+		// Hook sends tx sign to smartcontract after inserting block to chain.
 		importedHook := func(block *types.Block) {
 			snap, err := c.GetSnapshot(eth.blockchain, block.Header())
 			if err != nil {
@@ -209,42 +210,20 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 		}
 		eth.protocolManager.fetcher.SetImportedHook(importedHook)
 
-		// Hook will process when preparing block.
+		// Hook prepares validators M2 for the current epoch
-		c.HookPrepare = func(header *types.Header, signers []common.Address) error {
+		c.HookValidator = func(header *types.Header, signers []common.Address) error {
-			client, err := eth.blockchain.GetClient()
-			if err != nil {
-				log.Error("Fail to connect IPC client for penalty.", "error", err)
-			}
 			number := header.Number.Int64()
-			// Check m2 exists on chaindb.
-			// Get secrets and opening at epoc block checkpoint.
 			if number > 0 && number%common.EpocBlockRandomize == 0 {
-				var candidates []int64
+				validators, err := GetValidators(eth.blockchain, signers)
-				lenSigners := int64(len(signers))
+				if err != nil {
-
+					return err
-				if lenSigners > 0 {
-					for _, addr := range signers {
-						random, err := contracts.GetRandomizeFromContract(client, addr)
-						if err != nil {
-							log.Error("Fail to get random m2 from contract.", "error", err)
-						}
-						candidates = append(candidates, random)
-					}
-
-					// Get randomize m2 list.
-					m2, err := contracts.GenM2FromRandomize(candidates, lenSigners)
-					if err != nil {
-						log.Error("Can not get m2 from randomize SC", "error", err)
-					}
-					if len(m2) > 0 {
-						header.Validators = contracts.BuildValidatorFromM2(m2)
-						log.Debug("New set Validators", "m2", m2, "number", header.Number.Uint64())
-					}
 				}
+				header.Validators = validators
 			}
 			return nil
 		}
-		// Hook penalty.
+
+		// Hook scans for bad masternodes and decide to penalty them
 		c.HookPenalty = func(chain consensus.ChainReader, blockNumberEpoc uint64) ([]common.Address, error) {
 			client, err := eth.blockchain.GetClient()
 			if err != nil {
@@ -282,11 +261,10 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 				}
 				return penSigners, nil
 			}
-
 			return []common.Address{}, nil
 		}
 
-		// Hook reward for posv validator.
+		// Hook calculates reward for masternodes
 		c.HookReward = func(chain consensus.ChainReader, state *state.StateDB, header *types.Header) error {
 			client, err := eth.blockchain.GetClient()
 			if err != nil {
@@ -331,6 +309,21 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 
 			return nil
 		}
+
+		// Hook verifies masternodes set
+		c.HookVerifyMNs = func(header *types.Header, signers []common.Address) error {
+			number := header.Number.Int64()
+			if number > 0 && number%common.EpocBlockRandomize == 0 {
+				validators, err := GetValidators(eth.blockchain, signers)
+				if err != nil {
+					return err
+				}
+				if !bytes.Equal(header.Validators, validators) {
+					return posv.ErrInvalidCheckpointValidators
+				}
+			}
+			return nil
+		}
 	}
 
 	return eth, nil
@@ -607,3 +600,37 @@ func (s *Ethereum) Stop() error {
 
 	return nil
 }
+
+func GetValidators(bc *core.BlockChain, masternodes []common.Address) ([]byte, error) {
+	if bc.Config().Posv == nil {
+		return nil, core.ErrNotPoSV
+	}
+	client, err := bc.GetClient()
+	if err != nil {
+		return nil, err
+	}
+	// Check m2 exists on chaindb.
+	// Get secrets and opening at epoc block checkpoint.
+
+	var candidates []int64
+	if err != nil {
+		return nil, err
+	}
+	lenSigners := int64(len(masternodes))
+	if lenSigners > 0 {
+		for _, addr := range masternodes {
+			random, err := contracts.GetRandomizeFromContract(client, addr)
+			if err != nil {
+				return nil, err
+			}
+			candidates = append(candidates, random)
+		}
+		// Get randomize m2 list.
+		m2, err := contracts.GenM2FromRandomize(candidates, lenSigners)
+		if err != nil {
+			return nil, err
+		}
+		return contracts.BuildValidatorFromM2(m2), nil
+	}
+	return nil, core.ErrNotFoundM1
+}

eth/handler.go

@@ -83,6 +83,8 @@ 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
@@ -208,6 +210,11 @@ 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()
@@ -221,6 +228,7 @@ 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.
@@ -271,38 +279,40 @@ func (pm *ProtocolManager) handle(p *peer) error {
 		rw.Init(p.version)
 	}
 	// Register the peer locally
-	if err := pm.peers.Register(p); err != nil {
+	err := pm.peers.Register(p)
+	if err != nil && err != p2p.ErrAddPairPeer {
 		p.Log().Error("Ethereum peer registration failed", "err", err)
 		return err
 	}
 	defer pm.removePeer(p.id)
-
+	if err != p2p.ErrAddPairPeer {
-	// Register the peer in the downloader. If the downloader considers it banned, we disconnect
+		// Register the peer in the downloader. If the downloader considers it banned, we disconnect
-	if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
+		if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
-		return err
-	}
-	// Propagate existing transactions. new transactions appearing
-	// after this will be sent via broadcasts.
-	pm.syncTransactions(p)
-
-	// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork
-	if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {
-		// Request the peer's DAO fork header for extra-data validation
-		if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil {
 			return err
 		}
-		// Start a timer to disconnect if the peer doesn't reply in time
+		// Propagate existing transactions. new transactions appearing
-		p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() {
+		// after this will be sent via broadcasts.
-			p.Log().Debug("Timed out DAO fork-check, dropping")
+		pm.syncTransactions(p)
-			pm.removePeer(p.id)
+
-		})
+		// If we're DAO hard-fork aware, validate any remote peer with regard to the hard-fork
-		// Make sure it's cleaned up if the peer dies off
+		if daoBlock := pm.chainconfig.DAOForkBlock; daoBlock != nil {
-		defer func() {
+			// Request the peer's DAO fork header for extra-data validation
-			if p.forkDrop != nil {
+			if err := p.RequestHeadersByNumber(daoBlock.Uint64(), 1, 0, false); err != nil {
-				p.forkDrop.Stop()
+				return err
-				p.forkDrop = nil
 			}
-		}()
+			// Start a timer to disconnect if the peer doesn't reply in time
+			p.forkDrop = time.AfterFunc(daoChallengeTimeout, func() {
+				p.Log().Debug("Timed out DAO fork-check, dropping")
+				pm.removePeer(p.id)
+			})
+			// Make sure it's cleaned up if the peer dies off
+			defer func() {
+				if p.forkDrop != nil {
+					p.forkDrop.Stop()
+					p.forkDrop = nil
+				}
+			}()
+		}
 	}
 	// main loop. handle incoming messages.
 	for {
@@ -724,6 +734,16 @@ 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.Debug("Broadcast special transaction", "hash", hash, "recipients", len(peers))
+}
+
 // Mined broadcast loop
 func (self *ProtocolManager) minedBroadcastLoop() {
 	// automatically stops if unsubscribe
@@ -749,6 +769,19 @@ 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,6 +128,10 @@ 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,6 +25,7 @@ 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"
@@ -54,7 +55,8 @@ type peer struct {
 	id string
 
 	*p2p.Peer
-	rw p2p.MsgReadWriter
+	rw     p2p.MsgReadWriter
+	pairRw p2p.MsgReadWriter
 
 	version  int         // Protocol version negotiated
 	forkDrop *time.Timer // Timed connection dropper if forks aren't validated in time
@@ -139,6 +141,15 @@ 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 {
@@ -156,7 +167,13 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error
 // SendNewBlock propagates an entire block to a remote peer.
 func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error {
 	p.knownBlocks.Add(block.Hash())
-	return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td})
+	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.
@@ -321,8 +338,14 @@ func (ps *peerSet) Register(p *peer) error {
 	if ps.closed {
 		return errClosed
 	}
-	if _, ok := ps.peers[p.id]; ok {
+	if existPeer, ok := ps.peers[p.id]; ok {
-		return errAlreadyRegistered
+		if existPeer.pairRw != nil {
+			return errAlreadyRegistered
+		}
+		existPeer.PairPeer = p.Peer
+		existPeer.pairRw = p.rw
+		p.PairPeer = existPeer.Peer
+		return p2p.ErrAddPairPeer
 	}
 	ps.peers[p.id] = p
 	return nil

eth/protocol.go

@@ -106,6 +106,7 @@ 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.

internal/ethapi/api.go

@@ -1210,8 +1210,8 @@ func (args *SendTxArgs) toTransaction() *types.Transaction {
 
 // submitTransaction is a helper function that submits tx to txPool and logs a message.
 func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (common.Hash, error) {
-	if tx.To() != nil && tx.To().String() == common.BlockSigners {
+	if tx.To() != nil && tx.IsSpecialTransaction() {
-		return common.Hash{}, errors.New("Dont allow transaction sent to BlockSigners smart contract via API")
+		return common.Hash{}, errors.New("Dont allow transaction sent to BlockSigners & RandomizeSMC smart contract via API")
 	}
 	if err := b.SendTx(ctx, tx); err != nil {
 		return common.Hash{}, err

miner/worker.go

@@ -270,9 +270,9 @@ func (self *worker) update() {
 				self.currentMu.Lock()
 				acc, _ := types.Sender(self.current.signer, ev.Tx)
 				txs := map[common.Address]types.Transactions{acc: {ev.Tx}}
-				txset := types.NewTransactionsByPriceAndNonce(self.current.signer, txs)
+				txset, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, txs)
 
-				self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase)
+				self.current.commitTransactions(self.mux, txset, specialTxs, self.chain, self.coinbase)
 				self.currentMu.Unlock()
 			} else {
 				// If we're mining, but nothing is being processed, wake on new transactions
@@ -280,7 +280,6 @@ func (self *worker) update() {
 					self.commitNewWork()
 				}
 			}
-
 			// System stopped
 		case <-self.txSub.Err():
 			return
@@ -552,8 +551,8 @@ func (self *worker) commitNewWork() {
 		log.Error("Failed to fetch pending transactions", "err", err)
 		return
 	}
-	txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending)
+	txs, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending)
-	work.commitTransactions(self.mux, txs, self.chain, self.coinbase)
+	work.commitTransactions(self.mux, txs, specialTxs, self.chain, self.coinbase)
 
 	// compute uncles for the new block.
 	var (
@@ -584,7 +583,7 @@ func (self *worker) commitNewWork() {
 	}
 	// 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, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
+		log.Info("Commit new mining work", "number", work.Block.Number(), "txs", work.tcount, "special txs", len(specialTxs), "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
 		self.unconfirmed.Shift(work.Block.NumberU64() - 1)
 	}
 	if work.config.Posv != nil {
@@ -623,11 +622,52 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
 	return nil
 }
 
-func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) {
+func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, specialTxs types.Transactions, bc *core.BlockChain, coinbase common.Address) {
 	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 {
+			log.Trace("Not enough gas for further transactions", "gp", gp)
+			break
+		}
+		// Error may be ignored here. The error has already been checked
+		// during transaction acceptance is the transaction pool.
+		//
+		// We use the eip155 signer regardless of the current hf.
+		from, _ := types.Sender(env.signer, tx)
+		// Check whether the tx is replay protected. If we're not in the EIP155 hf
+		// phase, start ignoring the sender until we do.
+		if tx.Protected() && !env.config.IsEIP155(env.header.Number) {
+			log.Debug("Ignoring reply protected special transaction", "hash", tx.Hash(), "eip155", env.config.EIP155Block)
+			continue
+		}
+		// Start executing the transaction
+		env.state.Prepare(tx.Hash(), common.Hash{}, env.tcount)
+		err, logs := env.commitTransaction(tx, bc, coinbase, gp)
+		switch err {
+		case core.ErrNonceTooLow:
+			// New head notification data race between the transaction pool and miner, shift
+			log.Debug("Skipping special transaction with low nonce", "sender", from, "nonce", tx.Nonce(), "to", tx.To())
+
+		case core.ErrNonceTooHigh:
+			// Reorg notification data race between the transaction pool and miner, skip account =
+			log.Debug("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...)
+			env.tcount++
+
+		default:
+			// Strange error, discard the transaction and get the next in line (note, the
+			// nonce-too-high clause will prevent us from executing in vain).
+			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 {

p2p/dial.go

@@ -266,7 +266,10 @@ func (s *dialstate) checkDial(n *discover.Node, peers map[discover.NodeID]*Peer)
 	case dialing:
 		return errAlreadyDialing
 	case peers[n.ID] != nil:
-		return errAlreadyConnected
+		exitsPeer := peers[n.ID]
+		if exitsPeer.PairPeer != nil {
+			return errAlreadyConnected
+		}
 	case s.ntab != nil && n.ID == s.ntab.Self().ID:
 		return errSelf
 	case s.netrestrict != nil && !s.netrestrict.Contains(n.IP):
@@ -300,10 +303,26 @@ func (t *dialTask) Do(srv *Server) {
 		// Try resolving the ID of static nodes if dialing failed.
 		if _, ok := err.(*dialError); ok && t.flags&staticDialedConn != 0 {
 			if t.resolve(srv) {
-				t.dial(srv, t.dest)
+				err = t.dial(srv, t.dest)
 			}
 		}
 	}
+	if err == nil {
+		err = t.dial(srv, t.dest)
+		if err != nil {
+			// Try resolving the ID of static nodes if dialing failed.
+			if _, ok := err.(*dialError); ok && t.flags&staticDialedConn != 0 {
+				if t.resolve(srv) {
+					err = t.dial(srv, t.dest)
+				}
+			}
+		}
+		if err == nil {
+			log.Trace("Dial pair connection sucess", "task", t.dest)
+		} else {
+			log.Trace("Dial pair connection error", "task", t.dest, "err", err)
+		}
+	}
 }
 
 // resolve attempts to find the current endpoint for the destination

p2p/dial_test.go

@@ -116,9 +116,9 @@ func TestDialStateDynDial(t *testing.T) {
 					}},
 				},
 				new: []task{
+					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(2)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(3)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(4)}},
-					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(5)}},
 				},
 			},
 			// Some of the dials complete but no new ones are launched yet because
@@ -164,9 +164,7 @@ func TestDialStateDynDial(t *testing.T) {
 					{rw: &conn{flags: dynDialedConn, id: uintID(4)}},
 					{rw: &conn{flags: dynDialedConn, id: uintID(5)}},
 				},
-				new: []task{
+				new: []task{},
-					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(6)}},
-				},
 			},
 			// More peers (3,4) drop off and dial for ID 6 completes.
 			// The last query result from the discovery lookup is reused
@@ -181,8 +179,8 @@ func TestDialStateDynDial(t *testing.T) {
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(6)}},
 				},
 				new: []task{
+					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(5)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(7)}},
-					&discoverTask{},
 				},
 			},
 			// Peer 7 is connected, but there still aren't enough dynamic peers
@@ -212,7 +210,7 @@ func TestDialStateDynDial(t *testing.T) {
 					&discoverTask{},
 				},
 				new: []task{
-					&discoverTask{},
+					&waitExpireTask{Duration: 14 * time.Second},
 				},
 			},
 		},
@@ -302,6 +300,9 @@ func TestDialStateDynDialBootnode(t *testing.T) {
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(4)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(5)}},
 				},
+				new: []task{
+					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(4)}},
+				},
 			},
 		},
 	})
@@ -351,10 +352,11 @@ func TestDialStateDynDialFromTable(t *testing.T) {
 					}},
 				},
 				new: []task{
+					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(1)}},
+					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(2)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(10)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(11)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(12)}},
-					&discoverTask{},
 				},
 			},
 			// Dialing nodes 3,4,5 fails. The dials from the lookup succeed.
@@ -374,6 +376,9 @@ func TestDialStateDynDialFromTable(t *testing.T) {
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(11)}},
 					&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(12)}},
 				},
+				new: []task{
+					&discoverTask{},
+				},
 			},
 			// Waiting for expiry. No waitExpireTask is launched because the
 			// discovery query is still running.
@@ -453,6 +458,8 @@ func TestDialStateStaticDial(t *testing.T) {
 					{rw: &conn{flags: dynDialedConn, id: uintID(2)}},
 				},
 				new: []task{
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}},
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}},
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}},
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(5)}},
@@ -466,6 +473,9 @@ func TestDialStateStaticDial(t *testing.T) {
 					{rw: &conn{flags: dynDialedConn, id: uintID(2)}},
 					{rw: &conn{flags: staticDialedConn, id: uintID(3)}},
 				},
+				new: []task{
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}},
+				},
 				done: []task{
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}},
 				},
@@ -485,7 +495,8 @@ func TestDialStateStaticDial(t *testing.T) {
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(5)}},
 				},
 				new: []task{
-					&waitExpireTask{Duration: 14 * time.Second},
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}},
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(5)}},
 				},
 			},
 			// Wait a round for dial history to expire, no new tasks should spawn.
@@ -506,10 +517,7 @@ func TestDialStateStaticDial(t *testing.T) {
 					{rw: &conn{flags: staticDialedConn, id: uintID(3)}},
 					{rw: &conn{flags: staticDialedConn, id: uintID(5)}},
 				},
-				new: []task{
+				new: []task{},
-					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
-					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}},
-				},
 			},
 		},
 	})
@@ -542,7 +550,8 @@ func TestDialStaticAfterReset(t *testing.T) {
 				&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
 			},
 			new: []task{
-				&waitExpireTask{Duration: 30 * time.Second},
+				&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}},
+				&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
 			},
 		},
 	}
@@ -554,7 +563,9 @@ func TestDialStaticAfterReset(t *testing.T) {
 	for _, n := range wantStatic {
 		dTest.init.removeStatic(n)
 		dTest.init.addStatic(n)
+		delete(dTest.init.dialing, n.ID)
 	}
+
 	// without removing peers they will be considered recently dialed
 	runDialTest(t, dTest)
 }
@@ -591,6 +602,10 @@ func TestDialStateCache(t *testing.T) {
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}},
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
 				},
+				new: []task{
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(1)}},
+					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
+				},
 			},
 			// A salvage task is launched to wait for node 3's history
 			// entry to expire.
@@ -602,9 +617,6 @@ func TestDialStateCache(t *testing.T) {
 				done: []task{
 					&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(3)}},
 				},
-				new: []task{
-					&waitExpireTask{Duration: 14 * time.Second},
-				},
 			},
 			// Still waiting for node 3's entry to expire in the cache.
 			{

p2p/peer.go

@@ -108,7 +108,8 @@ type Peer struct {
 	disc     chan DiscReason
 
 	// events receives message send / receive events if set
-	events *event.Feed
+	events   *event.Feed
+	PairPeer *Peer
 }
 
 // NewPeer returns a peer for testing purposes.
@@ -157,7 +158,7 @@ func (p *Peer) Disconnect(reason DiscReason) {
 
 // String implements fmt.Stringer.
 func (p *Peer) String() string {
-	return fmt.Sprintf("Peer %x %v", p.rw.id[:8], p.RemoteAddr())
+	return fmt.Sprintf("Peer %x %v ", p.rw.id[:8], p.RemoteAddr())
 }
 
 // Inbound returns true if the peer is an inbound connection
@@ -225,10 +226,12 @@ loop:
 			break loop
 		}
 	}
-
 	close(p.closed)
 	p.rw.close(reason)
 	p.wg.Wait()
+	if p.PairPeer != nil {
+		go func() { p.PairPeer.Disconnect(DiscPairPeerStop) }()
+	}
 	return remoteRequested, err
 }
 
@@ -345,6 +348,7 @@ func (p *Peer) startProtocols(writeStart <-chan struct{}, writeErr chan<- error)
 			rw = newMsgEventer(rw, p.events, p.ID(), proto.Name)
 		}
 		p.log.Trace(fmt.Sprintf("Starting protocol %s/%d", proto.Name, proto.Version))
+
 		go func() {
 			err := proto.Run(p, rw)
 			if err == nil {

p2p/peer_error.go

@@ -54,6 +54,8 @@ func (self *peerError) Error() string {
 
 var errProtocolReturned = errors.New("protocol returned")
 
+var ErrAddPairPeer = errors.New("add a pair peer")
+
 type DiscReason uint
 
 const (
@@ -69,6 +71,7 @@ const (
 	DiscUnexpectedIdentity
 	DiscSelf
 	DiscReadTimeout
+	DiscPairPeerStop
 	DiscSubprotocolError = 0x10
 )
 
@@ -85,6 +88,7 @@ var discReasonToString = [...]string{
 	DiscUnexpectedIdentity:  "unexpected identity",
 	DiscSelf:                "connected to self",
 	DiscReadTimeout:         "read timeout",
+	DiscPairPeerStop:        "pair peer connection stop",
 	DiscSubprotocolError:    "subprotocol error",
 }
 

p2p/rlpx.go

@@ -122,6 +122,7 @@ func (t *rlpx) close(err error) {
 }
 
 func (t *rlpx) doProtoHandshake(our *protoHandshake) (their *protoHandshake, err error) {
+
 	// Writing our handshake happens concurrently, we prefer
 	// returning the handshake read error. If the remote side
 	// disconnects us early with a valid reason, we should return it

p2p/server.go

@@ -286,6 +286,7 @@ func (srv *Server) PeerCount() int {
 // server is shut down. If the connection fails for any reason, the server will
 // attempt to reconnect the peer.
 func (srv *Server) AddPeer(node *discover.Node) {
+
 	select {
 	case srv.addstatic <- node:
 	case <-srv.quit:
@@ -642,9 +643,15 @@ running:
 					p.events = &srv.peerFeed
 				}
 				name := truncateName(c.name)
-				srv.log.Debug("Adding p2p peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1)
+
 				go srv.runPeer(p)
-				peers[c.id] = p
+				if peers[c.id] != nil {
+					peers[c.id].PairPeer = p
+					srv.log.Debug("Adding p2p pair peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1)
+				} else {
+					peers[c.id] = p
+					srv.log.Debug("Adding p2p peer", "name", name, "addr", c.fd.RemoteAddr(), "peers", len(peers)+1)
+				}
 				if p.Inbound() {
 					inboundCount++
 				}
@@ -708,7 +715,11 @@ func (srv *Server) encHandshakeChecks(peers map[discover.NodeID]*Peer, inboundCo
 	case !c.is(trustedConn) && c.is(inboundConn) && inboundCount >= srv.maxInboundConns():
 		return DiscTooManyPeers
 	case peers[c.id] != nil:
-		return DiscAlreadyConnected
+		exitPeer := peers[c.id]
+		if exitPeer.PairPeer != nil {
+			return DiscAlreadyConnected
+		}
+		return nil
 	case c.id == srv.Self().ID:
 		return DiscSelf
 	default:

p2p/server_test.go

@@ -153,7 +153,6 @@ func TestServerDial(t *testing.T) {
 	select {
 	case conn := <-accepted:
 		defer conn.Close()
-
 		select {
 		case peer := <-connected:
 			if peer.ID() != remid {
@@ -174,6 +173,21 @@ func TestServerDial(t *testing.T) {
 			t.Error("server did not launch peer within one second")
 		}
 
+		select {
+		case peer := <-connected:
+			if peer.ID() != remid {
+				t.Errorf("peer has wrong id")
+			}
+			if peer.Name() != "test" {
+				t.Errorf("peer has wrong name")
+			}
+			if peer.RemoteAddr().String() != conn.LocalAddr().String() {
+				t.Errorf("peer started with wrong conn: got %v, want %v",
+					peer.RemoteAddr(), conn.LocalAddr())
+			}
+		case <-time.After(1 * time.Second):
+			t.Error("server did not launch peer within one second")
+		}
 	case <-time.After(1 * time.Second):
 		t.Error("server did not connect within one second")
 	}

p2p/simulations/adapters/inproc.go

@@ -91,11 +91,12 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 	}
 
 	simNode := &SimNode{
-		ID:      id,
+		ID:        id,
-		config:  config,
+		config:    config,
-		node:    n,
+		node:      n,
-		adapter: s,
+		adapter:   s,
-		running: make(map[string]node.Service),
+		running:   make(map[string]node.Service),
+		connected: make(map[discover.NodeID]bool),
 	}
 	s.nodes[id] = simNode
 	return simNode, nil
@@ -108,12 +109,16 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
 	if !ok {
 		return nil, fmt.Errorf("unknown node: %s", dest.ID)
 	}
+	if node.connected[dest.ID] {
+		return nil, fmt.Errorf("dialed node: %s", dest.ID)
+	}
 	srv := node.Server()
 	if srv == nil {
 		return nil, fmt.Errorf("node not running: %s", dest.ID)
 	}
 	pipe1, pipe2 := net.Pipe()
 	go srv.SetupConn(pipe1, 0, nil)
+	node.connected[dest.ID] = true
 	return pipe2, nil
 }
 
@@ -151,6 +156,7 @@ type SimNode struct {
 	running      map[string]node.Service
 	client       *rpc.Client
 	registerOnce sync.Once
+	connected    map[discover.NodeID]bool
 }
 
 // Addr returns the node's discovery address