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

Merge 0040e916cc into 1da33028ce

Browse source
This commit is contained in:
kiel barry 2018-04-30 20:37:14 +00:00 committed by GitHub
commit 2cb0db37c1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 604 additions and 603 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

110
les/flowcontrol/manager.go
View file

@ -89,16 +89,16 @@ func NewClientManager(rcTarget, maxSimReq, maxRcSum uint64) *ClientManager {
return cm return cm
} }
func (self *ClientManager) Stop() { func (m *ClientManager) Stop() {
self.lock.Lock() m.lock.Lock()
defer self.lock.Unlock() defer m.lock.Unlock()
// signal any waiting accept routines to return false // signal any waiting accept routines to return false
self.nodes = make(map[*cmNode]struct{}) m.nodes = make(map[*cmNode]struct{})
close(self.resumeQueue) close(m.resumeQueue)
} }
func (self *ClientManager) addNode(cnode *ClientNode) *cmNode { func (m *ClientManager) addNode(cnode *ClientNode) *cmNode {
time := mclock.Now() time := mclock.Now()
node := &cmNode{ node := &cmNode{
node: cnode, node: cnode,
@ -106,28 +106,28 @@ func (self *ClientManager) addNode(cnode *ClientNode) *cmNode {
finishRecharge: time, finishRecharge: time,
rcWeight: 1, rcWeight: 1,
} }
self.lock.Lock() m.lock.Lock()
defer self.lock.Unlock() defer m.lock.Unlock()
self.nodes[node] = struct{}{} m.nodes[node] = struct{}{}
self.update(mclock.Now()) m.update(mclock.Now())
return node return node
} }
func (self *ClientManager) removeNode(node *cmNode) { func (m *ClientManager) removeNode(node *cmNode) {
self.lock.Lock() m.lock.Lock()
defer self.lock.Unlock() defer m.lock.Unlock()
time := mclock.Now() time := mclock.Now()
self.stop(node, time) m.stop(node, time)
delete(self.nodes, node) delete(m.nodes, node)
self.update(time) m.update(time)
} }
// recalc sumWeight // recalc sumWeight
func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) { func (m *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
var sumWeight, rcSum uint64 var sumWeight, rcSum uint64
for node := range self.nodes { for node := range m.nodes {
rc := node.recharging rc := node.recharging
node.update(time) node.update(time)
if rc && !node.recharging { if rc && !node.recharging {
@ -138,44 +138,44 @@ func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
} }
rcSum += uint64(node.rcValue) rcSum += uint64(node.rcValue)
} }
self.sumWeight = sumWeight m.sumWeight = sumWeight
self.rcSumValue = rcSum m.rcSumValue = rcSum
return return
} }
func (self *ClientManager) update(time mclock.AbsTime) { func (m *ClientManager) update(time mclock.AbsTime) {
for { for {
firstTime := time firstTime := time
for node := range self.nodes { for node := range m.nodes {
if node.recharging && node.finishRecharge < firstTime { if node.recharging && node.finishRecharge < firstTime {
firstTime = node.finishRecharge firstTime = node.finishRecharge
} }
} }
if self.updateNodes(firstTime) { if m.updateNodes(firstTime) {
for node := range self.nodes { for node := range m.nodes {
if node.recharging { if node.recharging {
node.set(node.serving, self.simReqCnt, self.sumWeight) node.set(node.serving, m.simReqCnt, m.sumWeight)
} }
} }
} else { } else {
self.time = time m.time = time
return return
} }
} }
} }
func (self *ClientManager) canStartReq() bool { func (m *ClientManager) canStartReq() bool {
return self.simReqCnt < self.maxSimReq && self.rcSumValue < self.maxRcSum return m.simReqCnt < m.maxSimReq && m.rcSumValue < m.maxRcSum
} }
func (self *ClientManager) queueProc() { func (m *ClientManager) queueProc() {
for rc := range self.resumeQueue { for rc := range m.resumeQueue {
for { for {
time.Sleep(time.Millisecond * 10) time.Sleep(time.Millisecond * 10)
self.lock.Lock() m.lock.Lock()
self.update(mclock.Now()) m.update(mclock.Now())
cs := self.canStartReq() cs := m.canStartReq()
self.lock.Unlock() m.lock.Unlock()
if cs { if cs {
break break
} }
@ -184,41 +184,41 @@ func (self *ClientManager) queueProc() {
} }
} }
func (self *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool { func (m *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool {
self.lock.Lock() m.lock.Lock()
defer self.lock.Unlock() defer m.lock.Unlock()
self.update(time) m.update(time)
if !self.canStartReq() { if !m.canStartReq() {
resume := make(chan bool) resume := make(chan bool)
self.lock.Unlock() m.lock.Unlock()
self.resumeQueue <- resume m.resumeQueue <- resume
<-resume <-resume
self.lock.Lock() m.lock.Lock()
if _, ok := self.nodes[node]; !ok { if _, ok := m.nodes[node]; !ok {
return false // reject if node has been removed or manager has been stopped return false // reject if node has been removed or manager has been stopped
} }
} }
self.simReqCnt++ m.simReqCnt++
node.set(true, self.simReqCnt, self.sumWeight) node.set(true, m.simReqCnt, m.sumWeight)
node.startValue = node.rcValue node.startValue = node.rcValue
self.update(self.time) m.update(m.time)
return true return true
} }
func (self *ClientManager) stop(node *cmNode, time mclock.AbsTime) { func (m *ClientManager) stop(node *cmNode, time mclock.AbsTime) {
if node.serving { if node.serving {
self.update(time) m.update(time)
self.simReqCnt-- m.simReqCnt--
node.set(false, self.simReqCnt, self.sumWeight) node.set(false, m.simReqCnt, m.sumWeight)
self.update(time) m.update(time)
} }
} }
func (self *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) { func (m *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) {
self.lock.Lock() m.lock.Lock()
defer self.lock.Unlock() defer m.lock.Unlock()
self.stop(node, time) m.stop(node, time)
return uint64(node.rcValue), uint64(node.rcValue - node.startValue) return uint64(node.rcValue), uint64(node.rcValue - node.startValue)
} }

10
les/handler.go
View file

@ -1162,15 +1162,15 @@ type NodeInfo struct {
} }
// NodeInfo retrieves some protocol metadata about the running host node. // NodeInfo retrieves some protocol metadata about the running host node.
func (self *ProtocolManager) NodeInfo() *NodeInfo { func (pm *ProtocolManager) NodeInfo() *NodeInfo {
head := self.blockchain.CurrentHeader() head := self.blockchain.CurrentHeader()
hash := head.Hash() hash := head.Hash()
return &NodeInfo{ return &NodeInfo{
Network: self.networkId, Network: pm.networkId,
Difficulty: self.blockchain.GetTd(hash, head.Number.Uint64()), Difficulty: pm.blockchain.GetTd(hash, head.Number.Uint64()),
Genesis: self.blockchain.Genesis().Hash(), Genesis: pm.blockchain.Genesis().Hash(),
Config: self.blockchain.Config(), Config: pm.blockchain.Config(),
Head: hash, Head: hash,
} }
} }

80
les/txrelay.go
View file

@ -50,47 +50,47 @@ func NewLesTxRelay(ps *peerSet, reqDist *requestDistributor) *LesTxRelay {
return r return r
} }
func (self *LesTxRelay) registerPeer(p *peer) { func (r *LesTxRelay) registerPeer(p *peer) {
self.lock.Lock() r.lock.Lock()
defer self.lock.Unlock() defer r.lock.Unlock()
self.peerList = self.ps.AllPeers() r.peerList = r.ps.AllPeers()
} }
func (self *LesTxRelay) unregisterPeer(p *peer) { func (r *LesTxRelay) unregisterPeer(p *peer) {
self.lock.Lock() r.lock.Lock()
defer self.lock.Unlock() defer r.lock.Unlock()
self.peerList = self.ps.AllPeers() r.peerList = r.ps.AllPeers()
} }
// send sends a list of transactions to at most a given number of peers at // send sends a list of transactions to at most a given number of peers at
// once, never resending any particular transaction to the same peer twice // once, never resending any particular transaction to the same peer twice
func (self *LesTxRelay) send(txs types.Transactions, count int) { func (r *LesTxRelay) send(txs types.Transactions, count int) {
sendTo := make(map[*peer]types.Transactions) sendTo := make(map[*peer]types.Transactions)
self.peerStartPos++ // rotate the starting position of the peer list r.peerStartPos++ // rotate the starting position of the peer list
if self.peerStartPos >= len(self.peerList) { if r.peerStartPos >= len(r.peerList) {
self.peerStartPos = 0 r.peerStartPos = 0
} }
for _, tx := range txs { for _, tx := range txs {
hash := tx.Hash() hash := tx.Hash()
ltr, ok := self.txSent[hash] ltr, ok := r.txSent[hash]
if !ok { if !ok {
ltr = &ltrInfo{ ltr = &ltrInfo{
tx: tx, tx: tx,
sentTo: make(map[*peer]struct{}), sentTo: make(map[*peer]struct{}),
} }
self.txSent[hash] = ltr r.txSent[hash] = ltr
self.txPending[hash] = struct{}{} r.txPending[hash] = struct{}{}
} }
if len(self.peerList) > 0 { if len(r.peerList) > 0 {
cnt := count cnt := count
pos := self.peerStartPos pos := r.peerStartPos
for { for {
peer := self.peerList[pos] peer := r.peerList[pos]
if _, ok := ltr.sentTo[peer]; !ok { if _, ok := ltr.sentTo[peer]; !ok {
sendTo[peer] = append(sendTo[peer], tx) sendTo[peer] = append(sendTo[peer], tx)
ltr.sentTo[peer] = struct{}{} ltr.sentTo[peer] = struct{}{}
@ -100,10 +100,10 @@ func (self *LesTxRelay) send(txs types.Transactions, count int) {
break // sent it to the desired number of peers break // sent it to the desired number of peers
} }
pos++ pos++
if pos == len(self.peerList) { if pos == len(r.peerList) {
pos = 0 pos = 0
} }
if pos == self.peerStartPos { if pos == r.peerStartPos {
break // tried all available peers break // tried all available peers
} }
} }
@ -130,46 +130,46 @@ func (self *LesTxRelay) send(txs types.Transactions, count int) {
return func() { peer.SendTxs(reqID, cost, ll) } return func() { peer.SendTxs(reqID, cost, ll) }
}, },
} }
self.reqDist.queue(rq) r.reqDist.queue(rq)
} }
} }
func (self *LesTxRelay) Send(txs types.Transactions) { func (r *LesTxRelay) Send(txs types.Transactions) {
self.lock.Lock() r.lock.Lock()
defer self.lock.Unlock() defer r.lock.Unlock()
self.send(txs, 3) r.send(txs, 3)
} }
func (self *LesTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) { func (r *LesTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) {
self.lock.Lock() r.lock.Lock()
defer self.lock.Unlock() defer r.lock.Unlock()
for _, hash := range mined { for _, hash := range mined {
delete(self.txPending, hash) delete(r.txPending, hash)
} }
for _, hash := range rollback { for _, hash := range rollback {
self.txPending[hash] = struct{}{} r.txPending[hash] = struct{}{}
} }
if len(self.txPending) > 0 { if len(r.txPending) > 0 {
txs := make(types.Transactions, len(self.txPending)) txs := make(types.Transactions, len(r.txPending))
i := 0 i := 0
for hash := range self.txPending { for hash := range r.txPending {
txs[i] = self.txSent[hash].tx txs[i] = r.txSent[hash].tx
i++ i++
} }
self.send(txs, 1) r.send(txs, 1)
} }
} }
func (self *LesTxRelay) Discard(hashes []common.Hash) { func (r *LesTxRelay) Discard(hashes []common.Hash) {
self.lock.Lock() r.lock.Lock()
defer self.lock.Unlock() defer r.lock.Unlock()
for _, hash := range hashes { for _, hash := range hashes {
delete(self.txSent, hash) delete(r.txSent, hash)
delete(self.txPending, hash) delete(r.txPending, hash)
} }
} }

58
miner/agent.go
View file

@ -49,70 +49,70 @@ func NewCpuAgent(chain consensus.ChainReader, engine consensus.Engine) *CpuAgent
return miner return miner
} }
func (self *CpuAgent) Work() chan<- *Work { return self.workCh } func (a *CpuAgent) Work() chan<- *Work { return a.workCh }
func (self *CpuAgent) SetReturnCh(ch chan<- *Result) { self.returnCh = ch } func (a *CpuAgent) SetReturnCh(ch chan<- *Result) { a.returnCh = ch }
func (self *CpuAgent) Stop() { func (a *CpuAgent) Stop() {
if !atomic.CompareAndSwapInt32(&self.isMining, 1, 0) { if !atomic.CompareAndSwapInt32(&a.isMining, 1, 0) {
return // agent already stopped return // agent already stopped
} }
self.stop <- struct{}{} a.stop <- struct{}{}
done: done:
// Empty work channel // Empty work channel
for { for {
select { select {
case <-self.workCh: case <-a.workCh:
default: default:
break done break done
} }
} }
} }
func (self *CpuAgent) Start() { func (a *CpuAgent) Start() {
if !atomic.CompareAndSwapInt32(&self.isMining, 0, 1) { if !atomic.CompareAndSwapInt32(&a.isMining, 0, 1) {
return // agent already started return // agent already started
} }
go self.update() go a.update()
} }
func (self *CpuAgent) update() { func (a *CpuAgent) update() {
out: out:
for { for {
select { select {
case work := <-self.workCh: case work := <-a.workCh:
self.mu.Lock() a.mu.Lock()
if self.quitCurrentOp != nil { if a.quitCurrentOp != nil {
close(self.quitCurrentOp) close(a.quitCurrentOp)
} }
self.quitCurrentOp = make(chan struct{}) a.quitCurrentOp = make(chan struct{})
go self.mine(work, self.quitCurrentOp) go a.mine(work, a.quitCurrentOp)
self.mu.Unlock() a.mu.Unlock()
case <-self.stop: case <-a.stop:
self.mu.Lock() a.mu.Lock()
if self.quitCurrentOp != nil { if a.quitCurrentOp != nil {
close(self.quitCurrentOp) close(a.quitCurrentOp)
self.quitCurrentOp = nil a.quitCurrentOp = nil
} }
self.mu.Unlock() a.mu.Unlock()
break out break out
} }
} }
} }
func (self *CpuAgent) mine(work *Work, stop <-chan struct{}) { func (a *CpuAgent) mine(work *Work, stop <-chan struct{}) {
if result, err := self.engine.Seal(self.chain, work.Block, stop); result != nil { if result, err := a.engine.Seal(a.chain, work.Block, stop); result != nil {
log.Info("Successfully sealed new block", "number", result.Number(), "hash", result.Hash()) log.Info("Successfully sealed new block", "number", result.Number(), "hash", result.Hash())
self.returnCh <- &Result{work, result} a.returnCh <- &Result{work, result}
} else { } else {
if err != nil { if err != nil {
log.Warn("Block sealing failed", "err", err) log.Warn("Block sealing failed", "err", err)
} }
self.returnCh <- nil a.returnCh <- nil
} }
} }
func (self *CpuAgent) GetHashRate() int64 { func (a *CpuAgent) GetHashRate() int64 {
if pow, ok := self.engine.(consensus.PoW); ok { if pow, ok := a.engine.(consensus.PoW); ok {
return int64(pow.Hashrate()) return int64(pow.Hashrate())
} }
return 0 return 0

80
miner/miner.go
View file

@ -75,25 +75,25 @@ func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine con
// It's entered once and as soon as `Done` or `Failed` has been broadcasted the events are unregistered and // It's entered once and as soon as `Done` or `Failed` has been broadcasted the events are unregistered and
// the loop is exited. This to prevent a major security vuln where external parties can DOS you with blocks // the loop is exited. This to prevent a major security vuln where external parties can DOS you with blocks
// and halt your mining operation for as long as the DOS continues. // and halt your mining operation for as long as the DOS continues.
func (self *Miner) update() { func (m *Miner) update() {
events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{}) events := m.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
out: out:
for ev := range events.Chan() { for ev := range events.Chan() {
switch ev.Data.(type) { switch ev.Data.(type) {
case downloader.StartEvent: case downloader.StartEvent:
atomic.StoreInt32(&self.canStart, 0) atomic.StoreInt32(&m.canStart, 0)
if self.Mining() { if m.Mining() {
self.Stop() m.Stop()
atomic.StoreInt32(&self.shouldStart, 1) atomic.StoreInt32(&m.shouldStart, 1)
log.Info("Mining aborted due to sync") log.Info("Mining aborted due to sync")
} }
case downloader.DoneEvent, downloader.FailedEvent: case downloader.DoneEvent, downloader.FailedEvent:
shouldStart := atomic.LoadInt32(&self.shouldStart) == 1 shouldStart := atomic.LoadInt32(&m.shouldStart) == 1
atomic.StoreInt32(&self.canStart, 1) atomic.StoreInt32(&m.canStart, 1)
atomic.StoreInt32(&self.shouldStart, 0) atomic.StoreInt32(&m.shouldStart, 0)
if shouldStart { if shouldStart {
self.Start(self.coinbase) m.Start(m.coinbase)
} }
// unsubscribe. we're only interested in this event once // unsubscribe. we're only interested in this event once
events.Unsubscribe() events.Unsubscribe()
@ -103,50 +103,50 @@ out:
} }
} }
func (self *Miner) Start(coinbase common.Address) { func (m *Miner) Start(coinbase common.Address) {
atomic.StoreInt32(&self.shouldStart, 1) atomic.StoreInt32(&m.shouldStart, 1)
self.SetEtherbase(coinbase) m.SetEtherbase(coinbase)
if atomic.LoadInt32(&self.canStart) == 0 { if atomic.LoadInt32(&m.canStart) == 0 {
log.Info("Network syncing, will start miner afterwards") log.Info("Network syncing, will start miner afterwards")
return return
} }
atomic.StoreInt32(&self.mining, 1) atomic.StoreInt32(&m.mining, 1)
log.Info("Starting mining operation") log.Info("Starting mining operation")
self.worker.start() m.worker.start()
self.worker.commitNewWork() m.worker.commitNewWork()
} }
func (self *Miner) Stop() { func (m *Miner) Stop() {
self.worker.stop() m.worker.stop()
atomic.StoreInt32(&self.mining, 0) atomic.StoreInt32(&m.mining, 0)
atomic.StoreInt32(&self.shouldStart, 0) atomic.StoreInt32(&m.shouldStart, 0)
} }
func (self *Miner) Register(agent Agent) { func (m *Miner) Register(agent Agent) {
if self.Mining() { if m.Mining() {
agent.Start() agent.Start()
} }
self.worker.register(agent) m.worker.register(agent)
} }
func (self *Miner) Unregister(agent Agent) { func (m *Miner) Unregister(agent Agent) {
self.worker.unregister(agent) m.worker.unregister(agent)
} }
func (self *Miner) Mining() bool { func (m *Miner) Mining() bool {
return atomic.LoadInt32(&self.mining) > 0 return atomic.LoadInt32(&m.mining) > 0
} }
func (self *Miner) HashRate() (tot int64) { func (m *Miner) HashRate() (tot int64) {
if pow, ok := self.engine.(consensus.PoW); ok { if pow, ok := m.engine.(consensus.PoW); ok {
tot += int64(pow.Hashrate()) tot += int64(pow.Hashrate())
} }
// do we care this might race? is it worth we're rewriting some // do we care this might race? is it worth we're rewriting some
// aspects of the worker/locking up agents so we can get an accurate // aspects of the worker/locking up agents so we can get an accurate
// hashrate? // hashrate?
for agent := range self.worker.agents { for agent := range m.worker.agents {
if _, ok := agent.(*CpuAgent); !ok { if _, ok := agent.(*CpuAgent); !ok {
tot += agent.GetHashRate() tot += agent.GetHashRate()
} }
@ -154,17 +154,17 @@ func (self *Miner) HashRate() (tot int64) {
return return
} }
func (self *Miner) SetExtra(extra []byte) error { func (m *Miner) SetExtra(extra []byte) error {
if uint64(len(extra)) > params.MaximumExtraDataSize { if uint64(len(extra)) > params.MaximumExtraDataSize {
return fmt.Errorf("Extra exceeds max length. %d > %v", len(extra), params.MaximumExtraDataSize) return fmt.Errorf("Extra exceeds max length. %d > %v", len(extra), params.MaximumExtraDataSize)
} }
self.worker.setExtra(extra) m.worker.setExtra(extra)
return nil return nil
} }
// Pending returns the currently pending block and associated state. // Pending returns the currently pending block and associated state.
func (self *Miner) Pending() (*types.Block, *state.StateDB) { func (m *Miner) Pending() (*types.Block, *state.StateDB) {
return self.worker.pending() return m.worker.pending()
} }
// PendingBlock returns the currently pending block. // PendingBlock returns the currently pending block.
@ -172,11 +172,11 @@ func (self *Miner) Pending() (*types.Block, *state.StateDB) {
// Note, to access both the pending block and the pending state // Note, to access both the pending block and the pending state
// simultaneously, please use Pending(), as the pending state can // simultaneously, please use Pending(), as the pending state can
// change between multiple method calls // change between multiple method calls
func (self *Miner) PendingBlock() *types.Block { func (m *Miner) PendingBlock() *types.Block {
return self.worker.pendingBlock() return m.worker.pendingBlock()
} }
func (self *Miner) SetEtherbase(addr common.Address) { func (m *Miner) SetEtherbase(addr common.Address) {
self.coinbase = addr m.coinbase = addr
self.worker.setEtherbase(addr) m.worker.setEtherbase(addr)
} }

250
miner/worker.go
View file

@ -162,137 +162,137 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
return worker return worker
} }
func (self *worker) setEtherbase(addr common.Address) { func (w *worker) setEtherbase(addr common.Address) {
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
self.coinbase = addr w.coinbase = addr
} }
func (self *worker) setExtra(extra []byte) { func (w *worker) setExtra(extra []byte) {
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
self.extra = extra w.extra = extra
} }
func (self *worker) pending() (*types.Block, *state.StateDB) { func (w *worker) pending() (*types.Block, *state.StateDB) {
if atomic.LoadInt32(&self.mining) == 0 { if atomic.LoadInt32(&w.mining) == 0 {
// return a snapshot to avoid contention on currentMu mutex // return a snapshot to avoid contention on currentMu mutex
self.snapshotMu.RLock() w.snapshotMu.RLock()
defer self.snapshotMu.RUnlock() defer w.snapshotMu.RUnlock()
return self.snapshotBlock, self.snapshotState.Copy() return w.snapshotBlock, w.snapshotState.Copy()
} }
self.currentMu.Lock() w.currentMu.Lock()
defer self.currentMu.Unlock() defer w.currentMu.Unlock()
return self.current.Block, self.current.state.Copy() return w.current.Block, w.current.state.Copy()
} }
func (self *worker) pendingBlock() *types.Block { func (w *worker) pendingBlock() *types.Block {
if atomic.LoadInt32(&self.mining) == 0 { if atomic.LoadInt32(&w.mining) == 0 {
// return a snapshot to avoid contention on currentMu mutex // return a snapshot to avoid contention on currentMu mutex
self.snapshotMu.RLock() w.snapshotMu.RLock()
defer self.snapshotMu.RUnlock() defer w.snapshotMu.RUnlock()
return self.snapshotBlock return w.snapshotBlock
} }
self.currentMu.Lock() w.currentMu.Lock()
defer self.currentMu.Unlock() defer w.currentMu.Unlock()
return self.current.Block return w.current.Block
} }
func (self *worker) start() { func (w *worker) start() {
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
atomic.StoreInt32(&self.mining, 1) atomic.StoreInt32(&w.mining, 1)
// spin up agents // spin up agents
for agent := range self.agents { for agent := range w.agents {
agent.Start() agent.Start()
} }
} }
func (self *worker) stop() { func (w *worker) stop() {
self.wg.Wait() w.wg.Wait()
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
if atomic.LoadInt32(&self.mining) == 1 { if atomic.LoadInt32(&w.mining) == 1 {
for agent := range self.agents { for agent := range w.agents {
agent.Stop() agent.Stop()
} }
} }
atomic.StoreInt32(&self.mining, 0) atomic.StoreInt32(&w.mining, 0)
atomic.StoreInt32(&self.atWork, 0) atomic.StoreInt32(&w.atWork, 0)
} }
func (self *worker) register(agent Agent) { func (w *worker) register(agent Agent) {
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
self.agents[agent] = struct{}{} w.agents[agent] = struct{}{}
agent.SetReturnCh(self.recv) agent.SetReturnCh(w.recv)
} }
func (self *worker) unregister(agent Agent) { func (w *worker) unregister(agent Agent) {
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
delete(self.agents, agent) delete(w.agents, agent)
agent.Stop() agent.Stop()
} }
func (self *worker) update() { func (w *worker) update() {
defer self.txSub.Unsubscribe() defer w.txSub.Unsubscribe()
defer self.chainHeadSub.Unsubscribe() defer w.chainHeadSub.Unsubscribe()
defer self.chainSideSub.Unsubscribe() defer w.chainSideSub.Unsubscribe()
for { for {
// A real event arrived, process interesting content // A real event arrived, process interesting content
select { select {
// Handle ChainHeadEvent // Handle ChainHeadEvent
case <-self.chainHeadCh: case <-w.chainHeadCh:
self.commitNewWork() w.commitNewWork()
// Handle ChainSideEvent // Handle ChainSideEvent
case ev := <-self.chainSideCh: case ev := <-w.chainSideCh:
self.uncleMu.Lock() w.uncleMu.Lock()
self.possibleUncles[ev.Block.Hash()] = ev.Block w.possibleUncles[ev.Block.Hash()] = ev.Block
self.uncleMu.Unlock() w.uncleMu.Unlock()
// Handle TxPreEvent // Handle TxPreEvent
case ev := <-self.txCh: case ev := <-w.txCh:
// Apply transaction to the pending state if we're not mining // Apply transaction to the pending state if we're not mining
if atomic.LoadInt32(&self.mining) == 0 { if atomic.LoadInt32(&w.mining) == 0 {
self.currentMu.Lock() w.currentMu.Lock()
acc, _ := types.Sender(self.current.signer, ev.Tx) acc, _ := types.Sender(w.current.signer, ev.Tx)
txs := map[common.Address]types.Transactions{acc: {ev.Tx}} txs := map[common.Address]types.Transactions{acc: {ev.Tx}}
txset := types.NewTransactionsByPriceAndNonce(self.current.signer, txs) txset := types.NewTransactionsByPriceAndNonce(w.current.signer, txs)
self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase) w.current.commitTransactions(w.mux, txset, w.chain, w.coinbase)
self.updateSnapshot() w.updateSnapshot()
self.currentMu.Unlock() w.currentMu.Unlock()
} else { } else {
// If we're mining, but nothing is being processed, wake on new transactions // If we're mining, but nothing is being processed, wake on new transactions
if self.config.Clique != nil && self.config.Clique.Period == 0 { if w.config.Clique != nil && w.config.Clique.Period == 0 {
self.commitNewWork() w.commitNewWork()
} }
} }
// System stopped // System stopped
case <-self.txSub.Err(): case <-w.txSub.Err():
return return
case <-self.chainHeadSub.Err(): case <-w.chainHeadSub.Err():
return return
case <-self.chainSideSub.Err(): case <-w.chainSideSub.Err():
return return
} }
} }
} }
func (self *worker) wait() { func (w *worker) wait() {
for { for {
mustCommitNewWork := true mustCommitNewWork := true
for result := range self.recv { for result := range w.recv {
atomic.AddInt32(&self.atWork, -1) atomic.AddInt32(&w.atWork, -1)
if result == nil { if result == nil {
continue continue
@ -310,7 +310,7 @@ func (self *worker) wait() {
for _, log := range work.state.Logs() { for _, log := range work.state.Logs() {
log.BlockHash = block.Hash() log.BlockHash = block.Hash()
} }
stat, err := self.chain.WriteBlockWithState(block, work.receipts, work.state) stat, err := w.chain.WriteBlockWithState(block, work.receipts, work.state)
if err != nil { if err != nil {
log.Error("Failed writing block to chain", "err", err) log.Error("Failed writing block to chain", "err", err)
continue continue
@ -321,7 +321,7 @@ func (self *worker) wait() {
mustCommitNewWork = false mustCommitNewWork = false
} }
// Broadcast the block and announce chain insertion event // Broadcast the block and announce chain insertion event
self.mux.Post(core.NewMinedBlockEvent{Block: block}) w.mux.Post(core.NewMinedBlockEvent{Block: block})
var ( var (
events []interface{} events []interface{}
logs = work.state.Logs() logs = work.state.Logs()
@ -330,25 +330,25 @@ func (self *worker) wait() {
if stat == core.CanonStatTy { if stat == core.CanonStatTy {
events = append(events, core.ChainHeadEvent{Block: block}) events = append(events, core.ChainHeadEvent{Block: block})
} }
self.chain.PostChainEvents(events, logs) w.chain.PostChainEvents(events, logs)
// Insert the block into the set of pending ones to wait for confirmations // Insert the block into the set of pending ones to wait for confirmations
self.unconfirmed.Insert(block.NumberU64(), block.Hash()) w.unconfirmed.Insert(block.NumberU64(), block.Hash())
if mustCommitNewWork { if mustCommitNewWork {
self.commitNewWork() w.commitNewWork()
} }
} }
} }
} }
// push sends a new work task to currently live miner agents. // push sends a new work task to currently live miner agents.
func (self *worker) push(work *Work) { func (w *worker) push(work *Work) {
if atomic.LoadInt32(&self.mining) != 1 { if atomic.LoadInt32(&w.mining) != 1 {
return return
} }
for agent := range self.agents { for agent := range w.agents {
atomic.AddInt32(&self.atWork, 1) atomic.AddInt32(&w.atWork, 1)
if ch := agent.Work(); ch != nil { if ch := agent.Work(); ch != nil {
ch <- work ch <- work
} }
@ -356,14 +356,14 @@ func (self *worker) push(work *Work) {
} }
// makeCurrent creates a new environment for the current cycle. // makeCurrent creates a new environment for the current cycle.
func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error { func (w *worker) makeCurrent(parent *types.Block, header *types.Header) error {
state, err := self.chain.StateAt(parent.Root()) state, err := w.chain.StateAt(parent.Root())
if err != nil { if err != nil {
return err return err
} }
work := &Work{ work := &Work{
config: self.config, config: w.config,
signer: types.NewEIP155Signer(self.config.ChainId), signer: types.NewEIP155Signer(w.config.ChainId),
state: state, state: state,
ancestors: set.New(), ancestors: set.New(),
family: set.New(), family: set.New(),
@ -373,7 +373,7 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
} }
// when 08 is processed ancestors contain 07 (quick block) // when 08 is processed ancestors contain 07 (quick block)
for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) { for _, ancestor := range w.chain.GetBlocksFromHash(parent.Hash(), 7) {
for _, uncle := range ancestor.Uncles() { for _, uncle := range ancestor.Uncles() {
work.family.Add(uncle.Hash()) work.family.Add(uncle.Hash())
} }
@ -383,20 +383,20 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
// Keep track of transactions which return errors so they can be removed // Keep track of transactions which return errors so they can be removed
work.tcount = 0 work.tcount = 0
self.current = work w.current = work
return nil return nil
} }
func (self *worker) commitNewWork() { func (w *worker) commitNewWork() {
self.mu.Lock() w.mu.Lock()
defer self.mu.Unlock() defer w.mu.Unlock()
self.uncleMu.Lock() w.uncleMu.Lock()
defer self.uncleMu.Unlock() defer w.uncleMu.Unlock()
self.currentMu.Lock() w.currentMu.Lock()
defer self.currentMu.Unlock() defer w.currentMu.Unlock()
tstart := time.Now() tstart := time.Now()
parent := self.chain.CurrentBlock() parent := w.chain.CurrentBlock()
tstamp := tstart.Unix() tstamp := tstart.Unix()
if parent.Time().Cmp(new(big.Int).SetInt64(tstamp)) >= 0 { if parent.Time().Cmp(new(big.Int).SetInt64(tstamp)) >= 0 {
@ -414,24 +414,24 @@ func (self *worker) commitNewWork() {
ParentHash: parent.Hash(), ParentHash: parent.Hash(),
Number: num.Add(num, common.Big1), Number: num.Add(num, common.Big1),
GasLimit: core.CalcGasLimit(parent), GasLimit: core.CalcGasLimit(parent),
Extra: self.extra, Extra: w.extra,
Time: big.NewInt(tstamp), Time: big.NewInt(tstamp),
} }
// Only set the coinbase if we are mining (avoid spurious block rewards) // Only set the coinbase if we are mining (avoid spurious block rewards)
if atomic.LoadInt32(&self.mining) == 1 { if atomic.LoadInt32(&w.mining) == 1 {
header.Coinbase = self.coinbase header.Coinbase = w.coinbase
} }
if err := self.engine.Prepare(self.chain, header); err != nil { if err := w.engine.Prepare(w.chain, header); err != nil {
log.Error("Failed to prepare header for mining", "err", err) log.Error("Failed to prepare header for mining", "err", err)
return return
} }
// If we are care about TheDAO hard-fork check whether to override the extra-data or not // If we are care about TheDAO hard-fork check whether to override the extra-data or not
if daoBlock := self.config.DAOForkBlock; daoBlock != nil { if daoBlock := w.config.DAOForkBlock; daoBlock != nil {
// Check whether the block is among the fork extra-override range // Check whether the block is among the fork extra-override range
limit := new(big.Int).Add(daoBlock, params.DAOForkExtraRange) limit := new(big.Int).Add(daoBlock, params.DAOForkExtraRange)
if header.Number.Cmp(daoBlock) >= 0 && header.Number.Cmp(limit) < 0 { if header.Number.Cmp(daoBlock) >= 0 && header.Number.Cmp(limit) < 0 {
// Depending whether we support or oppose the fork, override differently // Depending whether we support or oppose the fork, override differently
if self.config.DAOForkSupport { if w.config.DAOForkSupport {
header.Extra = common.CopyBytes(params.DAOForkBlockExtra) header.Extra = common.CopyBytes(params.DAOForkBlockExtra)
} else if bytes.Equal(header.Extra, params.DAOForkBlockExtra) { } else if bytes.Equal(header.Extra, params.DAOForkBlockExtra) {
header.Extra = []byte{} // If miner opposes, don't let it use the reserved extra-data header.Extra = []byte{} // If miner opposes, don't let it use the reserved extra-data
@ -439,34 +439,34 @@ func (self *worker) commitNewWork() {
} }
} }
// Could potentially happen if starting to mine in an odd state. // Could potentially happen if starting to mine in an odd state.
err := self.makeCurrent(parent, header) err := w.makeCurrent(parent, header)
if err != nil { if err != nil {
log.Error("Failed to create mining context", "err", err) log.Error("Failed to create mining context", "err", err)
return return
} }
// Create the current work task and check any fork transitions needed // Create the current work task and check any fork transitions needed
work := self.current work := w.current
if self.config.DAOForkSupport && self.config.DAOForkBlock != nil && self.config.DAOForkBlock.Cmp(header.Number) == 0 { if w.config.DAOForkSupport && w.config.DAOForkBlock != nil && w.config.DAOForkBlock.Cmp(header.Number) == 0 {
misc.ApplyDAOHardFork(work.state) misc.ApplyDAOHardFork(work.state)
} }
pending, err := self.eth.TxPool().Pending() pending, err := w.eth.TxPool().Pending()
if err != nil { if err != nil {
log.Error("Failed to fetch pending transactions", "err", err) log.Error("Failed to fetch pending transactions", "err", err)
return return
} }
txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending) txs := types.NewTransactionsByPriceAndNonce(w.current.signer, pending)
work.commitTransactions(self.mux, txs, self.chain, self.coinbase) work.commitTransactions(w.mux, txs, w.chain, w.coinbase)
// compute uncles for the new block. // compute uncles for the new block.
var ( var (
uncles []*types.Header uncles []*types.Header
badUncles []common.Hash badUncles []common.Hash
) )
for hash, uncle := range self.possibleUncles { for hash, uncle := range w.possibleUncles {
if len(uncles) == 2 { if len(uncles) == 2 {
break break
} }
if err := self.commitUncle(work, uncle.Header()); err != nil { if err := w.commitUncle(work, uncle.Header()); err != nil {
log.Trace("Bad uncle found and will be removed", "hash", hash) log.Trace("Bad uncle found and will be removed", "hash", hash)
log.Trace(fmt.Sprint(uncle)) log.Trace(fmt.Sprint(uncle))
@ -477,23 +477,23 @@ func (self *worker) commitNewWork() {
} }
} }
for _, hash := range badUncles { for _, hash := range badUncles {
delete(self.possibleUncles, hash) delete(w.possibleUncles, hash)
} }
// Create the new block to seal with the consensus engine // 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 { if work.Block, err = w.engine.Finalize(w.chain, header, work.state, work.txs, uncles, work.receipts); err != nil {
log.Error("Failed to finalize block for sealing", "err", err) log.Error("Failed to finalize block for sealing", "err", err)
return return
} }
// We only care about logging if we're actually mining. // We only care about logging if we're actually mining.
if atomic.LoadInt32(&self.mining) == 1 { if atomic.LoadInt32(&w.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, "uncles", len(uncles), "elapsed", common.PrettyDuration(time.Since(tstart)))
self.unconfirmed.Shift(work.Block.NumberU64() - 1) w.unconfirmed.Shift(work.Block.NumberU64() - 1)
} }
self.push(work) w.push(work)
self.updateSnapshot() w.updateSnapshot()
} }
func (self *worker) commitUncle(work *Work, uncle *types.Header) error { func (w *worker) commitUncle(work *Work, uncle *types.Header) error {
hash := uncle.Hash() hash := uncle.Hash()
if work.uncles.Has(hash) { if work.uncles.Has(hash) {
return fmt.Errorf("uncle not unique") return fmt.Errorf("uncle not unique")
@ -508,17 +508,17 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
return nil return nil
} }
func (self *worker) updateSnapshot() { func (w *worker) updateSnapshot() {
self.snapshotMu.Lock() w.snapshotMu.Lock()
defer self.snapshotMu.Unlock() defer w.snapshotMu.Unlock()
self.snapshotBlock = types.NewBlock( w.snapshotBlock = types.NewBlock(
self.current.header, w.current.header,
self.current.txs, w.current.txs,
nil, nil,
self.current.receipts, w.current.receipts,
) )
self.snapshotState = self.current.state.Copy() w.snapshotState = w.current.state.Copy()
} }
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, bc *core.BlockChain, coinbase common.Address) {
@ -616,4 +616,4 @@ func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, c
env.receipts = append(env.receipts, receipt) env.receipts = append(env.receipts, receipt)
return nil, receipt.Logs return nil, receipt.Logs
} }

24
p2p/discover/table_test.go
View file

@ -562,7 +562,7 @@ type preminedTestnet struct {
dists [hashBits + 1][]NodeID dists [hashBits + 1][]NodeID
} }
func (tn *preminedTestnet) findnode(toid NodeID, toaddr *net.UDPAddr, target NodeID) ([]*Node, error) { func (net *preminedTestnet) findnode(toid NodeID, toaddr *net.UDPAddr, target NodeID) ([]*Node, error) {
// current log distance is encoded in port number // current log distance is encoded in port number
// fmt.Println("findnode query at dist", toaddr.Port) // fmt.Println("findnode query at dist", toaddr.Port)
if toaddr.Port == 0 { if toaddr.Port == 0 {
@ -570,7 +570,7 @@ func (tn *preminedTestnet) findnode(toid NodeID, toaddr *net.UDPAddr, target Nod
} }
next := uint16(toaddr.Port) - 1 next := uint16(toaddr.Port) - 1
var result []*Node var result []*Node
for i, id := range tn.dists[toaddr.Port] { for i, id := range net.dists[toaddr.Port] {
result = append(result, NewNode(id, net.ParseIP("127.0.0.1"), next, uint16(i))) result = append(result, NewNode(id, net.ParseIP("127.0.0.1"), next, uint16(i)))
} }
return result, nil return result, nil
@ -582,26 +582,26 @@ func (*preminedTestnet) ping(toid NodeID, toaddr *net.UDPAddr) error { return ni
// mine generates a testnet struct literal with nodes at // mine generates a testnet struct literal with nodes at
// various distances to the given target. // various distances to the given target.
func (n *preminedTestnet) mine(target NodeID) { func (net *preminedTestnet) mine(target NodeID) {
n.target = target net.target = target
n.targetSha = crypto.Keccak256Hash(n.target[:]) net.targetSha = crypto.Keccak256Hash(net.target[:])
found := 0 found := 0
for found < bucketSize*10 { for found < bucketSize*10 {
k := newkey() k := newkey()
id := PubkeyID(&k.PublicKey) id := PubkeyID(&k.PublicKey)
sha := crypto.Keccak256Hash(id[:]) sha := crypto.Keccak256Hash(id[:])
ld := logdist(n.targetSha, sha) ld := logdist(net.targetSha, sha)
if len(n.dists[ld]) < bucketSize { if len(net.dists[ld]) < bucketSize {
n.dists[ld] = append(n.dists[ld], id) net.dists[ld] = append(net.dists[ld], id)
fmt.Println("found ID with ld", ld) fmt.Println("found ID with ld", ld)
found++ found++
} }
} }
fmt.Println("&preminedTestnet{") fmt.Println("&preminedTestnet{")
fmt.Printf(" target: %#v,\n", n.target) fmt.Printf(" target: %#v,\n", net.target)
fmt.Printf(" targetSha: %#v,\n", n.targetSha) fmt.Printf(" targetSha: %#v,\n", net.targetSha)
fmt.Printf(" dists: [%d][]NodeID{\n", len(n.dists)) fmt.Printf(" dists: [%d][]NodeID{\n", len(net.dists))
for ld, ns := range n.dists { for ld, ns := range net.dists {
if len(ns) == 0 { if len(ns) == 0 {
continue continue
} }

50
p2p/discv5/net_test.go
View file

@ -265,11 +265,11 @@ type preminedTestnet struct {
net *Network net *Network
} }
func (tn *preminedTestnet) sendFindnode(to *Node, target NodeID) { func (net *preminedTestnet) sendFindnode(to *Node, target NodeID) {
panic("sendFindnode called") panic("sendFindnode called")
} }
func (tn *preminedTestnet) sendFindnodeHash(to *Node, target common.Hash) { func (net *preminedTestnet) sendFindnodeHash(to *Node, target common.Hash) {
// current log distance is encoded in port number // current log distance is encoded in port number
// fmt.Println("findnode query at dist", toaddr.Port) // fmt.Println("findnode query at dist", toaddr.Port)
if to.UDP <= lowPort { if to.UDP <= lowPort {
@ -277,21 +277,21 @@ func (tn *preminedTestnet) sendFindnodeHash(to *Node, target common.Hash) {
} }
next := to.UDP - 1 next := to.UDP - 1
var result []rpcNode var result []rpcNode
for i, id := range tn.dists[to.UDP-lowPort] { for i, id := range net.dists[to.UDP-lowPort] {
result = append(result, nodeToRPC(NewNode(id, net.ParseIP("10.0.2.99"), next, uint16(i)+1+lowPort))) result = append(result, nodeToRPC(NewNode(id, net.ParseIP("10.0.2.99"), next, uint16(i)+1+lowPort)))
} }
injectResponse(tn.net, to, neighborsPacket, &neighbors{Nodes: result}) injectResponse(net.net, to, neighborsPacket, &neighbors{Nodes: result})
} }
func (tn *preminedTestnet) sendPing(to *Node, addr *net.UDPAddr, topics []Topic) []byte { func (net *preminedTestnet) sendPing(to *Node, addr *net.UDPAddr, topics []Topic) []byte {
injectResponse(tn.net, to, pongPacket, &pong{ReplyTok: []byte{1}}) injectResponse(net.net, to, pongPacket, &pong{ReplyTok: []byte{1}})
return []byte{1} return []byte{1}
} }
func (tn *preminedTestnet) send(to *Node, ptype nodeEvent, data interface{}) (hash []byte) { func (net *preminedTestnet) send(to *Node, ptype nodeEvent, data interface{}) (hash []byte) {
switch ptype { switch ptype {
case pingPacket: case pingPacket:
injectResponse(tn.net, to, pongPacket, &pong{ReplyTok: []byte{1}}) injectResponse(net.net, to, pongPacket, &pong{ReplyTok: []byte{1}})
case pongPacket: case pongPacket:
// ignored // ignored
case findnodeHashPacket: case findnodeHashPacket:
@ -302,29 +302,29 @@ func (tn *preminedTestnet) send(to *Node, ptype nodeEvent, data interface{}) (ha
} }
next := to.UDP - 1 next := to.UDP - 1
var result []rpcNode var result []rpcNode
for i, id := range tn.dists[to.UDP-lowPort] { for i, id := range net.dists[to.UDP-lowPort] {
result = append(result, nodeToRPC(NewNode(id, net.ParseIP("10.0.2.99"), next, uint16(i)+1+lowPort))) result = append(result, nodeToRPC(NewNode(id, net.ParseIP("10.0.2.99"), next, uint16(i)+1+lowPort)))
} }
injectResponse(tn.net, to, neighborsPacket, &neighbors{Nodes: result}) injectResponse(net.net, to, neighborsPacket, &neighbors{Nodes: result})
default: default:
panic("send(" + ptype.String() + ")") panic("send(" + ptype.String() + ")")
} }
return []byte{2} return []byte{2}
} }
func (tn *preminedTestnet) sendNeighbours(to *Node, nodes []*Node) { func (net *preminedTestnet) sendNeighbours(to *Node, nodes []*Node) {
panic("sendNeighbours called") panic("sendNeighbours called")
} }
func (tn *preminedTestnet) sendTopicQuery(to *Node, topic Topic) { func (net *preminedTestnet) sendTopicQuery(to *Node, topic Topic) {
panic("sendTopicQuery called") panic("sendTopicQuery called")
} }
func (tn *preminedTestnet) sendTopicNodes(to *Node, queryHash common.Hash, nodes []*Node) { func (net *preminedTestnet) sendTopicNodes(to *Node, queryHash common.Hash, nodes []*Node) {
panic("sendTopicNodes called") panic("sendTopicNodes called")
} }
func (tn *preminedTestnet) sendTopicRegister(to *Node, topics []Topic, idx int, pong []byte) { func (net *preminedTestnet) sendTopicRegister(to *Node, topics []Topic, idx int, pong []byte) {
panic("sendTopicRegister called") panic("sendTopicRegister called")
} }
@ -336,26 +336,26 @@ func (*preminedTestnet) localAddr() *net.UDPAddr {
// mine generates a testnet struct literal with nodes at // mine generates a testnet struct literal with nodes at
// various distances to the given target. // various distances to the given target.
func (n *preminedTestnet) mine(target NodeID) { func (net *preminedTestnet) mine(target NodeID) {
n.target = target net.target = target
n.targetSha = crypto.Keccak256Hash(n.target[:]) net.targetSha = crypto.Keccak256Hash(net.target[:])
found := 0 found := 0
for found < bucketSize*10 { for found < bucketSize*10 {
k := newkey() k := newkey()
id := PubkeyID(&k.PublicKey) id := PubkeyID(&k.PublicKey)
sha := crypto.Keccak256Hash(id[:]) sha := crypto.Keccak256Hash(id[:])
ld := logdist(n.targetSha, sha) ld := logdist(net.targetSha, sha)
if len(n.dists[ld]) < bucketSize { if len(net.dists[ld]) < bucketSize {
n.dists[ld] = append(n.dists[ld], id) net.dists[ld] = append(net.dists[ld], id)
fmt.Println("found ID with ld", ld) fmt.Println("found ID with ld", ld)
found++ found++
} }
} }
fmt.Println("&preminedTestnet{") fmt.Println("&preminedTesnetet{")
fmt.Printf(" target: %#v,\n", n.target) fmt.Printf(" target: %#v,\n", net.target)
fmt.Printf(" targetSha: %#v,\n", n.targetSha) fmt.Printf(" targetSha: %#v,\n", net.targetSha)
fmt.Printf(" dists: [%d][]NodeID{\n", len(n.dists)) fmt.Printf(" dists: [%d][]NodeID{\n", len(net.dists))
for ld, ns := range n.dists { for ld, ns := range net.dists {
if len(ns) == 0 { if len(ns) == 0 {
continue continue
} }

8
p2p/discv5/node.go
View file

@ -315,11 +315,11 @@ func PubkeyID(pub *ecdsa.PublicKey) NodeID {
// Pubkey returns the public key represented by the node ID. // Pubkey returns the public key represented by the node ID.
// It returns an error if the ID is not a point on the curve. // It returns an error if the ID is not a point on the curve.
func (id NodeID) Pubkey() (*ecdsa.PublicKey, error) { func (n NodeID) Pubkey() (*ecdsa.PublicKey, error) {
p := &ecdsa.PublicKey{Curve: crypto.S256(), X: new(big.Int), Y: new(big.Int)} p := &ecdsa.PublicKey{Curve: crypto.S256(), X: new(big.Int), Y: new(big.Int)}
half := len(id) / 2 half := len(n) / 2
p.X.SetBytes(id[:half]) p.X.SetBytes(n[:half])
p.Y.SetBytes(id[half:]) p.Y.SetBytes(n[half:])
if !p.Curve.IsOnCurve(p.X, p.Y) { if !p.Curve.IsOnCurve(p.X, p.Y) {
return nil, errors.New("id is invalid secp256k1 curve point") return nil, errors.New("id is invalid secp256k1 curve point")
} }

4
p2p/discv5/ticket.go
View file

@ -304,8 +304,8 @@ func (s ticketRefByWaitTime) Len() int {
return len(s) return len(s)
} }
func (r ticketRef) waitTime() mclock.AbsTime { func (ref ticketRef) waitTime() mclock.AbsTime {
return r.t.regTime[r.idx] - r.t.issueTime return ref.t.regTime[ref.idx] - ref.t.issueTime
} }
// Less reports whether the element with // Less reports whether the element with

16
p2p/discv5/topic.go
View file

@ -271,15 +271,15 @@ func (t *topicTable) useTicket(node *Node, serialNo uint32, topics []Topic, idx
return false return false
} }
func (topictab *topicTable) getTicket(node *Node, topics []Topic) *ticket { func (t *topicTable) getTicket(node *Node, topics []Topic) *ticket {
topictab.collectGarbage() t.collectGarbage()
now := mclock.Now() now := mclock.Now()
n := topictab.getOrNewNode(node) n := t.getOrNewNode(node)
n.lastIssuedTicket++ n.lastIssuedTicket++
topictab.storeTicketCounters(node) t.storeTicketCounters(node)
t := &ticket{ tic := &ticket{
issueTime: now, issueTime: now,
topics: topics, topics: topics,
serial: n.lastIssuedTicket, serial: n.lastIssuedTicket,
@ -287,15 +287,15 @@ func (topictab *topicTable) getTicket(node *Node, topics []Topic) *ticket {
} }
for i, topic := range topics { for i, topic := range topics {
var waitPeriod time.Duration var waitPeriod time.Duration
if topic := topictab.topics[topic]; topic != nil { if topic := t.topics[topic]; topic != nil {
waitPeriod = topic.wcl.waitPeriod waitPeriod = topic.wcl.waitPeriod
} else { } else {
waitPeriod = minWaitPeriod waitPeriod = minWaitPeriod
} }
t.regTime[i] = now + mclock.AbsTime(waitPeriod) tic.regTime[i] = now + mclock.AbsTime(waitPeriod)
} }
return t return tic
} }
const gcInterval = time.Minute const gcInterval = time.Minute

24
p2p/message.go
View file

@ -271,15 +271,15 @@ func newMsgEventer(rw MsgReadWriter, feed *event.Feed, peerID discover.NodeID, p
// ReadMsg reads a message from the underlying MsgReadWriter and emits a // ReadMsg reads a message from the underlying MsgReadWriter and emits a
// "message received" event // "message received" event
func (self *msgEventer) ReadMsg() (Msg, error) { func (ev *msgEventer) ReadMsg() (Msg, error) {
msg, err := self.MsgReadWriter.ReadMsg() msg, err := ev.MsgReadWriter.ReadMsg()
if err != nil { if err != nil {
return msg, err return msg, err
} }
self.feed.Send(&PeerEvent{ ev.feed.Send(&PeerEvent{
Type: PeerEventTypeMsgRecv, Type: PeerEventTypeMsgRecv,
Peer: self.peerID, Peer: ev.peerID,
Protocol: self.Protocol, Protocol: ev.Protocol,
MsgCode: &msg.Code, MsgCode: &msg.Code,
MsgSize: &msg.Size, MsgSize: &msg.Size,
}) })
@ -288,15 +288,15 @@ func (self *msgEventer) ReadMsg() (Msg, error) {
// WriteMsg writes a message to the underlying MsgReadWriter and emits a // WriteMsg writes a message to the underlying MsgReadWriter and emits a
// "message sent" event // "message sent" event
func (self *msgEventer) WriteMsg(msg Msg) error { func (ev *msgEventer) WriteMsg(msg Msg) error {
err := self.MsgReadWriter.WriteMsg(msg) err := ev.MsgReadWriter.WriteMsg(msg)
if err != nil { if err != nil {
return err return err
} }
self.feed.Send(&PeerEvent{ ev.feed.Send(&PeerEvent{
Type: PeerEventTypeMsgSend, Type: PeerEventTypeMsgSend,
Peer: self.peerID, Peer: ev.peerID,
Protocol: self.Protocol, Protocol: ev.Protocol,
MsgCode: &msg.Code, MsgCode: &msg.Code,
MsgSize: &msg.Size, MsgSize: &msg.Size,
}) })
@ -305,8 +305,8 @@ func (self *msgEventer) WriteMsg(msg Msg) error {
// Close closes the underlying MsgReadWriter if it implements the io.Closer // Close closes the underlying MsgReadWriter if it implements the io.Closer
// interface // interface
func (self *msgEventer) Close() error { func (ev *msgEventer) Close() error {
if v, ok := self.MsgReadWriter.(io.Closer); ok { if v, ok := ev.MsgReadWriter.(io.Closer); ok {
return v.Close() return v.Close()
} }
return nil return nil

4
p2p/peer_error.go
View file

@ -48,8 +48,8 @@ func newPeerError(code int, format string, v ...interface{}) *peerError {
return err return err
} }
func (self *peerError) Error() string { func (pe *peerError) Error() string {
return self.message return pe.message
} }
var errProtocolReturned = errors.New("protocol returned") var errProtocolReturned = errors.New("protocol returned")

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

@ -154,30 +154,30 @@ type SimNode struct {
} }
// Addr returns the node's discovery address // Addr returns the node's discovery address
func (self *SimNode) Addr() []byte { func (sn *SimNode) Addr() []byte {
return []byte(self.Node().String()) return []byte(sn.Node().String())
} }
// Node returns a discover.Node representing the SimNode // Node returns a discover.Node representing the SimNode
func (self *SimNode) Node() *discover.Node { func (sn *SimNode) Node() *discover.Node {
return discover.NewNode(self.ID, net.IP{127, 0, 0, 1}, 30303, 30303) return discover.NewNode(sn.ID, net.IP{127, 0, 0, 1}, 30303, 30303)
} }
// Client returns an rpc.Client which can be used to communicate with the // Client returns an rpc.Client which can be used to communicate with the
// underlying services (it is set once the node has started) // underlying services (it is set once the node has started)
func (self *SimNode) Client() (*rpc.Client, error) { func (sn *SimNode) Client() (*rpc.Client, error) {
self.lock.RLock() sn.lock.RLock()
defer self.lock.RUnlock() defer sn.lock.RUnlock()
if self.client == nil { if sn.client == nil {
return nil, errors.New("node not started") return nil, errors.New("node not started")
} }
return self.client, nil return sn.client, nil
} }
// ServeRPC serves RPC requests over the given connection by creating an // ServeRPC serves RPC requests over the given connection by creating an
// in-memory client to the node's RPC server // in-memory client to the node's RPC server
func (self *SimNode) ServeRPC(conn net.Conn) error { func (sn *SimNode) ServeRPC(conn net.Conn) error {
handler, err := self.node.RPCHandler() handler, err := sn.node.RPCHandler()
if err != nil { if err != nil {
return err return err
} }
@ -187,13 +187,13 @@ func (self *SimNode) ServeRPC(conn net.Conn) error {
// Snapshots creates snapshots of the services by calling the // Snapshots creates snapshots of the services by calling the
// simulation_snapshot RPC method // simulation_snapshot RPC method
func (self *SimNode) Snapshots() (map[string][]byte, error) { func (sn *SimNode) Snapshots() (map[string][]byte, error) {
self.lock.RLock() sn.lock.RLock()
services := make(map[string]node.Service, len(self.running)) services := make(map[string]node.Service, len(sn.running))
for name, service := range self.running { for name, service := range sn.running {
services[name] = service services[name] = service
} }
self.lock.RUnlock() sn.lock.RUnlock()
if len(services) == 0 { if len(services) == 0 {
return nil, errors.New("no running services") return nil, errors.New("no running services")
} }
@ -213,23 +213,23 @@ func (self *SimNode) Snapshots() (map[string][]byte, error) {
} }
// Start registers the services and starts the underlying devp2p node // Start registers the services and starts the underlying devp2p node
func (self *SimNode) Start(snapshots map[string][]byte) error { func (sn *SimNode) Start(snapshots map[string][]byte) error {
newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) { newService := func(name string) func(ctx *node.ServiceContext) (node.Service, error) {
return func(nodeCtx *node.ServiceContext) (node.Service, error) { return func(nodeCtx *node.ServiceContext) (node.Service, error) {
ctx := &ServiceContext{ ctx := &ServiceContext{
RPCDialer: self.adapter, RPCDialer: sn.adapter,
NodeContext: nodeCtx, NodeContext: nodeCtx,
Config: self.config, Config: sn.config,
} }
if snapshots != nil { if snapshots != nil {
ctx.Snapshot = snapshots[name] ctx.Snapshot = snapshots[name]
} }
serviceFunc := self.adapter.services[name] serviceFunc := sn.adapter.services[name]
service, err := serviceFunc(ctx) service, err := serviceFunc(ctx)
if err != nil { if err != nil {
return nil, err return nil, err
} }
self.running[name] = service sn.running[name] = service
return service, nil return service, nil
} }
} }
@ -237,9 +237,9 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
// ensure we only register the services once in the case of the node // ensure we only register the services once in the case of the node
// being stopped and then started again // being stopped and then started again
var regErr error var regErr error
self.registerOnce.Do(func() { sn.registerOnce.Do(func() {
for _, name := range self.config.Services { for _, name := range sn.config.Services {
if err := self.node.Register(newService(name)); err != nil { if err := sn.node.Register(newService(name)); err != nil {
regErr = err regErr = err
return return
} }
@ -249,54 +249,55 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
return regErr return regErr
} }
if err := self.node.Start(); err != nil { if err := sn.node.Start(); err != nil {
return err return err
} }
// create an in-process RPC client // create an in-process RPC client
handler, err := self.node.RPCHandler() handler, err := sn.node.RPCHandler()
if err != nil { if err != nil {
return err return err
} }
self.lock.Lock() sn.lock.Lock()
self.client = rpc.DialInProc(handler) sn.client = rpc.DialInProc(handler)
self.lock.Unlock() sn.lock.Unlock()
return nil return nil
} }
// Stop closes the RPC client and stops the underlying devp2p node // Stop closes the RPC client and stops the underlying devp2p node
func (self *SimNode) Stop() error { func (sn *SimNode) Stop() error {
self.lock.Lock() sn.lock.Lock()
if self.client != nil { if sn.client != nil {
self.client.Close() sn.client.Close()
self.client = nil sn.client = nil
} }
self.lock.Unlock() sn.lock.Unlock()
return self.node.Stop() return sn.node.Stop()
} }
// Services returns a copy of the underlying services // Services returns a copy of the underlying services
func (self *SimNode) Services() []node.Service { func (sn *SimNode) Services() []node.Service {
self.lock.RLock() sn.lock.RLock()
defer self.lock.RUnlock() defer sn.lock.RUnlock()
services := make([]node.Service, 0, len(self.running)) services := make([]node.Service, 0, len(sn.running))
for _, service := range self.running { for _, service := range sn.running {
services = append(services, service) services = append(services, service)
} }
return services return services
} }
// Server returns the underlying p2p.Server // Server returns the underlying p2p.Server
func (self *SimNode) Server() *p2p.Server { func (sn *SimNode) Server() *p2p.Server {
return self.node.Server() return sn.node.Server()
} }
// SubscribeEvents subscribes the given channel to peer events from the // SubscribeEvents subscribes the given channel to peer events from the
// underlying p2p.Server // underlying p2p.Server
func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription { func (sn *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
srv := self.Server() srv := sn.Server()
if srv == nil { if srv == nil {
panic("node not running") panic("node not running")
} }
@ -304,12 +305,12 @@ func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription
} }
// NodeInfo returns information about the node // NodeInfo returns information about the node
func (self *SimNode) NodeInfo() *p2p.NodeInfo { func (sn *SimNode) NodeInfo() *p2p.NodeInfo {
server := self.Server() server := sn.Server()
if server == nil { if server == nil {
return &p2p.NodeInfo{ return &p2p.NodeInfo{
ID: self.ID.String(), ID: sn.ID.String(),
Enode: self.Node().String(), Enode: sn.Node().String(),
} }
} }
return server.NodeInfo() return server.NodeInfo()

8
p2p/simulations/adapters/state.go
View file

@ -20,12 +20,12 @@ type SimStateStore struct {
m map[string][]byte m map[string][]byte
} }
func (self *SimStateStore) Load(s string) ([]byte, error) { func (st *SimStateStore) Load(s string) ([]byte, error) {
return self.m[s], nil return st.m[s], nil
} }
func (self *SimStateStore) Save(s string, data []byte) error { func (st *SimStateStore) Save(s string, data []byte) error {
self.m[s] = data st.m[s] = data
return nil return nil
} }

310
p2p/simulations/network.go
View file

@ -74,22 +74,22 @@ func NewNetwork(nodeAdapter adapters.NodeAdapter, conf *NetworkConfig) *Network
} }
// Events returns the output event feed of the Network. // Events returns the output event feed of the Network.
func (self *Network) Events() *event.Feed { func (net *Network) Events() *event.Feed {
return &self.events return &net.events
} }
// NewNode adds a new node to the network with a random ID // NewNode adds a new node to the network with a random ID
func (self *Network) NewNode() (*Node, error) { func (net *Network) NewNode() (*Node, error) {
conf := adapters.RandomNodeConfig() conf := adapters.RandomNodeConfig()
conf.Services = []string{self.DefaultService} conf.Services = []string{net.DefaultService}
return self.NewNodeWithConfig(conf) return net.NewNodeWithConfig(conf)
} }
// NewNodeWithConfig adds a new node to the network with the given config, // NewNodeWithConfig adds a new node to the network with the given config,
// returning an error if a node with the same ID or name already exists // returning an error if a node with the same ID or name already exists
func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) { func (net *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
// create a random ID and PrivateKey if not set // create a random ID and PrivateKey if not set
if conf.ID == (discover.NodeID{}) { if conf.ID == (discover.NodeID{}) {
@ -100,31 +100,31 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
id := conf.ID id := conf.ID
if conf.Reachable == nil { if conf.Reachable == nil {
conf.Reachable = func(otherID discover.NodeID) bool { conf.Reachable = func(otherID discover.NodeID) bool {
_, err := self.InitConn(conf.ID, otherID) _, err := net.InitConn(conf.ID, otherID)
return err == nil return err == nil
} }
} }
// assign a name to the node if not set // assign a name to the node if not set
if conf.Name == "" { if conf.Name == "" {
conf.Name = fmt.Sprintf("node%02d", len(self.Nodes)+1) conf.Name = fmt.Sprintf("node%02d", len(net.Nodes)+1)
} }
// check the node doesn't already exist // check the node doesn't already exist
if node := self.getNode(id); node != nil { if node := net.getNode(id); node != nil {
return nil, fmt.Errorf("node with ID %q already exists", id) return nil, fmt.Errorf("node with ID %q already exists", id)
} }
if node := self.getNodeByName(conf.Name); node != nil { if node := net.getNodeByName(conf.Name); node != nil {
return nil, fmt.Errorf("node with name %q already exists", conf.Name) return nil, fmt.Errorf("node with name %q already exists", conf.Name)
} }
// if no services are configured, use the default service // if no services are configured, use the default service
if len(conf.Services) == 0 { if len(conf.Services) == 0 {
conf.Services = []string{self.DefaultService} conf.Services = []string{net.DefaultService}
} }
// use the NodeAdapter to create the node // use the NodeAdapter to create the node
adapterNode, err := self.nodeAdapter.NewNode(conf) adapterNode, err := net.nodeAdapter.NewNode(conf)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -133,27 +133,27 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
Config: conf, Config: conf,
} }
log.Trace(fmt.Sprintf("node %v created", id)) log.Trace(fmt.Sprintf("node %v created", id))
self.nodeMap[id] = len(self.Nodes) net.nodeMap[id] = len(net.Nodes)
self.Nodes = append(self.Nodes, node) net.Nodes = append(net.Nodes, node)
// emit a "control" event // emit a "control" event
self.events.Send(ControlEvent(node)) net.events.Send(ControlEvent(node))
return node, nil return node, nil
} }
// Config returns the network configuration // Config returns the network configuration
func (self *Network) Config() *NetworkConfig { func (net *Network) Config() *NetworkConfig {
return &self.NetworkConfig return &net.NetworkConfig
} }
// StartAll starts all nodes in the network // StartAll starts all nodes in the network
func (self *Network) StartAll() error { func (net *Network) StartAll() error {
for _, node := range self.Nodes { for _, node := range net.Nodes {
if node.Up { if node.Up {
continue continue
} }
if err := self.Start(node.ID()); err != nil { if err := net.Start(node.ID()); err != nil {
return err return err
} }
} }
@ -161,12 +161,12 @@ func (self *Network) StartAll() error {
} }
// StopAll stops all nodes in the network // StopAll stops all nodes in the network
func (self *Network) StopAll() error { func (net *Network) StopAll() error {
for _, node := range self.Nodes { for _, node := range net.Nodes {
if !node.Up { if !node.Up {
continue continue
} }
if err := self.Stop(node.ID()); err != nil { if err := net.Stop(node.ID()); err != nil {
return err return err
} }
} }
@ -174,21 +174,21 @@ func (self *Network) StopAll() error {
} }
// Start starts the node with the given ID // Start starts the node with the given ID
func (self *Network) Start(id discover.NodeID) error { func (net *Network) Start(id discover.NodeID) error {
return self.startWithSnapshots(id, nil) return net.startWithSnapshots(id, nil)
} }
// startWithSnapshots starts the node with the given ID using the give // startWithSnapshots starts the node with the given ID using the give
// snapshots // snapshots
func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error { func (net *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error {
node := self.GetNode(id) node := net.GetNode(id)
if node == nil { if node == nil {
return fmt.Errorf("node %v does not exist", id) return fmt.Errorf("node %v does not exist", id)
} }
if node.Up { if node.Up {
return fmt.Errorf("node %v already up", id) return fmt.Errorf("node %v already up", id)
} }
log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, self.nodeAdapter.Name())) log.Trace(fmt.Sprintf("starting node %v: %v using %v", id, node.Up, net.nodeAdapter.Name()))
if err := node.Start(snapshots); err != nil { if err := node.Start(snapshots); err != nil {
log.Warn(fmt.Sprintf("start up failed: %v", err)) log.Warn(fmt.Sprintf("start up failed: %v", err))
return err return err
@ -196,7 +196,7 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string
node.Up = true node.Up = true
log.Info(fmt.Sprintf("started node %v: %v", id, node.Up)) log.Info(fmt.Sprintf("started node %v: %v", id, node.Up))
self.events.Send(NewEvent(node)) net.events.Send(NewEvent(node))
// subscribe to peer events // subscribe to peer events
client, err := node.Client() client, err := node.Client()
@ -208,22 +208,22 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string
if err != nil { if err != nil {
return fmt.Errorf("error getting peer events for node %v: %s", id, err) return fmt.Errorf("error getting peer events for node %v: %s", id, err)
} }
go self.watchPeerEvents(id, events, sub) go net.watchPeerEvents(id, events, sub)
return nil return nil
} }
// watchPeerEvents reads peer events from the given channel and emits // watchPeerEvents reads peer events from the given channel and emits
// corresponding network events // corresponding network events
func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEvent, sub event.Subscription) { func (net *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEvent, sub event.Subscription) {
defer func() { defer func() {
sub.Unsubscribe() sub.Unsubscribe()
// assume the node is now down // assume the node is now down
self.lock.Lock() net.lock.Lock()
node := self.getNode(id) node := net.getNode(id)
node.Up = false node.Up = false
self.lock.Unlock() net.lock.Unlock()
self.events.Send(NewEvent(node)) net.events.Send(NewEvent(node))
}() }()
for { for {
select { select {
@ -235,16 +235,16 @@ func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEv
switch event.Type { switch event.Type {
case p2p.PeerEventTypeAdd: case p2p.PeerEventTypeAdd:
self.DidConnect(id, peer) net.DidConnect(id, peer)
case p2p.PeerEventTypeDrop: case p2p.PeerEventTypeDrop:
self.DidDisconnect(id, peer) net.DidDisconnect(id, peer)
case p2p.PeerEventTypeMsgSend: case p2p.PeerEventTypeMsgSend:
self.DidSend(id, peer, event.Protocol, *event.MsgCode) net.DidSend(id, peer, event.Protocol, *event.MsgCode)
case p2p.PeerEventTypeMsgRecv: case p2p.PeerEventTypeMsgRecv:
self.DidReceive(peer, id, event.Protocol, *event.MsgCode) net.DidReceive(peer, id, event.Protocol, *event.MsgCode)
} }
@ -258,8 +258,8 @@ func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEv
} }
// Stop stops the node with the given ID // Stop stops the node with the given ID
func (self *Network) Stop(id discover.NodeID) error { func (net *Network) Stop(id discover.NodeID) error {
node := self.GetNode(id) node := net.GetNode(id)
if node == nil { if node == nil {
return fmt.Errorf("node %v does not exist", id) return fmt.Errorf("node %v does not exist", id)
} }
@ -272,15 +272,15 @@ func (self *Network) Stop(id discover.NodeID) error {
node.Up = false node.Up = false
log.Info(fmt.Sprintf("stop node %v: %v", id, node.Up)) log.Info(fmt.Sprintf("stop node %v: %v", id, node.Up))
self.events.Send(ControlEvent(node)) net.events.Send(ControlEvent(node))
return nil return nil
} }
// Connect connects two nodes together by calling the "admin_addPeer" RPC // Connect connects two nodes together by calling the "admin_addPeer" RPC
// method on the "one" node so that it connects to the "other" node // method on the "one" node so that it connects to the "other" node
func (self *Network) Connect(oneID, otherID discover.NodeID) error { func (net *Network) Connect(oneID, otherID discover.NodeID) error {
log.Debug(fmt.Sprintf("connecting %s to %s", oneID, otherID)) log.Debug(fmt.Sprintf("connecting %s to %s", oneID, otherID))
conn, err := self.InitConn(oneID, otherID) conn, err := net.InitConn(oneID, otherID)
if err != nil { if err != nil {
return err return err
} }
@ -288,14 +288,14 @@ func (self *Network) Connect(oneID, otherID discover.NodeID) error {
if err != nil { if err != nil {
return err return err
} }
self.events.Send(ControlEvent(conn)) net.events.Send(ControlEvent(conn))
return client.Call(nil, "admin_addPeer", string(conn.other.Addr())) return client.Call(nil, "admin_addPeer", string(conn.other.Addr()))
} }
// Disconnect disconnects two nodes by calling the "admin_removePeer" RPC // Disconnect disconnects two nodes by calling the "admin_removePeer" RPC
// method on the "one" node so that it disconnects from the "other" node // method on the "one" node so that it disconnects from the "other" node
func (self *Network) Disconnect(oneID, otherID discover.NodeID) error { func (net *Network) Disconnect(oneID, otherID discover.NodeID) error {
conn := self.GetConn(oneID, otherID) conn := net.GetConn(oneID, otherID)
if conn == nil { if conn == nil {
return fmt.Errorf("connection between %v and %v does not exist", oneID, otherID) return fmt.Errorf("connection between %v and %v does not exist", oneID, otherID)
} }
@ -306,13 +306,13 @@ func (self *Network) Disconnect(oneID, otherID discover.NodeID) error {
if err != nil { if err != nil {
return err return err
} }
self.events.Send(ControlEvent(conn)) net.events.Send(ControlEvent(conn))
return client.Call(nil, "admin_removePeer", string(conn.other.Addr())) return client.Call(nil, "admin_removePeer", string(conn.other.Addr()))
} }
// DidConnect tracks the fact that the "one" node connected to the "other" node // DidConnect tracks the fact that the "one" node connected to the "other" node
func (self *Network) DidConnect(one, other discover.NodeID) error { func (net *Network) DidConnect(one, other discover.NodeID) error {
conn, err := self.GetOrCreateConn(one, other) conn, err := net.GetOrCreateConn(one, other)
if err != nil { if err != nil {
return fmt.Errorf("connection between %v and %v does not exist", one, other) return fmt.Errorf("connection between %v and %v does not exist", one, other)
} }
@ -320,14 +320,14 @@ func (self *Network) DidConnect(one, other discover.NodeID) error {
return fmt.Errorf("%v and %v already connected", one, other) return fmt.Errorf("%v and %v already connected", one, other)
} }
conn.Up = true conn.Up = true
self.events.Send(NewEvent(conn)) net.events.Send(NewEvent(conn))
return nil return nil
} }
// DidDisconnect tracks the fact that the "one" node disconnected from the // DidDisconnect tracks the fact that the "one" node disconnected from the
// "other" node // "other" node
func (self *Network) DidDisconnect(one, other discover.NodeID) error { func (net *Network) DidDisconnect(one, other discover.NodeID) error {
conn := self.GetConn(one, other) conn := net.GetConn(one, other)
if conn == nil { if conn == nil {
return fmt.Errorf("connection between %v and %v does not exist", one, other) return fmt.Errorf("connection between %v and %v does not exist", one, other)
} }
@ -336,12 +336,12 @@ func (self *Network) DidDisconnect(one, other discover.NodeID) error {
} }
conn.Up = false conn.Up = false
conn.initiated = time.Now().Add(-dialBanTimeout) conn.initiated = time.Now().Add(-dialBanTimeout)
self.events.Send(NewEvent(conn)) net.events.Send(NewEvent(conn))
return nil return nil
} }
// DidSend tracks the fact that "sender" sent a message to "receiver" // DidSend tracks the fact that "sender" sent a message to "receiver"
func (self *Network) DidSend(sender, receiver discover.NodeID, proto string, code uint64) error { func (net *Network) DidSend(sender, receiver discover.NodeID, proto string, code uint64) error {
msg := &Msg{ msg := &Msg{
One: sender, One: sender,
Other: receiver, Other: receiver,
@ -349,12 +349,12 @@ func (self *Network) DidSend(sender, receiver discover.NodeID, proto string, cod
Code: code, Code: code,
Received: false, Received: false,
} }
self.events.Send(NewEvent(msg)) net.events.Send(NewEvent(msg))
return nil return nil
} }
// DidReceive tracks the fact that "receiver" received a message from "sender" // DidReceive tracks the fact that "receiver" received a message from "sender"
func (self *Network) DidReceive(sender, receiver discover.NodeID, proto string, code uint64) error { func (net *Network) DidReceive(sender, receiver discover.NodeID, proto string, code uint64) error {
msg := &Msg{ msg := &Msg{
One: sender, One: sender,
Other: receiver, Other: receiver,
@ -362,36 +362,36 @@ func (self *Network) DidReceive(sender, receiver discover.NodeID, proto string,
Code: code, Code: code,
Received: true, Received: true,
} }
self.events.Send(NewEvent(msg)) net.events.Send(NewEvent(msg))
return nil return nil
} }
// GetNode gets the node with the given ID, returning nil if the node does not // GetNode gets the node with the given ID, returning nil if the node does not
// exist // exist
func (self *Network) GetNode(id discover.NodeID) *Node { func (net *Network) GetNode(id discover.NodeID) *Node {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getNode(id) return net.getNode(id)
} }
// GetNode gets the node with the given name, returning nil if the node does // GetNode gets the node with the given name, returning nil if the node does
// not exist // not exist
func (self *Network) GetNodeByName(name string) *Node { func (net *Network) GetNodeByName(name string) *Node {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getNodeByName(name) return net.getNodeByName(name)
} }
func (self *Network) getNode(id discover.NodeID) *Node { func (net *Network) getNode(id discover.NodeID) *Node {
i, found := self.nodeMap[id] i, found := net.nodeMap[id]
if !found { if !found {
return nil return nil
} }
return self.Nodes[i] return net.Nodes[i]
} }
func (self *Network) getNodeByName(name string) *Node { func (net *Network) getNodeByName(name string) *Node {
for _, node := range self.Nodes { for _, node := range net.Nodes {
if node.Config.Name == name { if node.Config.Name == name {
return node return node
} }
@ -400,40 +400,40 @@ func (self *Network) getNodeByName(name string) *Node {
} }
// GetNodes returns the existing nodes // GetNodes returns the existing nodes
func (self *Network) GetNodes() (nodes []*Node) { func (net *Network) GetNodes() (nodes []*Node) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
nodes = append(nodes, self.Nodes...) nodes = append(nodes, net.Nodes...)
return nodes return nodes
} }
// GetConn returns the connection which exists between "one" and "other" // GetConn returns the connection which exists between "one" and "other"
// regardless of which node initiated the connection // regardless of which node initiated the connection
func (self *Network) GetConn(oneID, otherID discover.NodeID) *Conn { func (net *Network) GetConn(oneID, otherID discover.NodeID) *Conn {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getConn(oneID, otherID) return net.getConn(oneID, otherID)
} }
// GetOrCreateConn is like GetConn but creates the connection if it doesn't // GetOrCreateConn is like GetConn but creates the connection if it doesn't
// already exist // already exist
func (self *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) { func (net *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
return self.getOrCreateConn(oneID, otherID) return net.getOrCreateConn(oneID, otherID)
} }
func (self *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) { func (net *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
if conn := self.getConn(oneID, otherID); conn != nil { if conn := net.getConn(oneID, otherID); conn != nil {
return conn, nil return conn, nil
} }
one := self.getNode(oneID) one := net.getNode(oneID)
if one == nil { if one == nil {
return nil, fmt.Errorf("node %v does not exist", oneID) return nil, fmt.Errorf("node %v does not exist", oneID)
} }
other := self.getNode(otherID) other := net.getNode(otherID)
if other == nil { if other == nil {
return nil, fmt.Errorf("node %v does not exist", otherID) return nil, fmt.Errorf("node %v does not exist", otherID)
} }
@ -444,18 +444,18 @@ func (self *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, err
other: other, other: other,
} }
label := ConnLabel(oneID, otherID) label := ConnLabel(oneID, otherID)
self.connMap[label] = len(self.Conns) net.connMap[label] = len(net.Conns)
self.Conns = append(self.Conns, conn) net.Conns = append(net.Conns, conn)
return conn, nil return conn, nil
} }
func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn { func (net *Network) getConn(oneID, otherID discover.NodeID) *Conn {
label := ConnLabel(oneID, otherID) label := ConnLabel(oneID, otherID)
i, found := self.connMap[label] i, found := net.connMap[label]
if !found { if !found {
return nil return nil
} }
return self.Conns[i] return net.Conns[i]
} }
// InitConn(one, other) retrieves the connectiton model for the connection between // InitConn(one, other) retrieves the connectiton model for the connection between
@ -466,13 +466,13 @@ func (self *Network) getConn(oneID, otherID discover.NodeID) *Conn {
// it also checks whether there has been recent attempt to connect the peers // it also checks whether there has been recent attempt to connect the peers
// this is cheating as the simulation is used as an oracle and know about // this is cheating as the simulation is used as an oracle and know about
// remote peers attempt to connect to a node which will then not initiate the connection // remote peers attempt to connect to a node which will then not initiate the connection
func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) { func (net *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
if oneID == otherID { if oneID == otherID {
return nil, fmt.Errorf("refusing to connect to self %v", oneID) return nil, fmt.Errorf("refusing to connect to self %v", oneID)
} }
conn, err := self.getOrCreateConn(oneID, otherID) conn, err := net.getOrCreateConn(oneID, otherID)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -491,28 +491,28 @@ func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) {
} }
// Shutdown stops all nodes in the network and closes the quit channel // Shutdown stops all nodes in the network and closes the quit channel
func (self *Network) Shutdown() { func (net *Network) Shutdown() {
for _, node := range self.Nodes { for _, node := range net.Nodes {
log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString())) log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString()))
if err := node.Stop(); err != nil { if err := node.Stop(); err != nil {
log.Warn(fmt.Sprintf("error stopping node %s", node.ID().TerminalString()), "err", err) log.Warn(fmt.Sprintf("error stopping node %s", node.ID().TerminalString()), "err", err)
} }
} }
close(self.quitc) close(net.quitc)
} }
//Reset resets all network properties: //Reset resets all network properties:
//emtpies the nodes and the connection list //emtpies the nodes and the connection list
func (self *Network) Reset() { func (net *Network) Reset() {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
//re-initialize the maps //re-initialize the maps
self.connMap = make(map[string]int) net.connMap = make(map[string]int)
self.nodeMap = make(map[discover.NodeID]int) net.nodeMap = make(map[discover.NodeID]int)
self.Nodes = nil net.Nodes = nil
self.Conns = nil net.Conns = nil
} }
// Node is a wrapper around adapters.Node which is used to track the status // Node is a wrapper around adapters.Node which is used to track the status
@ -528,37 +528,37 @@ type Node struct {
} }
// ID returns the ID of the node // ID returns the ID of the node
func (self *Node) ID() discover.NodeID { func (n *Node) ID() discover.NodeID {
return self.Config.ID return n.Config.ID
} }
// String returns a log-friendly string // String returns a log-friendly string
func (self *Node) String() string { func (n *Node) String() string {
return fmt.Sprintf("Node %v", self.ID().TerminalString()) return fmt.Sprintf("Node %v", n.ID().TerminalString())
} }
// NodeInfo returns information about the node // NodeInfo returns information about the node
func (self *Node) NodeInfo() *p2p.NodeInfo { func (n *Node) NodeInfo() *p2p.NodeInfo {
// avoid a panic if the node is not started yet // avoid a panic if the node is not started yet
if self.Node == nil { if n.Node == nil {
return nil return nil
} }
info := self.Node.NodeInfo() info := n.Node.NodeInfo()
info.Name = self.Config.Name info.Name = n.Config.Name
return info return info
} }
// MarshalJSON implements the json.Marshaler interface so that the encoded // MarshalJSON implements the json.Marshaler interface so that the encoded
// JSON includes the NodeInfo // JSON includes the NodeInfo
func (self *Node) MarshalJSON() ([]byte, error) { func (n *Node) MarshalJSON() ([]byte, error) {
return json.Marshal(struct { return json.Marshal(struct {
Info *p2p.NodeInfo `json:"info,omitempty"` Info *p2p.NodeInfo `json:"info,omitempty"`
Config *adapters.NodeConfig `json:"config,omitempty"` Config *adapters.NodeConfig `json:"config,omitempty"`
Up bool `json:"up"` Up bool `json:"up"`
}{ }{
Info: self.NodeInfo(), Info: n.NodeInfo(),
Config: self.Config, Config: n.Config,
Up: self.Up, Up: n.Up,
}) })
} }
@ -580,19 +580,19 @@ type Conn struct {
} }
// nodesUp returns whether both nodes are currently up // nodesUp returns whether both nodes are currently up
func (self *Conn) nodesUp() error { func (c *Conn) nodesUp() error {
if !self.one.Up { if !c.one.Up {
return fmt.Errorf("one %v is not up", self.One) return fmt.Errorf("one %v is not up", c.One)
} }
if !self.other.Up { if !c.other.Up {
return fmt.Errorf("other %v is not up", self.Other) return fmt.Errorf("other %v is not up", c.Other)
} }
return nil return nil
} }
// String returns a log-friendly string // String returns a log-friendly string
func (self *Conn) String() string { func (c *Conn) String() string {
return fmt.Sprintf("Conn %v->%v", self.One.TerminalString(), self.Other.TerminalString()) return fmt.Sprintf("Conn %v->%v", c.One.TerminalString(), c.Other.TerminalString())
} }
// Msg represents a p2p message sent between two nodes in the network // Msg represents a p2p message sent between two nodes in the network
@ -605,8 +605,8 @@ type Msg struct {
} }
// String returns a log-friendly string // String returns a log-friendly string
func (self *Msg) String() string { func (m *Msg) String() string {
return fmt.Sprintf("Msg(%d) %v->%v", self.Code, self.One.TerminalString(), self.Other.TerminalString()) return fmt.Sprintf("Msg(%d) %v->%v", m.Code, m.One.TerminalString(), m.Other.TerminalString())
} }
// ConnLabel generates a deterministic string which represents a connection // ConnLabel generates a deterministic string which represents a connection
@ -640,14 +640,14 @@ type NodeSnapshot struct {
} }
// Snapshot creates a network snapshot // Snapshot creates a network snapshot
func (self *Network) Snapshot() (*Snapshot, error) { func (net *Network) Snapshot() (*Snapshot, error) {
self.lock.Lock() net.lock.Lock()
defer self.lock.Unlock() defer net.lock.Unlock()
snap := &Snapshot{ snap := &Snapshot{
Nodes: make([]NodeSnapshot, len(self.Nodes)), Nodes: make([]NodeSnapshot, len(net.Nodes)),
Conns: make([]Conn, len(self.Conns)), Conns: make([]Conn, len(net.Conns)),
} }
for i, node := range self.Nodes { for i, node := range net.Nodes {
snap.Nodes[i] = NodeSnapshot{Node: *node} snap.Nodes[i] = NodeSnapshot{Node: *node}
if !node.Up { if !node.Up {
continue continue
@ -658,33 +658,33 @@ func (self *Network) Snapshot() (*Snapshot, error) {
} }
snap.Nodes[i].Snapshots = snapshots snap.Nodes[i].Snapshots = snapshots
} }
for i, conn := range self.Conns { for i, conn := range net.Conns {
snap.Conns[i] = *conn snap.Conns[i] = *conn
} }
return snap, nil return snap, nil
} }
// Load loads a network snapshot // Load loads a network snapshot
func (self *Network) Load(snap *Snapshot) error { func (net *Network) Load(snap *Snapshot) error {
for _, n := range snap.Nodes { for _, n := range snap.Nodes {
if _, err := self.NewNodeWithConfig(n.Node.Config); err != nil { if _, err := net.NewNodeWithConfig(n.Node.Config); err != nil {
return err return err
} }
if !n.Node.Up { if !n.Node.Up {
continue continue
} }
if err := self.startWithSnapshots(n.Node.Config.ID, n.Snapshots); err != nil { if err := net.startWithSnapshots(n.Node.Config.ID, n.Snapshots); err != nil {
return err return err
} }
} }
for _, conn := range snap.Conns { for _, conn := range snap.Conns {
if !self.GetNode(conn.One).Up || !self.GetNode(conn.Other).Up { if !net.GetNode(conn.One).Up || !net.GetNode(conn.Other).Up {
//in this case, at least one of the nodes of a connection is not up, //in this case, at least one of the nodes of a connection is not up,
//so it would result in the snapshot `Load` to fail //so it would result in the snapshot `Load` to fail
continue continue
} }
if err := self.Connect(conn.One, conn.Other); err != nil { if err := net.Connect(conn.One, conn.Other); err != nil {
return err return err
} }
} }
@ -692,7 +692,7 @@ func (self *Network) Load(snap *Snapshot) error {
} }
// Subscribe reads control events from a channel and executes them // Subscribe reads control events from a channel and executes them
func (self *Network) Subscribe(events chan *Event) { func (net *Network) Subscribe(events chan *Event) {
for { for {
select { select {
case event, ok := <-events: case event, ok := <-events:
@ -700,23 +700,23 @@ func (self *Network) Subscribe(events chan *Event) {
return return
} }
if event.Control { if event.Control {
self.executeControlEvent(event) net.executeControlEvent(event)
} }
case <-self.quitc: case <-net.quitc:
return return
} }
} }
} }
func (self *Network) executeControlEvent(event *Event) { func (net *Network) executeControlEvent(event *Event) {
log.Trace("execute control event", "type", event.Type, "event", event) log.Trace("execute control event", "type", event.Type, "event", event)
switch event.Type { switch event.Type {
case EventTypeNode: case EventTypeNode:
if err := self.executeNodeEvent(event); err != nil { if err := net.executeNodeEvent(event); err != nil {
log.Error("error executing node event", "event", event, "err", err) log.Error("error executing node event", "event", event, "err", err)
} }
case EventTypeConn: case EventTypeConn:
if err := self.executeConnEvent(event); err != nil { if err := net.executeConnEvent(event); err != nil {
log.Error("error executing conn event", "event", event, "err", err) log.Error("error executing conn event", "event", event, "err", err)
} }
case EventTypeMsg: case EventTypeMsg:
@ -724,21 +724,21 @@ func (self *Network) executeControlEvent(event *Event) {
} }
} }
func (self *Network) executeNodeEvent(e *Event) error { func (net *Network) executeNodeEvent(e *Event) error {
if !e.Node.Up { if !e.Node.Up {
return self.Stop(e.Node.ID()) return net.Stop(e.Node.ID())
} }
if _, err := self.NewNodeWithConfig(e.Node.Config); err != nil { if _, err := net.NewNodeWithConfig(e.Node.Config); err != nil {
return err return err
} }
return self.Start(e.Node.ID()) return net.Start(e.Node.ID())
} }
func (self *Network) executeConnEvent(e *Event) error { func (net *Network) executeConnEvent(e *Event) error {
if e.Conn.Up { if e.Conn.Up {
return self.Connect(e.Conn.One, e.Conn.Other) return net.Connect(e.Conn.One, e.Conn.Other)
} else { } else {
return self.Disconnect(e.Conn.One, e.Conn.Other) return net.Disconnect(e.Conn.One, e.Conn.Other)
} }
} }

34
p2p/testing/peerpool.go
View file

@ -39,29 +39,29 @@ func NewTestPeerPool() *TestPeerPool {
return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)} return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)}
} }
func (self *TestPeerPool) Add(p TestPeer) { func (p *TestPeerPool) Add(peer TestPeer) {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
log.Trace(fmt.Sprintf("pp add peer %v", p.ID())) log.Trace(fmt.Sprintf("pp add peer %v", peer.ID()))
self.peers[p.ID()] = p p.peers[peer.ID()] = peer
} }
func (self *TestPeerPool) Remove(p TestPeer) { func (p *TestPeerPool) Remove(peer TestPeer) {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
delete(self.peers, p.ID()) delete(p.peers, peer.ID())
} }
func (self *TestPeerPool) Has(id discover.NodeID) bool { func (p *TestPeerPool) Has(id discover.NodeID) bool {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
_, ok := self.peers[id] _, ok := p.peers[id]
return ok return ok
} }
func (self *TestPeerPool) Get(id discover.NodeID) TestPeer { func (p *TestPeerPool) Get(id discover.NodeID) TestPeer {
self.lock.Lock() p.lock.Lock()
defer self.lock.Unlock() defer p.lock.Unlock()
return self.peers[id] return p.peers[id]
} }

26
p2p/testing/protocolsession.go
View file

@ -78,10 +78,10 @@ type Disconnect struct {
} }
// trigger sends messages from peers // trigger sends messages from peers
func (self *ProtocolSession) trigger(trig Trigger) error { func (s *ProtocolSession) trigger(trig Trigger) error {
simNode, ok := self.adapter.GetNode(trig.Peer) simNode, ok := s.adapter.GetNode(trig.Peer)
if !ok { if !ok {
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(self.IDs)) return fmt.Errorf("trigger: peer %v does not exist (1- %v)", trig.Peer, len(s.IDs))
} }
mockNode, ok := simNode.Services()[0].(*mockNode) mockNode, ok := simNode.Services()[0].(*mockNode)
if !ok { if !ok {
@ -107,7 +107,7 @@ func (self *ProtocolSession) trigger(trig Trigger) error {
} }
// expect checks an expectation of a message sent out by the pivot node // expect checks an expectation of a message sent out by the pivot node
func (self *ProtocolSession) expect(exps []Expect) error { func (s *ProtocolSession) expect(exps []Expect) error {
// construct a map of expectations for each node // construct a map of expectations for each node
peerExpects := make(map[discover.NodeID][]Expect) peerExpects := make(map[discover.NodeID][]Expect)
for _, exp := range exps { for _, exp := range exps {
@ -120,9 +120,9 @@ func (self *ProtocolSession) expect(exps []Expect) error {
// construct a map of mockNodes for each node // construct a map of mockNodes for each node
mockNodes := make(map[discover.NodeID]*mockNode) mockNodes := make(map[discover.NodeID]*mockNode)
for nodeID := range peerExpects { for nodeID := range peerExpects {
simNode, ok := self.adapter.GetNode(nodeID) simNode, ok := s.adapter.GetNode(nodeID)
if !ok { if !ok {
return fmt.Errorf("trigger: peer %v does not exist (1- %v)", nodeID, len(self.IDs)) return fmt.Errorf("trigger: peer %v does not exist (1- %v)", nodeID, len(s.IDs))
} }
mockNode, ok := simNode.Services()[0].(*mockNode) mockNode, ok := simNode.Services()[0].(*mockNode)
if !ok { if !ok {
@ -202,9 +202,9 @@ func (self *ProtocolSession) expect(exps []Expect) error {
} }
// TestExchanges tests a series of exchanges against the session // TestExchanges tests a series of exchanges against the session
func (self *ProtocolSession) TestExchanges(exchanges ...Exchange) error { func (s *ProtocolSession) TestExchanges(exchanges ...Exchange) error {
for i, e := range exchanges { for i, e := range exchanges {
if err := self.testExchange(e); err != nil { if err := s.testExchange(e); err != nil {
return fmt.Errorf("exchange #%d %q: %v", i, e.Label, err) return fmt.Errorf("exchange #%d %q: %v", i, e.Label, err)
} }
log.Trace(fmt.Sprintf("exchange #%d %q: run successfully", i, e.Label)) log.Trace(fmt.Sprintf("exchange #%d %q: run successfully", i, e.Label))
@ -214,14 +214,14 @@ func (self *ProtocolSession) TestExchanges(exchanges ...Exchange) error {
// testExchange tests a single Exchange. // testExchange tests a single Exchange.
// Default timeout value is 2 seconds. // Default timeout value is 2 seconds.
func (self *ProtocolSession) testExchange(e Exchange) error { func (s *ProtocolSession) testExchange(e Exchange) error {
errc := make(chan error) errc := make(chan error)
done := make(chan struct{}) done := make(chan struct{})
defer close(done) defer close(done)
go func() { go func() {
for _, trig := range e.Triggers { for _, trig := range e.Triggers {
err := self.trigger(trig) err := s.trigger(trig)
if err != nil { if err != nil {
errc <- err errc <- err
return return
@ -229,7 +229,7 @@ func (self *ProtocolSession) testExchange(e Exchange) error {
} }
select { select {
case errc <- self.expect(e.Expects): case errc <- s.expect(e.Expects):
case <-done: case <-done:
} }
}() }()
@ -250,7 +250,7 @@ func (self *ProtocolSession) testExchange(e Exchange) error {
// TestDisconnected tests the disconnections given as arguments // TestDisconnected tests the disconnections given as arguments
// the disconnect structs describe what disconnect error is expected on which peer // the disconnect structs describe what disconnect error is expected on which peer
func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error { func (s *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error {
expects := make(map[discover.NodeID]error) expects := make(map[discover.NodeID]error)
for _, disconnect := range disconnects { for _, disconnect := range disconnects {
expects[disconnect.Peer] = disconnect.Error expects[disconnect.Peer] = disconnect.Error
@ -259,7 +259,7 @@ func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error
timeout := time.After(time.Second) timeout := time.After(time.Second)
for len(expects) > 0 { for len(expects) > 0 {
select { select {
case event := <-self.events: case event := <-s.events:
if event.Type != p2p.PeerEventTypeDrop { if event.Type != p2p.PeerEventTypeDrop {
continue continue
} }

12
p2p/testing/protocoltester.go
View file

@ -101,24 +101,24 @@ func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Pe
} }
// Stop stops the p2p server // Stop stops the p2p server
func (self *ProtocolTester) Stop() error { func (t *ProtocolTester) Stop() error {
self.Server.Stop() t.Server.Stop()
return nil return nil
} }
// Connect brings up the remote peer node and connects it using the // Connect brings up the remote peer node and connects it using the
// p2p/simulations network connection with the in memory network adapter // p2p/simulations network connection with the in memory network adapter
func (self *ProtocolTester) Connect(selfID discover.NodeID, peers ...*adapters.NodeConfig) { func (t *ProtocolTester) Connect(selfID discover.NodeID, peers ...*adapters.NodeConfig) {
for _, peer := range peers { for _, peer := range peers {
log.Trace(fmt.Sprintf("start node %v", peer.ID)) log.Trace(fmt.Sprintf("start node %v", peer.ID))
if _, err := self.network.NewNodeWithConfig(peer); err != nil { if _, err := t.network.NewNodeWithConfig(peer); err != nil {
panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err)) panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err))
} }
if err := self.network.Start(peer.ID); err != nil { if err := t.network.Start(peer.ID); err != nil {
panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err)) panic(fmt.Sprintf("error starting peer %v: %v", peer.ID, err))
} }
log.Trace(fmt.Sprintf("connect to %v", peer.ID)) log.Trace(fmt.Sprintf("connect to %v", peer.ID))
if err := self.network.Connect(selfID, peer.ID); err != nil { if err := t.network.Connect(selfID, peer.ID); err != nil {
panic(fmt.Sprintf("error connecting to peer %v: %v", peer.ID, err)) panic(fmt.Sprintf("error connecting to peer %v: %v", peer.ID, err))
} }
} }