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

Merge pull request #199 from nguyenbatam/duplicate_connection_with_each_peer

Browse source 
duplicate connection with each peer
This commit is contained in:
Tuna 2018-10-23 10:33:03 +07:00 committed by GitHub
commit f7c2902315
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 358 additions and 101 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
core/tx_list.go
View file

@ -251,8 +251,9 @@ 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 (tx.To() != nil && tx.To().String() != common.RandomizeSMC) || tx.To() == nil {
if old != nil && old.IsSpecialTransaction() {
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))
// Have to ensure that the new gas price is higher than the old gas
@ -262,8 +263,6 @@ func (l *txList) Add(tx *types.Transaction, priceBump uint64) (bool, *types.Tran
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 {

66
core/tx_pool.go
View file

@ -80,6 +80,8 @@ var (
ErrOversizedData = errors.New("oversized data")
ErrZeroGasPrice = errors.New("zero gas price")
ErrDuplicateSpecialTransaction = errors.New("duplicate a specail transaction")
)
var (
@ -191,6 +193,7 @@ type TxPool struct {
chain blockChain
gasPrice *big.Int
txFeed event.Feed
specialTxFeed event.Feed
scope event.SubscriptionScope
chainHeadCh chan ChainHeadEvent
chainHeadSub event.Subscription
@ -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.

39
core/types/transaction.go
View file

@ -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
lastSpecialTx := -1
for i, tx := range accTxs {
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
acc, _ := Sender(signer, accTxs[0])
txs[acc] = accTxs[1:]
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.

2
core/types/transaction_test.go
View file

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

1
eth/backend.go
View file

@ -282,7 +282,6 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
}
return penSigners, nil
}
return []common.Address{}, nil
}

37
eth/handler.go
View file

@ -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,12 +279,13 @@ 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
if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
return err
@ -304,6 +313,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
}
}()
}
}
// main loop. handle incoming messages.
for {
if err := pm.handleMsg(p); err != nil {
@ -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 {

4
eth/helper_test.go
View file

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

25
eth/peer.go
View file

@ -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"
@ -55,6 +56,7 @@ type peer struct {
*p2p.Peer
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())
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,9 +338,15 @@ 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 {
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
}

1
eth/protocol.go
View file

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

4
internal/ethapi/api.go
View file

@ -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 {
return common.Hash{}, errors.New("Dont allow transaction sent to BlockSigners smart contract via API")
if tx.To() != nil && tx.IsSpecialTransaction() {
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

54
miner/worker.go
View file

@ -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)
work.commitTransactions(self.mux, txs, self.chain, self.coinbase)
txs, specialTxs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending)
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 {

21
p2p/dial.go
View file

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

44
p2p/dial_test.go
View file

@ -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{
&dialTask{flags: dynDialedConn, dest: &discover.Node{ID: uintID(6)}},
},
new: []task{},
},
// 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{
&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(2)}},
&dialTask{flags: staticDialedConn, dest: &discover.Node{ID: uintID(4)}},
},
new: []task{},
},
},
})
@ -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.
{

8
p2p/peer.go
View file

@ -109,6 +109,7 @@ type Peer struct {
// events receives message send / receive events if set
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 {

4
p2p/peer_error.go
View file

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

1
p2p/rlpx.go
View file

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

13
p2p/server.go
View file

@ -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)
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:
exitPeer := peers[c.id]
if exitPeer.PairPeer != nil {
return DiscAlreadyConnected
}
return nil
case c.id == srv.Self().ID:
return DiscSelf
default:

16
p2p/server_test.go
View file

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

6
p2p/simulations/adapters/inproc.go
View file

@ -96,6 +96,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
node: n,
adapter: s,
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