mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-18 18:02:24 +00:00
Merge 0a57bbfa99 into 1da33028ce
This commit is contained in:
commit
58273ddb09
22 changed files with 752 additions and 751 deletions
|
|
@ -89,16 +89,16 @@ func NewClientManager(rcTarget, maxSimReq, maxRcSum uint64) *ClientManager {
|
|||
return cm
|
||||
}
|
||||
|
||||
func (self *ClientManager) Stop() {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (m *ClientManager) Stop() {
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
|
||||
// signal any waiting accept routines to return false
|
||||
self.nodes = make(map[*cmNode]struct{})
|
||||
close(self.resumeQueue)
|
||||
m.nodes = make(map[*cmNode]struct{})
|
||||
close(m.resumeQueue)
|
||||
}
|
||||
|
||||
func (self *ClientManager) addNode(cnode *ClientNode) *cmNode {
|
||||
func (m *ClientManager) addNode(cnode *ClientNode) *cmNode {
|
||||
time := mclock.Now()
|
||||
node := &cmNode{
|
||||
node: cnode,
|
||||
|
|
@ -106,28 +106,28 @@ func (self *ClientManager) addNode(cnode *ClientNode) *cmNode {
|
|||
finishRecharge: time,
|
||||
rcWeight: 1,
|
||||
}
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
|
||||
self.nodes[node] = struct{}{}
|
||||
self.update(mclock.Now())
|
||||
m.nodes[node] = struct{}{}
|
||||
m.update(mclock.Now())
|
||||
return node
|
||||
}
|
||||
|
||||
func (self *ClientManager) removeNode(node *cmNode) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (m *ClientManager) removeNode(node *cmNode) {
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
|
||||
time := mclock.Now()
|
||||
self.stop(node, time)
|
||||
delete(self.nodes, node)
|
||||
self.update(time)
|
||||
m.stop(node, time)
|
||||
delete(m.nodes, node)
|
||||
m.update(time)
|
||||
}
|
||||
|
||||
// recalc sumWeight
|
||||
func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
|
||||
func (m *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
|
||||
var sumWeight, rcSum uint64
|
||||
for node := range self.nodes {
|
||||
for node := range m.nodes {
|
||||
rc := node.recharging
|
||||
node.update(time)
|
||||
if rc && !node.recharging {
|
||||
|
|
@ -138,44 +138,44 @@ func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
|
|||
}
|
||||
rcSum += uint64(node.rcValue)
|
||||
}
|
||||
self.sumWeight = sumWeight
|
||||
self.rcSumValue = rcSum
|
||||
m.sumWeight = sumWeight
|
||||
m.rcSumValue = rcSum
|
||||
return
|
||||
}
|
||||
|
||||
func (self *ClientManager) update(time mclock.AbsTime) {
|
||||
func (m *ClientManager) update(time mclock.AbsTime) {
|
||||
for {
|
||||
firstTime := time
|
||||
for node := range self.nodes {
|
||||
for node := range m.nodes {
|
||||
if node.recharging && node.finishRecharge < firstTime {
|
||||
firstTime = node.finishRecharge
|
||||
}
|
||||
}
|
||||
if self.updateNodes(firstTime) {
|
||||
for node := range self.nodes {
|
||||
if m.updateNodes(firstTime) {
|
||||
for node := range m.nodes {
|
||||
if node.recharging {
|
||||
node.set(node.serving, self.simReqCnt, self.sumWeight)
|
||||
node.set(node.serving, m.simReqCnt, m.sumWeight)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
self.time = time
|
||||
m.time = time
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *ClientManager) canStartReq() bool {
|
||||
return self.simReqCnt < self.maxSimReq && self.rcSumValue < self.maxRcSum
|
||||
func (m *ClientManager) canStartReq() bool {
|
||||
return m.simReqCnt < m.maxSimReq && m.rcSumValue < m.maxRcSum
|
||||
}
|
||||
|
||||
func (self *ClientManager) queueProc() {
|
||||
for rc := range self.resumeQueue {
|
||||
func (m *ClientManager) queueProc() {
|
||||
for rc := range m.resumeQueue {
|
||||
for {
|
||||
time.Sleep(time.Millisecond * 10)
|
||||
self.lock.Lock()
|
||||
self.update(mclock.Now())
|
||||
cs := self.canStartReq()
|
||||
self.lock.Unlock()
|
||||
m.lock.Lock()
|
||||
m.update(mclock.Now())
|
||||
cs := m.canStartReq()
|
||||
m.lock.Unlock()
|
||||
if cs {
|
||||
break
|
||||
}
|
||||
|
|
@ -184,41 +184,41 @@ func (self *ClientManager) queueProc() {
|
|||
}
|
||||
}
|
||||
|
||||
func (self *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (m *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool {
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
|
||||
self.update(time)
|
||||
if !self.canStartReq() {
|
||||
m.update(time)
|
||||
if !m.canStartReq() {
|
||||
resume := make(chan bool)
|
||||
self.lock.Unlock()
|
||||
self.resumeQueue <- resume
|
||||
m.lock.Unlock()
|
||||
m.resumeQueue <- resume
|
||||
<-resume
|
||||
self.lock.Lock()
|
||||
if _, ok := self.nodes[node]; !ok {
|
||||
m.lock.Lock()
|
||||
if _, ok := m.nodes[node]; !ok {
|
||||
return false // reject if node has been removed or manager has been stopped
|
||||
}
|
||||
}
|
||||
self.simReqCnt++
|
||||
node.set(true, self.simReqCnt, self.sumWeight)
|
||||
m.simReqCnt++
|
||||
node.set(true, m.simReqCnt, m.sumWeight)
|
||||
node.startValue = node.rcValue
|
||||
self.update(self.time)
|
||||
m.update(m.time)
|
||||
return true
|
||||
}
|
||||
|
||||
func (self *ClientManager) stop(node *cmNode, time mclock.AbsTime) {
|
||||
func (m *ClientManager) stop(node *cmNode, time mclock.AbsTime) {
|
||||
if node.serving {
|
||||
self.update(time)
|
||||
self.simReqCnt--
|
||||
node.set(false, self.simReqCnt, self.sumWeight)
|
||||
self.update(time)
|
||||
m.update(time)
|
||||
m.simReqCnt--
|
||||
node.set(false, m.simReqCnt, m.sumWeight)
|
||||
m.update(time)
|
||||
}
|
||||
}
|
||||
|
||||
func (self *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (m *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) {
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
|
||||
self.stop(node, time)
|
||||
m.stop(node, time)
|
||||
return uint64(node.rcValue), uint64(node.rcValue - node.startValue)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1162,15 +1162,15 @@ type NodeInfo struct {
|
|||
}
|
||||
|
||||
// NodeInfo retrieves some protocol metadata about the running host node.
|
||||
func (self *ProtocolManager) NodeInfo() *NodeInfo {
|
||||
func (pm *ProtocolManager) NodeInfo() *NodeInfo {
|
||||
head := self.blockchain.CurrentHeader()
|
||||
hash := head.Hash()
|
||||
|
||||
return &NodeInfo{
|
||||
Network: self.networkId,
|
||||
Difficulty: self.blockchain.GetTd(hash, head.Number.Uint64()),
|
||||
Genesis: self.blockchain.Genesis().Hash(),
|
||||
Config: self.blockchain.Config(),
|
||||
Network: pm.networkId,
|
||||
Difficulty: pm.blockchain.GetTd(hash, head.Number.Uint64()),
|
||||
Genesis: pm.blockchain.Genesis().Hash(),
|
||||
Config: pm.blockchain.Config(),
|
||||
Head: hash,
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -50,47 +50,47 @@ func NewLesTxRelay(ps *peerSet, reqDist *requestDistributor) *LesTxRelay {
|
|||
return r
|
||||
}
|
||||
|
||||
func (self *LesTxRelay) registerPeer(p *peer) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (r *LesTxRelay) registerPeer(p *peer) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
|
||||
self.peerList = self.ps.AllPeers()
|
||||
r.peerList = r.ps.AllPeers()
|
||||
}
|
||||
|
||||
func (self *LesTxRelay) unregisterPeer(p *peer) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (r *LesTxRelay) unregisterPeer(p *peer) {
|
||||
r.lock.Lock()
|
||||
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
|
||||
// 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)
|
||||
|
||||
self.peerStartPos++ // rotate the starting position of the peer list
|
||||
if self.peerStartPos >= len(self.peerList) {
|
||||
self.peerStartPos = 0
|
||||
r.peerStartPos++ // rotate the starting position of the peer list
|
||||
if r.peerStartPos >= len(r.peerList) {
|
||||
r.peerStartPos = 0
|
||||
}
|
||||
|
||||
for _, tx := range txs {
|
||||
hash := tx.Hash()
|
||||
ltr, ok := self.txSent[hash]
|
||||
ltr, ok := r.txSent[hash]
|
||||
if !ok {
|
||||
ltr = <rInfo{
|
||||
tx: tx,
|
||||
sentTo: make(map[*peer]struct{}),
|
||||
}
|
||||
self.txSent[hash] = ltr
|
||||
self.txPending[hash] = struct{}{}
|
||||
r.txSent[hash] = ltr
|
||||
r.txPending[hash] = struct{}{}
|
||||
}
|
||||
|
||||
if len(self.peerList) > 0 {
|
||||
if len(r.peerList) > 0 {
|
||||
cnt := count
|
||||
pos := self.peerStartPos
|
||||
pos := r.peerStartPos
|
||||
for {
|
||||
peer := self.peerList[pos]
|
||||
peer := r.peerList[pos]
|
||||
if _, ok := ltr.sentTo[peer]; !ok {
|
||||
sendTo[peer] = append(sendTo[peer], tx)
|
||||
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
|
||||
}
|
||||
pos++
|
||||
if pos == len(self.peerList) {
|
||||
if pos == len(r.peerList) {
|
||||
pos = 0
|
||||
}
|
||||
if pos == self.peerStartPos {
|
||||
if pos == r.peerStartPos {
|
||||
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) }
|
||||
},
|
||||
}
|
||||
self.reqDist.queue(rq)
|
||||
r.reqDist.queue(rq)
|
||||
}
|
||||
}
|
||||
|
||||
func (self *LesTxRelay) Send(txs types.Transactions) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (r *LesTxRelay) Send(txs types.Transactions) {
|
||||
r.lock.Lock()
|
||||
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) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (r *LesTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
|
||||
for _, hash := range mined {
|
||||
delete(self.txPending, hash)
|
||||
delete(r.txPending, hash)
|
||||
}
|
||||
|
||||
for _, hash := range rollback {
|
||||
self.txPending[hash] = struct{}{}
|
||||
r.txPending[hash] = struct{}{}
|
||||
}
|
||||
|
||||
if len(self.txPending) > 0 {
|
||||
txs := make(types.Transactions, len(self.txPending))
|
||||
if len(r.txPending) > 0 {
|
||||
txs := make(types.Transactions, len(r.txPending))
|
||||
i := 0
|
||||
for hash := range self.txPending {
|
||||
txs[i] = self.txSent[hash].tx
|
||||
for hash := range r.txPending {
|
||||
txs[i] = r.txSent[hash].tx
|
||||
i++
|
||||
}
|
||||
self.send(txs, 1)
|
||||
r.send(txs, 1)
|
||||
}
|
||||
}
|
||||
|
||||
func (self *LesTxRelay) Discard(hashes []common.Hash) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (r *LesTxRelay) Discard(hashes []common.Hash) {
|
||||
r.lock.Lock()
|
||||
defer r.lock.Unlock()
|
||||
|
||||
for _, hash := range hashes {
|
||||
delete(self.txSent, hash)
|
||||
delete(self.txPending, hash)
|
||||
delete(r.txSent, hash)
|
||||
delete(r.txPending, hash)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -115,45 +115,45 @@ func NewLightChain(odr OdrBackend, config *params.ChainConfig, engine consensus.
|
|||
}
|
||||
|
||||
// addTrustedCheckpoint adds a trusted checkpoint to the blockchain
|
||||
func (self *LightChain) addTrustedCheckpoint(cp trustedCheckpoint) {
|
||||
if self.odr.ChtIndexer() != nil {
|
||||
StoreChtRoot(self.chainDb, cp.sectionIdx, cp.sectionHead, cp.chtRoot)
|
||||
self.odr.ChtIndexer().AddKnownSectionHead(cp.sectionIdx, cp.sectionHead)
|
||||
func (bc *LightChain) addTrustedCheckpoint(cp trustedCheckpoint) {
|
||||
if bc.odr.ChtIndexer() != nil {
|
||||
StoreChtRoot(bc.chainDb, cp.sectionIdx, cp.sectionHead, cp.chtRoot)
|
||||
bc.odr.ChtIndexer().AddKnownSectionHead(cp.sectionIdx, cp.sectionHead)
|
||||
}
|
||||
if self.odr.BloomTrieIndexer() != nil {
|
||||
StoreBloomTrieRoot(self.chainDb, cp.sectionIdx, cp.sectionHead, cp.bloomTrieRoot)
|
||||
self.odr.BloomTrieIndexer().AddKnownSectionHead(cp.sectionIdx, cp.sectionHead)
|
||||
if bc.odr.BloomTrieIndexer() != nil {
|
||||
StoreBloomTrieRoot(bc.chainDb, cp.sectionIdx, cp.sectionHead, cp.bloomTrieRoot)
|
||||
bc.odr.BloomTrieIndexer().AddKnownSectionHead(cp.sectionIdx, cp.sectionHead)
|
||||
}
|
||||
if self.odr.BloomIndexer() != nil {
|
||||
self.odr.BloomIndexer().AddKnownSectionHead(cp.sectionIdx, cp.sectionHead)
|
||||
if bc.odr.BloomIndexer() != nil {
|
||||
bc.odr.BloomIndexer().AddKnownSectionHead(cp.sectionIdx, cp.sectionHead)
|
||||
}
|
||||
log.Info("Added trusted checkpoint", "chain", cp.name, "block", (cp.sectionIdx+1)*CHTFrequencyClient-1, "hash", cp.sectionHead)
|
||||
}
|
||||
|
||||
func (self *LightChain) getProcInterrupt() bool {
|
||||
return atomic.LoadInt32(&self.procInterrupt) == 1
|
||||
func (bc *LightChain) getProcInterrupt() bool {
|
||||
return atomic.LoadInt32(&bc.procInterrupt) == 1
|
||||
}
|
||||
|
||||
// Odr returns the ODR backend of the chain
|
||||
func (self *LightChain) Odr() OdrBackend {
|
||||
return self.odr
|
||||
func (bc *LightChain) Odr() OdrBackend {
|
||||
return bc.odr
|
||||
}
|
||||
|
||||
// loadLastState loads the last known chain state from the database. This method
|
||||
// assumes that the chain manager mutex is held.
|
||||
func (self *LightChain) loadLastState() error {
|
||||
if head := core.GetHeadHeaderHash(self.chainDb); head == (common.Hash{}) {
|
||||
func (bc *LightChain) loadLastState() error {
|
||||
if head := core.GetHeadHeaderHash(bc.chainDb); head == (common.Hash{}) {
|
||||
// Corrupt or empty database, init from scratch
|
||||
self.Reset()
|
||||
bc.Reset()
|
||||
} else {
|
||||
if header := self.GetHeaderByHash(head); header != nil {
|
||||
self.hc.SetCurrentHeader(header)
|
||||
if header := bc.GetHeaderByHash(head); header != nil {
|
||||
bc.hc.SetCurrentHeader(header)
|
||||
}
|
||||
}
|
||||
|
||||
// Issue a status log and return
|
||||
header := self.hc.CurrentHeader()
|
||||
headerTd := self.GetTd(header.Hash(), header.Number.Uint64())
|
||||
header := bc.hc.CurrentHeader()
|
||||
headerTd := bc.GetTd(header.Hash(), header.Number.Uint64())
|
||||
log.Info("Loaded most recent local header", "number", header.Number, "hash", header.Hash(), "td", headerTd)
|
||||
|
||||
return nil
|
||||
|
|
@ -170,8 +170,8 @@ func (bc *LightChain) SetHead(head uint64) {
|
|||
}
|
||||
|
||||
// GasLimit returns the gas limit of the current HEAD block.
|
||||
func (self *LightChain) GasLimit() uint64 {
|
||||
return self.hc.CurrentHeader().GasLimit
|
||||
func (bc *LightChain) GasLimit() uint64 {
|
||||
return bc.hc.CurrentHeader().GasLimit
|
||||
}
|
||||
|
||||
// Reset purges the entire blockchain, restoring it to its genesis state.
|
||||
|
|
@ -217,34 +217,34 @@ func (bc *LightChain) State() (*state.StateDB, error) {
|
|||
|
||||
// GetBody retrieves a block body (transactions and uncles) from the database
|
||||
// or ODR service by hash, caching it if found.
|
||||
func (self *LightChain) GetBody(ctx context.Context, hash common.Hash) (*types.Body, error) {
|
||||
func (bc *LightChain) GetBody(ctx context.Context, hash common.Hash) (*types.Body, error) {
|
||||
// Short circuit if the body's already in the cache, retrieve otherwise
|
||||
if cached, ok := self.bodyCache.Get(hash); ok {
|
||||
if cached, ok := bc.bodyCache.Get(hash); ok {
|
||||
body := cached.(*types.Body)
|
||||
return body, nil
|
||||
}
|
||||
body, err := GetBody(ctx, self.odr, hash, self.hc.GetBlockNumber(hash))
|
||||
body, err := GetBody(ctx, bc.odr, hash, bc.hc.GetBlockNumber(hash))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Cache the found body for next time and return
|
||||
self.bodyCache.Add(hash, body)
|
||||
bc.bodyCache.Add(hash, body)
|
||||
return body, nil
|
||||
}
|
||||
|
||||
// GetBodyRLP retrieves a block body in RLP encoding from the database or
|
||||
// ODR service by hash, caching it if found.
|
||||
func (self *LightChain) GetBodyRLP(ctx context.Context, hash common.Hash) (rlp.RawValue, error) {
|
||||
func (bc *LightChain) GetBodyRLP(ctx context.Context, hash common.Hash) (rlp.RawValue, error) {
|
||||
// Short circuit if the body's already in the cache, retrieve otherwise
|
||||
if cached, ok := self.bodyRLPCache.Get(hash); ok {
|
||||
if cached, ok := bc.bodyRLPCache.Get(hash); ok {
|
||||
return cached.(rlp.RawValue), nil
|
||||
}
|
||||
body, err := GetBodyRLP(ctx, self.odr, hash, self.hc.GetBlockNumber(hash))
|
||||
body, err := GetBodyRLP(ctx, bc.odr, hash, bc.hc.GetBlockNumber(hash))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Cache the found body for next time and return
|
||||
self.bodyRLPCache.Add(hash, body)
|
||||
bc.bodyRLPCache.Add(hash, body)
|
||||
return body, nil
|
||||
}
|
||||
|
||||
|
|
@ -257,34 +257,34 @@ func (bc *LightChain) HasBlock(hash common.Hash, number uint64) bool {
|
|||
|
||||
// GetBlock retrieves a block from the database or ODR service by hash and number,
|
||||
// caching it if found.
|
||||
func (self *LightChain) GetBlock(ctx context.Context, hash common.Hash, number uint64) (*types.Block, error) {
|
||||
func (bc *LightChain) GetBlock(ctx context.Context, hash common.Hash, number uint64) (*types.Block, error) {
|
||||
// Short circuit if the block's already in the cache, retrieve otherwise
|
||||
if block, ok := self.blockCache.Get(hash); ok {
|
||||
if block, ok := bc.blockCache.Get(hash); ok {
|
||||
return block.(*types.Block), nil
|
||||
}
|
||||
block, err := GetBlock(ctx, self.odr, hash, number)
|
||||
block, err := GetBlock(ctx, bc.odr, hash, number)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Cache the found block for next time and return
|
||||
self.blockCache.Add(block.Hash(), block)
|
||||
bc.blockCache.Add(block.Hash(), block)
|
||||
return block, nil
|
||||
}
|
||||
|
||||
// GetBlockByHash retrieves a block from the database or ODR service by hash,
|
||||
// caching it if found.
|
||||
func (self *LightChain) GetBlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
return self.GetBlock(ctx, hash, self.hc.GetBlockNumber(hash))
|
||||
func (bc *LightChain) GetBlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error) {
|
||||
return bc.GetBlock(ctx, hash, bc.hc.GetBlockNumber(hash))
|
||||
}
|
||||
|
||||
// GetBlockByNumber retrieves a block from the database or ODR service by
|
||||
// number, caching it (associated with its hash) if found.
|
||||
func (self *LightChain) GetBlockByNumber(ctx context.Context, number uint64) (*types.Block, error) {
|
||||
hash, err := GetCanonicalHash(ctx, self.odr, number)
|
||||
func (bc *LightChain) GetBlockByNumber(ctx context.Context, number uint64) (*types.Block, error) {
|
||||
hash, err := GetCanonicalHash(ctx, bc.odr, number)
|
||||
if hash == (common.Hash{}) || err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return self.GetBlock(ctx, hash, number)
|
||||
return bc.GetBlock(ctx, hash, number)
|
||||
}
|
||||
|
||||
// Stop stops the blockchain service. If any imports are currently in progress
|
||||
|
|
@ -302,31 +302,31 @@ func (bc *LightChain) Stop() {
|
|||
|
||||
// Rollback is designed to remove a chain of links from the database that aren't
|
||||
// certain enough to be valid.
|
||||
func (self *LightChain) Rollback(chain []common.Hash) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
func (bc *LightChain) Rollback(chain []common.Hash) {
|
||||
bc.mu.Lock()
|
||||
defer bc.mu.Unlock()
|
||||
|
||||
for i := len(chain) - 1; i >= 0; i-- {
|
||||
hash := chain[i]
|
||||
|
||||
if head := self.hc.CurrentHeader(); head.Hash() == hash {
|
||||
self.hc.SetCurrentHeader(self.GetHeader(head.ParentHash, head.Number.Uint64()-1))
|
||||
if head := bc.hc.CurrentHeader(); head.Hash() == hash {
|
||||
bc.hc.SetCurrentHeader(bc.GetHeader(head.ParentHash, head.Number.Uint64()-1))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// postChainEvents iterates over the events generated by a chain insertion and
|
||||
// posts them into the event feed.
|
||||
func (self *LightChain) postChainEvents(events []interface{}) {
|
||||
func (bc *LightChain) postChainEvents(events []interface{}) {
|
||||
for _, event := range events {
|
||||
switch ev := event.(type) {
|
||||
case core.ChainEvent:
|
||||
if self.CurrentHeader().Hash() == ev.Hash {
|
||||
self.chainHeadFeed.Send(core.ChainHeadEvent{Block: ev.Block})
|
||||
if bc.CurrentHeader().Hash() == ev.Hash {
|
||||
bc.chainHeadFeed.Send(core.ChainHeadEvent{Block: ev.Block})
|
||||
}
|
||||
self.chainFeed.Send(ev)
|
||||
bc.chainFeed.Send(ev)
|
||||
case core.ChainSideEvent:
|
||||
self.chainSideFeed.Send(ev)
|
||||
bc.chainSideFeed.Send(ev)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -342,28 +342,28 @@ func (self *LightChain) postChainEvents(events []interface{}) {
|
|||
//
|
||||
// In the case of a light chain, InsertHeaderChain also creates and posts light
|
||||
// chain events when necessary.
|
||||
func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) (int, error) {
|
||||
func (bc *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int) (int, error) {
|
||||
start := time.Now()
|
||||
if i, err := self.hc.ValidateHeaderChain(chain, checkFreq); err != nil {
|
||||
if i, err := bc.hc.ValidateHeaderChain(chain, checkFreq); err != nil {
|
||||
return i, err
|
||||
}
|
||||
|
||||
// Make sure only one thread manipulates the chain at once
|
||||
self.chainmu.Lock()
|
||||
bc.chainmu.Lock()
|
||||
defer func() {
|
||||
self.chainmu.Unlock()
|
||||
bc.chainmu.Unlock()
|
||||
time.Sleep(time.Millisecond * 10) // ugly hack; do not hog chain lock in case syncing is CPU-limited by validation
|
||||
}()
|
||||
|
||||
self.wg.Add(1)
|
||||
defer self.wg.Done()
|
||||
bc.wg.Add(1)
|
||||
defer bc.wg.Done()
|
||||
|
||||
var events []interface{}
|
||||
whFunc := func(header *types.Header) error {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
bc.mu.Lock()
|
||||
defer bc.mu.Unlock()
|
||||
|
||||
status, err := self.hc.WriteHeader(header)
|
||||
status, err := bc.hc.WriteHeader(header)
|
||||
|
||||
switch status {
|
||||
case core.CanonStatTy:
|
||||
|
|
@ -376,39 +376,39 @@ func (self *LightChain) InsertHeaderChain(chain []*types.Header, checkFreq int)
|
|||
}
|
||||
return err
|
||||
}
|
||||
i, err := self.hc.InsertHeaderChain(chain, whFunc, start)
|
||||
self.postChainEvents(events)
|
||||
i, err := bc.hc.InsertHeaderChain(chain, whFunc, start)
|
||||
bc.postChainEvents(events)
|
||||
return i, err
|
||||
}
|
||||
|
||||
// CurrentHeader retrieves the current head header of the canonical chain. The
|
||||
// header is retrieved from the HeaderChain's internal cache.
|
||||
func (self *LightChain) CurrentHeader() *types.Header {
|
||||
return self.hc.CurrentHeader()
|
||||
func (bc *LightChain) CurrentHeader() *types.Header {
|
||||
return bc.hc.CurrentHeader()
|
||||
}
|
||||
|
||||
// GetTd retrieves a block's total difficulty in the canonical chain from the
|
||||
// database by hash and number, caching it if found.
|
||||
func (self *LightChain) GetTd(hash common.Hash, number uint64) *big.Int {
|
||||
return self.hc.GetTd(hash, number)
|
||||
func (bc *LightChain) GetTd(hash common.Hash, number uint64) *big.Int {
|
||||
return bc.hc.GetTd(hash, number)
|
||||
}
|
||||
|
||||
// GetTdByHash retrieves a block's total difficulty in the canonical chain from the
|
||||
// database by hash, caching it if found.
|
||||
func (self *LightChain) GetTdByHash(hash common.Hash) *big.Int {
|
||||
return self.hc.GetTdByHash(hash)
|
||||
func (bc *LightChain) GetTdByHash(hash common.Hash) *big.Int {
|
||||
return bc.hc.GetTdByHash(hash)
|
||||
}
|
||||
|
||||
// GetHeader retrieves a block header from the database by hash and number,
|
||||
// caching it if found.
|
||||
func (self *LightChain) GetHeader(hash common.Hash, number uint64) *types.Header {
|
||||
return self.hc.GetHeader(hash, number)
|
||||
func (bc *LightChain) GetHeader(hash common.Hash, number uint64) *types.Header {
|
||||
return bc.hc.GetHeader(hash, number)
|
||||
}
|
||||
|
||||
// GetHeaderByHash retrieves a block header from the database by hash, caching it if
|
||||
// found.
|
||||
func (self *LightChain) GetHeaderByHash(hash common.Hash) *types.Header {
|
||||
return self.hc.GetHeaderByHash(hash)
|
||||
func (bc *LightChain) GetHeaderByHash(hash common.Hash) *types.Header {
|
||||
return bc.hc.GetHeaderByHash(hash)
|
||||
}
|
||||
|
||||
// HasHeader checks if a block header is present in the database or not, caching
|
||||
|
|
@ -419,43 +419,43 @@ func (bc *LightChain) HasHeader(hash common.Hash, number uint64) bool {
|
|||
|
||||
// GetBlockHashesFromHash retrieves a number of block hashes starting at a given
|
||||
// hash, fetching towards the genesis block.
|
||||
func (self *LightChain) GetBlockHashesFromHash(hash common.Hash, max uint64) []common.Hash {
|
||||
return self.hc.GetBlockHashesFromHash(hash, max)
|
||||
func (bc *LightChain) GetBlockHashesFromHash(hash common.Hash, max uint64) []common.Hash {
|
||||
return bc.hc.GetBlockHashesFromHash(hash, max)
|
||||
}
|
||||
|
||||
// GetHeaderByNumber retrieves a block header from the database by number,
|
||||
// caching it (associated with its hash) if found.
|
||||
func (self *LightChain) GetHeaderByNumber(number uint64) *types.Header {
|
||||
return self.hc.GetHeaderByNumber(number)
|
||||
func (bc *LightChain) GetHeaderByNumber(number uint64) *types.Header {
|
||||
return bc.hc.GetHeaderByNumber(number)
|
||||
}
|
||||
|
||||
// GetHeaderByNumberOdr retrieves a block header from the database or network
|
||||
// by number, caching it (associated with its hash) if found.
|
||||
func (self *LightChain) GetHeaderByNumberOdr(ctx context.Context, number uint64) (*types.Header, error) {
|
||||
if header := self.hc.GetHeaderByNumber(number); header != nil {
|
||||
func (bc *LightChain) GetHeaderByNumberOdr(ctx context.Context, number uint64) (*types.Header, error) {
|
||||
if header := bc.hc.GetHeaderByNumber(number); header != nil {
|
||||
return header, nil
|
||||
}
|
||||
return GetHeaderByNumber(ctx, self.odr, number)
|
||||
return GetHeaderByNumber(ctx, bc.odr, number)
|
||||
}
|
||||
|
||||
// Config retrieves the header chain's chain configuration.
|
||||
func (self *LightChain) Config() *params.ChainConfig { return self.hc.Config() }
|
||||
func (bc *LightChain) Config() *params.ChainConfig { return bc.hc.Config() }
|
||||
|
||||
func (self *LightChain) SyncCht(ctx context.Context) bool {
|
||||
if self.odr.ChtIndexer() == nil {
|
||||
func (bc *LightChain) SyncCht(ctx context.Context) bool {
|
||||
if bc.odr.ChtIndexer() == nil {
|
||||
return false
|
||||
}
|
||||
headNum := self.CurrentHeader().Number.Uint64()
|
||||
chtCount, _, _ := self.odr.ChtIndexer().Sections()
|
||||
headNum := bc.CurrentHeader().Number.Uint64()
|
||||
chtCount, _, _ := bc.odr.ChtIndexer().Sections()
|
||||
if headNum+1 < chtCount*CHTFrequencyClient {
|
||||
num := chtCount*CHTFrequencyClient - 1
|
||||
header, err := GetHeaderByNumber(ctx, self.odr, num)
|
||||
header, err := GetHeaderByNumber(ctx, bc.odr, num)
|
||||
if header != nil && err == nil {
|
||||
self.mu.Lock()
|
||||
if self.hc.CurrentHeader().Number.Uint64() < header.Number.Uint64() {
|
||||
self.hc.SetCurrentHeader(header)
|
||||
bc.mu.Lock()
|
||||
if bc.hc.CurrentHeader().Number.Uint64() < header.Number.Uint64() {
|
||||
bc.hc.SetCurrentHeader(header)
|
||||
}
|
||||
self.mu.Unlock()
|
||||
bc.mu.Unlock()
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
@ -464,38 +464,38 @@ func (self *LightChain) SyncCht(ctx context.Context) bool {
|
|||
|
||||
// LockChain locks the chain mutex for reading so that multiple canonical hashes can be
|
||||
// retrieved while it is guaranteed that they belong to the same version of the chain
|
||||
func (self *LightChain) LockChain() {
|
||||
self.chainmu.RLock()
|
||||
func (bc *LightChain) LockChain() {
|
||||
bc.chainmu.RLock()
|
||||
}
|
||||
|
||||
// UnlockChain unlocks the chain mutex
|
||||
func (self *LightChain) UnlockChain() {
|
||||
self.chainmu.RUnlock()
|
||||
func (bc *LightChain) UnlockChain() {
|
||||
bc.chainmu.RUnlock()
|
||||
}
|
||||
|
||||
// SubscribeChainEvent registers a subscription of ChainEvent.
|
||||
func (self *LightChain) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
|
||||
return self.scope.Track(self.chainFeed.Subscribe(ch))
|
||||
func (bc *LightChain) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
|
||||
return bc.scope.Track(bc.chainFeed.Subscribe(ch))
|
||||
}
|
||||
|
||||
// SubscribeChainHeadEvent registers a subscription of ChainHeadEvent.
|
||||
func (self *LightChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
|
||||
return self.scope.Track(self.chainHeadFeed.Subscribe(ch))
|
||||
func (bc *LightChain) SubscribeChainHeadEvent(ch chan<- core.ChainHeadEvent) event.Subscription {
|
||||
return bc.scope.Track(bc.chainHeadFeed.Subscribe(ch))
|
||||
}
|
||||
|
||||
// SubscribeChainSideEvent registers a subscription of ChainSideEvent.
|
||||
func (self *LightChain) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription {
|
||||
return self.scope.Track(self.chainSideFeed.Subscribe(ch))
|
||||
func (bc *LightChain) SubscribeChainSideEvent(ch chan<- core.ChainSideEvent) event.Subscription {
|
||||
return bc.scope.Track(bc.chainSideFeed.Subscribe(ch))
|
||||
}
|
||||
|
||||
// SubscribeLogsEvent implements the interface of filters.Backend
|
||||
// LightChain does not send logs events, so return an empty subscription.
|
||||
func (self *LightChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
|
||||
return self.scope.Track(new(event.Feed).Subscribe(ch))
|
||||
func (bc *LightChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
|
||||
return bc.scope.Track(new(event.Feed).Subscribe(ch))
|
||||
}
|
||||
|
||||
// SubscribeRemovedLogsEvent implements the interface of filters.Backend
|
||||
// LightChain does not send core.RemovedLogsEvent, so return an empty subscription.
|
||||
func (self *LightChain) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
|
||||
return self.scope.Track(new(event.Feed).Subscribe(ch))
|
||||
func (bc *LightChain) SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription {
|
||||
return bc.scope.Track(new(event.Feed).Subscribe(ch))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -388,81 +388,81 @@ func (pool *TxPool) validateTx(ctx context.Context, tx *types.Transaction) error
|
|||
|
||||
// add validates a new transaction and sets its state pending if processable.
|
||||
// It also updates the locally stored nonce if necessary.
|
||||
func (self *TxPool) add(ctx context.Context, tx *types.Transaction) error {
|
||||
func (pool *TxPool) add(ctx context.Context, tx *types.Transaction) error {
|
||||
hash := tx.Hash()
|
||||
|
||||
if self.pending[hash] != nil {
|
||||
if pool.pending[hash] != nil {
|
||||
return fmt.Errorf("Known transaction (%x)", hash[:4])
|
||||
}
|
||||
err := self.validateTx(ctx, tx)
|
||||
err := pool.validateTx(ctx, tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, ok := self.pending[hash]; !ok {
|
||||
self.pending[hash] = tx
|
||||
if _, ok := pool.pending[hash]; !ok {
|
||||
pool.pending[hash] = tx
|
||||
|
||||
nonce := tx.Nonce() + 1
|
||||
|
||||
addr, _ := types.Sender(self.signer, tx)
|
||||
if nonce > self.nonce[addr] {
|
||||
self.nonce[addr] = nonce
|
||||
addr, _ := types.Sender(pool.signer, tx)
|
||||
if nonce > pool.nonce[addr] {
|
||||
pool.nonce[addr] = nonce
|
||||
}
|
||||
|
||||
// Notify the subscribers. This event is posted in a goroutine
|
||||
// because it's possible that somewhere during the post "Remove transaction"
|
||||
// gets called which will then wait for the global tx pool lock and deadlock.
|
||||
go self.txFeed.Send(core.TxPreEvent{Tx: tx})
|
||||
go pool.txFeed.Send(core.TxPreEvent{Tx: tx})
|
||||
}
|
||||
|
||||
// Print a log message if low enough level is set
|
||||
log.Debug("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(self.signer, tx); return from }}, "to", tx.To())
|
||||
log.Debug("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(pool.signer, tx); return from }}, "to", tx.To())
|
||||
return nil
|
||||
}
|
||||
|
||||
// Add adds a transaction to the pool if valid and passes it to the tx relay
|
||||
// backend
|
||||
func (self *TxPool) Add(ctx context.Context, tx *types.Transaction) error {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
func (pool *TxPool) Add(ctx context.Context, tx *types.Transaction) error {
|
||||
pool.mu.Lock()
|
||||
defer pool.mu.Unlock()
|
||||
|
||||
data, err := rlp.EncodeToBytes(tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := self.add(ctx, tx); err != nil {
|
||||
if err := pool.add(ctx, tx); err != nil {
|
||||
return err
|
||||
}
|
||||
//fmt.Println("Send", tx.Hash())
|
||||
self.relay.Send(types.Transactions{tx})
|
||||
pool.relay.Send(types.Transactions{tx})
|
||||
|
||||
self.chainDb.Put(tx.Hash().Bytes(), data)
|
||||
pool.chainDb.Put(tx.Hash().Bytes(), data)
|
||||
return nil
|
||||
}
|
||||
|
||||
// AddTransactions adds all valid transactions to the pool and passes them to
|
||||
// AddBatch adds all valid transactions to the pool and passes them to
|
||||
// the tx relay backend
|
||||
func (self *TxPool) AddBatch(ctx context.Context, txs []*types.Transaction) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
func (pool *TxPool) AddBatch(ctx context.Context, txs []*types.Transaction) {
|
||||
pool.mu.Lock()
|
||||
defer pool.mu.Unlock()
|
||||
var sendTx types.Transactions
|
||||
|
||||
for _, tx := range txs {
|
||||
if err := self.add(ctx, tx); err == nil {
|
||||
if err := pool.add(ctx, tx); err == nil {
|
||||
sendTx = append(sendTx, tx)
|
||||
}
|
||||
}
|
||||
if len(sendTx) > 0 {
|
||||
self.relay.Send(sendTx)
|
||||
pool.relay.Send(sendTx)
|
||||
}
|
||||
}
|
||||
|
||||
// GetTransaction returns a transaction if it is contained in the pool
|
||||
// and nil otherwise.
|
||||
func (tp *TxPool) GetTransaction(hash common.Hash) *types.Transaction {
|
||||
func (pool *TxPool) GetTransaction(hash common.Hash) *types.Transaction {
|
||||
// check the txs first
|
||||
if tx, ok := tp.pending[hash]; ok {
|
||||
if tx, ok := pool.pending[hash]; ok {
|
||||
return tx
|
||||
}
|
||||
return nil
|
||||
|
|
@ -470,13 +470,13 @@ func (tp *TxPool) GetTransaction(hash common.Hash) *types.Transaction {
|
|||
|
||||
// GetTransactions returns all currently processable transactions.
|
||||
// The returned slice may be modified by the caller.
|
||||
func (self *TxPool) GetTransactions() (txs types.Transactions, err error) {
|
||||
self.mu.RLock()
|
||||
defer self.mu.RUnlock()
|
||||
func (pool *TxPool) GetTransactions() (txs types.Transactions, err error) {
|
||||
pool.mu.RLock()
|
||||
defer pool.mu.RUnlock()
|
||||
|
||||
txs = make(types.Transactions, len(self.pending))
|
||||
txs = make(types.Transactions, len(pool.pending))
|
||||
i := 0
|
||||
for _, tx := range self.pending {
|
||||
for _, tx := range pool.pending {
|
||||
txs[i] = tx
|
||||
i++
|
||||
}
|
||||
|
|
@ -485,14 +485,14 @@ func (self *TxPool) GetTransactions() (txs types.Transactions, err error) {
|
|||
|
||||
// Content retrieves the data content of the transaction pool, returning all the
|
||||
// pending as well as queued transactions, grouped by account and nonce.
|
||||
func (self *TxPool) Content() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {
|
||||
self.mu.RLock()
|
||||
defer self.mu.RUnlock()
|
||||
func (pool *TxPool) Content() (map[common.Address]types.Transactions, map[common.Address]types.Transactions) {
|
||||
pool.mu.RLock()
|
||||
defer pool.mu.RUnlock()
|
||||
|
||||
// Retrieve all the pending transactions and sort by account and by nonce
|
||||
pending := make(map[common.Address]types.Transactions)
|
||||
for _, tx := range self.pending {
|
||||
account, _ := types.Sender(self.signer, tx)
|
||||
for _, tx := range pool.pending {
|
||||
account, _ := types.Sender(pool.signer, tx)
|
||||
pending[account] = append(pending[account], tx)
|
||||
}
|
||||
// There are no queued transactions in a light pool, just return an empty map
|
||||
|
|
@ -501,18 +501,18 @@ func (self *TxPool) Content() (map[common.Address]types.Transactions, map[common
|
|||
}
|
||||
|
||||
// RemoveTransactions removes all given transactions from the pool.
|
||||
func (self *TxPool) RemoveTransactions(txs types.Transactions) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
func (pool *TxPool) RemoveTransactions(txs types.Transactions) {
|
||||
pool.mu.Lock()
|
||||
defer pool.mu.Unlock()
|
||||
var hashes []common.Hash
|
||||
for _, tx := range txs {
|
||||
//self.RemoveTx(tx.Hash())
|
||||
//pool.RemoveTx(tx.Hash())
|
||||
hash := tx.Hash()
|
||||
delete(self.pending, hash)
|
||||
self.chainDb.Delete(hash[:])
|
||||
delete(pool.pending, hash)
|
||||
pool.chainDb.Delete(hash[:])
|
||||
hashes = append(hashes, hash)
|
||||
}
|
||||
self.relay.Discard(hashes)
|
||||
pool.relay.Discard(hashes)
|
||||
}
|
||||
|
||||
// RemoveTx removes the transaction with the given hash from the pool.
|
||||
|
|
|
|||
|
|
@ -36,19 +36,19 @@ type testTxRelay struct {
|
|||
send, discard, mined chan int
|
||||
}
|
||||
|
||||
func (self *testTxRelay) Send(txs types.Transactions) {
|
||||
self.send <- len(txs)
|
||||
func (r *testTxRelay) Send(txs types.Transactions) {
|
||||
r.send <- len(txs)
|
||||
}
|
||||
|
||||
func (self *testTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) {
|
||||
func (r *testTxRelay) NewHead(head common.Hash, mined []common.Hash, rollback []common.Hash) {
|
||||
m := len(mined)
|
||||
if m != 0 {
|
||||
self.mined <- m
|
||||
r.mined <- m
|
||||
}
|
||||
}
|
||||
|
||||
func (self *testTxRelay) Discard(hashes []common.Hash) {
|
||||
self.discard <- len(hashes)
|
||||
func (r *testTxRelay) Discard(hashes []common.Hash) {
|
||||
r.discard <- len(hashes)
|
||||
}
|
||||
|
||||
const poolTestTxs = 1000
|
||||
|
|
|
|||
|
|
@ -49,70 +49,70 @@ func NewCpuAgent(chain consensus.ChainReader, engine consensus.Engine) *CpuAgent
|
|||
return miner
|
||||
}
|
||||
|
||||
func (self *CpuAgent) Work() chan<- *Work { return self.workCh }
|
||||
func (self *CpuAgent) SetReturnCh(ch chan<- *Result) { self.returnCh = ch }
|
||||
func (a *CpuAgent) Work() chan<- *Work { return a.workCh }
|
||||
func (a *CpuAgent) SetReturnCh(ch chan<- *Result) { a.returnCh = ch }
|
||||
|
||||
func (self *CpuAgent) Stop() {
|
||||
if !atomic.CompareAndSwapInt32(&self.isMining, 1, 0) {
|
||||
func (a *CpuAgent) Stop() {
|
||||
if !atomic.CompareAndSwapInt32(&a.isMining, 1, 0) {
|
||||
return // agent already stopped
|
||||
}
|
||||
self.stop <- struct{}{}
|
||||
a.stop <- struct{}{}
|
||||
done:
|
||||
// Empty work channel
|
||||
for {
|
||||
select {
|
||||
case <-self.workCh:
|
||||
case <-a.workCh:
|
||||
default:
|
||||
break done
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *CpuAgent) Start() {
|
||||
if !atomic.CompareAndSwapInt32(&self.isMining, 0, 1) {
|
||||
func (a *CpuAgent) Start() {
|
||||
if !atomic.CompareAndSwapInt32(&a.isMining, 0, 1) {
|
||||
return // agent already started
|
||||
}
|
||||
go self.update()
|
||||
go a.update()
|
||||
}
|
||||
|
||||
func (self *CpuAgent) update() {
|
||||
func (a *CpuAgent) update() {
|
||||
out:
|
||||
for {
|
||||
select {
|
||||
case work := <-self.workCh:
|
||||
self.mu.Lock()
|
||||
if self.quitCurrentOp != nil {
|
||||
close(self.quitCurrentOp)
|
||||
case work := <-a.workCh:
|
||||
a.mu.Lock()
|
||||
if a.quitCurrentOp != nil {
|
||||
close(a.quitCurrentOp)
|
||||
}
|
||||
self.quitCurrentOp = make(chan struct{})
|
||||
go self.mine(work, self.quitCurrentOp)
|
||||
self.mu.Unlock()
|
||||
case <-self.stop:
|
||||
self.mu.Lock()
|
||||
if self.quitCurrentOp != nil {
|
||||
close(self.quitCurrentOp)
|
||||
self.quitCurrentOp = nil
|
||||
a.quitCurrentOp = make(chan struct{})
|
||||
go a.mine(work, a.quitCurrentOp)
|
||||
a.mu.Unlock()
|
||||
case <-a.stop:
|
||||
a.mu.Lock()
|
||||
if a.quitCurrentOp != nil {
|
||||
close(a.quitCurrentOp)
|
||||
a.quitCurrentOp = nil
|
||||
}
|
||||
self.mu.Unlock()
|
||||
a.mu.Unlock()
|
||||
break out
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *CpuAgent) mine(work *Work, stop <-chan struct{}) {
|
||||
if result, err := self.engine.Seal(self.chain, work.Block, stop); result != nil {
|
||||
func (a *CpuAgent) mine(work *Work, stop <-chan struct{}) {
|
||||
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())
|
||||
self.returnCh <- &Result{work, result}
|
||||
a.returnCh <- &Result{work, result}
|
||||
} else {
|
||||
if err != nil {
|
||||
log.Warn("Block sealing failed", "err", err)
|
||||
}
|
||||
self.returnCh <- nil
|
||||
a.returnCh <- nil
|
||||
}
|
||||
}
|
||||
|
||||
func (self *CpuAgent) GetHashRate() int64 {
|
||||
if pow, ok := self.engine.(consensus.PoW); ok {
|
||||
func (a *CpuAgent) GetHashRate() int64 {
|
||||
if pow, ok := a.engine.(consensus.PoW); ok {
|
||||
return int64(pow.Hashrate())
|
||||
}
|
||||
return 0
|
||||
|
|
|
|||
|
|
@ -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
|
||||
// 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.
|
||||
func (self *Miner) update() {
|
||||
events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
|
||||
func (m *Miner) update() {
|
||||
events := m.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
|
||||
out:
|
||||
for ev := range events.Chan() {
|
||||
switch ev.Data.(type) {
|
||||
case downloader.StartEvent:
|
||||
atomic.StoreInt32(&self.canStart, 0)
|
||||
if self.Mining() {
|
||||
self.Stop()
|
||||
atomic.StoreInt32(&self.shouldStart, 1)
|
||||
atomic.StoreInt32(&m.canStart, 0)
|
||||
if m.Mining() {
|
||||
m.Stop()
|
||||
atomic.StoreInt32(&m.shouldStart, 1)
|
||||
log.Info("Mining aborted due to sync")
|
||||
}
|
||||
case downloader.DoneEvent, downloader.FailedEvent:
|
||||
shouldStart := atomic.LoadInt32(&self.shouldStart) == 1
|
||||
shouldStart := atomic.LoadInt32(&m.shouldStart) == 1
|
||||
|
||||
atomic.StoreInt32(&self.canStart, 1)
|
||||
atomic.StoreInt32(&self.shouldStart, 0)
|
||||
atomic.StoreInt32(&m.canStart, 1)
|
||||
atomic.StoreInt32(&m.shouldStart, 0)
|
||||
if shouldStart {
|
||||
self.Start(self.coinbase)
|
||||
m.Start(m.coinbase)
|
||||
}
|
||||
// unsubscribe. we're only interested in this event once
|
||||
events.Unsubscribe()
|
||||
|
|
@ -103,50 +103,50 @@ out:
|
|||
}
|
||||
}
|
||||
|
||||
func (self *Miner) Start(coinbase common.Address) {
|
||||
atomic.StoreInt32(&self.shouldStart, 1)
|
||||
self.SetEtherbase(coinbase)
|
||||
func (m *Miner) Start(coinbase common.Address) {
|
||||
atomic.StoreInt32(&m.shouldStart, 1)
|
||||
m.SetEtherbase(coinbase)
|
||||
|
||||
if atomic.LoadInt32(&self.canStart) == 0 {
|
||||
if atomic.LoadInt32(&m.canStart) == 0 {
|
||||
log.Info("Network syncing, will start miner afterwards")
|
||||
return
|
||||
}
|
||||
atomic.StoreInt32(&self.mining, 1)
|
||||
atomic.StoreInt32(&m.mining, 1)
|
||||
|
||||
log.Info("Starting mining operation")
|
||||
self.worker.start()
|
||||
self.worker.commitNewWork()
|
||||
m.worker.start()
|
||||
m.worker.commitNewWork()
|
||||
}
|
||||
|
||||
func (self *Miner) Stop() {
|
||||
self.worker.stop()
|
||||
atomic.StoreInt32(&self.mining, 0)
|
||||
atomic.StoreInt32(&self.shouldStart, 0)
|
||||
func (m *Miner) Stop() {
|
||||
m.worker.stop()
|
||||
atomic.StoreInt32(&m.mining, 0)
|
||||
atomic.StoreInt32(&m.shouldStart, 0)
|
||||
}
|
||||
|
||||
func (self *Miner) Register(agent Agent) {
|
||||
if self.Mining() {
|
||||
func (m *Miner) Register(agent Agent) {
|
||||
if m.Mining() {
|
||||
agent.Start()
|
||||
}
|
||||
self.worker.register(agent)
|
||||
m.worker.register(agent)
|
||||
}
|
||||
|
||||
func (self *Miner) Unregister(agent Agent) {
|
||||
self.worker.unregister(agent)
|
||||
func (m *Miner) Unregister(agent Agent) {
|
||||
m.worker.unregister(agent)
|
||||
}
|
||||
|
||||
func (self *Miner) Mining() bool {
|
||||
return atomic.LoadInt32(&self.mining) > 0
|
||||
func (m *Miner) Mining() bool {
|
||||
return atomic.LoadInt32(&m.mining) > 0
|
||||
}
|
||||
|
||||
func (self *Miner) HashRate() (tot int64) {
|
||||
if pow, ok := self.engine.(consensus.PoW); ok {
|
||||
func (m *Miner) HashRate() (tot int64) {
|
||||
if pow, ok := m.engine.(consensus.PoW); ok {
|
||||
tot += int64(pow.Hashrate())
|
||||
}
|
||||
// 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
|
||||
// hashrate?
|
||||
for agent := range self.worker.agents {
|
||||
for agent := range m.worker.agents {
|
||||
if _, ok := agent.(*CpuAgent); !ok {
|
||||
tot += agent.GetHashRate()
|
||||
}
|
||||
|
|
@ -154,17 +154,17 @@ func (self *Miner) HashRate() (tot int64) {
|
|||
return
|
||||
}
|
||||
|
||||
func (self *Miner) SetExtra(extra []byte) error {
|
||||
func (m *Miner) SetExtra(extra []byte) error {
|
||||
if uint64(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
|
||||
}
|
||||
|
||||
// Pending returns the currently pending block and associated state.
|
||||
func (self *Miner) Pending() (*types.Block, *state.StateDB) {
|
||||
return self.worker.pending()
|
||||
func (m *Miner) Pending() (*types.Block, *state.StateDB) {
|
||||
return m.worker.pending()
|
||||
}
|
||||
|
||||
// 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
|
||||
// simultaneously, please use Pending(), as the pending state can
|
||||
// change between multiple method calls
|
||||
func (self *Miner) PendingBlock() *types.Block {
|
||||
return self.worker.pendingBlock()
|
||||
func (m *Miner) PendingBlock() *types.Block {
|
||||
return m.worker.pendingBlock()
|
||||
}
|
||||
|
||||
func (self *Miner) SetEtherbase(addr common.Address) {
|
||||
self.coinbase = addr
|
||||
self.worker.setEtherbase(addr)
|
||||
func (m *Miner) SetEtherbase(addr common.Address) {
|
||||
m.coinbase = addr
|
||||
m.worker.setEtherbase(addr)
|
||||
}
|
||||
|
|
|
|||
248
miner/worker.go
248
miner/worker.go
|
|
@ -162,137 +162,137 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
|
|||
return worker
|
||||
}
|
||||
|
||||
func (self *worker) setEtherbase(addr common.Address) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
self.coinbase = addr
|
||||
func (w *worker) setEtherbase(addr common.Address) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.coinbase = addr
|
||||
}
|
||||
|
||||
func (self *worker) setExtra(extra []byte) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
self.extra = extra
|
||||
func (w *worker) setExtra(extra []byte) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.extra = extra
|
||||
}
|
||||
|
||||
func (self *worker) pending() (*types.Block, *state.StateDB) {
|
||||
if atomic.LoadInt32(&self.mining) == 0 {
|
||||
func (w *worker) pending() (*types.Block, *state.StateDB) {
|
||||
if atomic.LoadInt32(&w.mining) == 0 {
|
||||
// return a snapshot to avoid contention on currentMu mutex
|
||||
self.snapshotMu.RLock()
|
||||
defer self.snapshotMu.RUnlock()
|
||||
return self.snapshotBlock, self.snapshotState.Copy()
|
||||
w.snapshotMu.RLock()
|
||||
defer w.snapshotMu.RUnlock()
|
||||
return w.snapshotBlock, w.snapshotState.Copy()
|
||||
}
|
||||
|
||||
self.currentMu.Lock()
|
||||
defer self.currentMu.Unlock()
|
||||
return self.current.Block, self.current.state.Copy()
|
||||
w.currentMu.Lock()
|
||||
defer w.currentMu.Unlock()
|
||||
return w.current.Block, w.current.state.Copy()
|
||||
}
|
||||
|
||||
func (self *worker) pendingBlock() *types.Block {
|
||||
if atomic.LoadInt32(&self.mining) == 0 {
|
||||
func (w *worker) pendingBlock() *types.Block {
|
||||
if atomic.LoadInt32(&w.mining) == 0 {
|
||||
// return a snapshot to avoid contention on currentMu mutex
|
||||
self.snapshotMu.RLock()
|
||||
defer self.snapshotMu.RUnlock()
|
||||
return self.snapshotBlock
|
||||
w.snapshotMu.RLock()
|
||||
defer w.snapshotMu.RUnlock()
|
||||
return w.snapshotBlock
|
||||
}
|
||||
|
||||
self.currentMu.Lock()
|
||||
defer self.currentMu.Unlock()
|
||||
return self.current.Block
|
||||
w.currentMu.Lock()
|
||||
defer w.currentMu.Unlock()
|
||||
return w.current.Block
|
||||
}
|
||||
|
||||
func (self *worker) start() {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
func (w *worker) start() {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
|
||||
atomic.StoreInt32(&self.mining, 1)
|
||||
atomic.StoreInt32(&w.mining, 1)
|
||||
|
||||
// spin up agents
|
||||
for agent := range self.agents {
|
||||
for agent := range w.agents {
|
||||
agent.Start()
|
||||
}
|
||||
}
|
||||
|
||||
func (self *worker) stop() {
|
||||
self.wg.Wait()
|
||||
func (w *worker) stop() {
|
||||
w.wg.Wait()
|
||||
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
if atomic.LoadInt32(&self.mining) == 1 {
|
||||
for agent := range self.agents {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if atomic.LoadInt32(&w.mining) == 1 {
|
||||
for agent := range w.agents {
|
||||
agent.Stop()
|
||||
}
|
||||
}
|
||||
atomic.StoreInt32(&self.mining, 0)
|
||||
atomic.StoreInt32(&self.atWork, 0)
|
||||
atomic.StoreInt32(&w.mining, 0)
|
||||
atomic.StoreInt32(&w.atWork, 0)
|
||||
}
|
||||
|
||||
func (self *worker) register(agent Agent) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
self.agents[agent] = struct{}{}
|
||||
agent.SetReturnCh(self.recv)
|
||||
func (w *worker) register(agent Agent) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.agents[agent] = struct{}{}
|
||||
agent.SetReturnCh(w.recv)
|
||||
}
|
||||
|
||||
func (self *worker) unregister(agent Agent) {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
delete(self.agents, agent)
|
||||
func (w *worker) unregister(agent Agent) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
delete(w.agents, agent)
|
||||
agent.Stop()
|
||||
}
|
||||
|
||||
func (self *worker) update() {
|
||||
defer self.txSub.Unsubscribe()
|
||||
defer self.chainHeadSub.Unsubscribe()
|
||||
defer self.chainSideSub.Unsubscribe()
|
||||
func (w *worker) update() {
|
||||
defer w.txSub.Unsubscribe()
|
||||
defer w.chainHeadSub.Unsubscribe()
|
||||
defer w.chainSideSub.Unsubscribe()
|
||||
|
||||
for {
|
||||
// A real event arrived, process interesting content
|
||||
select {
|
||||
// Handle ChainHeadEvent
|
||||
case <-self.chainHeadCh:
|
||||
self.commitNewWork()
|
||||
case <-w.chainHeadCh:
|
||||
w.commitNewWork()
|
||||
|
||||
// Handle ChainSideEvent
|
||||
case ev := <-self.chainSideCh:
|
||||
self.uncleMu.Lock()
|
||||
self.possibleUncles[ev.Block.Hash()] = ev.Block
|
||||
self.uncleMu.Unlock()
|
||||
case ev := <-w.chainSideCh:
|
||||
w.uncleMu.Lock()
|
||||
w.possibleUncles[ev.Block.Hash()] = ev.Block
|
||||
w.uncleMu.Unlock()
|
||||
|
||||
// Handle TxPreEvent
|
||||
case ev := <-self.txCh:
|
||||
case ev := <-w.txCh:
|
||||
// Apply transaction to the pending state if we're not mining
|
||||
if atomic.LoadInt32(&self.mining) == 0 {
|
||||
self.currentMu.Lock()
|
||||
acc, _ := types.Sender(self.current.signer, ev.Tx)
|
||||
if atomic.LoadInt32(&w.mining) == 0 {
|
||||
w.currentMu.Lock()
|
||||
acc, _ := types.Sender(w.current.signer, 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)
|
||||
self.updateSnapshot()
|
||||
self.currentMu.Unlock()
|
||||
w.current.commitTransactions(w.mux, txset, w.chain, w.coinbase)
|
||||
w.updateSnapshot()
|
||||
w.currentMu.Unlock()
|
||||
} else {
|
||||
// If we're mining, but nothing is being processed, wake on new transactions
|
||||
if self.config.Clique != nil && self.config.Clique.Period == 0 {
|
||||
self.commitNewWork()
|
||||
if w.config.Clique != nil && w.config.Clique.Period == 0 {
|
||||
w.commitNewWork()
|
||||
}
|
||||
}
|
||||
|
||||
// System stopped
|
||||
case <-self.txSub.Err():
|
||||
case <-w.txSub.Err():
|
||||
return
|
||||
case <-self.chainHeadSub.Err():
|
||||
case <-w.chainHeadSub.Err():
|
||||
return
|
||||
case <-self.chainSideSub.Err():
|
||||
case <-w.chainSideSub.Err():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *worker) wait() {
|
||||
func (w *worker) wait() {
|
||||
for {
|
||||
mustCommitNewWork := true
|
||||
for result := range self.recv {
|
||||
atomic.AddInt32(&self.atWork, -1)
|
||||
for result := range w.recv {
|
||||
atomic.AddInt32(&w.atWork, -1)
|
||||
|
||||
if result == nil {
|
||||
continue
|
||||
|
|
@ -310,7 +310,7 @@ func (self *worker) wait() {
|
|||
for _, log := range work.state.Logs() {
|
||||
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 {
|
||||
log.Error("Failed writing block to chain", "err", err)
|
||||
continue
|
||||
|
|
@ -321,7 +321,7 @@ func (self *worker) wait() {
|
|||
mustCommitNewWork = false
|
||||
}
|
||||
// Broadcast the block and announce chain insertion event
|
||||
self.mux.Post(core.NewMinedBlockEvent{Block: block})
|
||||
w.mux.Post(core.NewMinedBlockEvent{Block: block})
|
||||
var (
|
||||
events []interface{}
|
||||
logs = work.state.Logs()
|
||||
|
|
@ -330,25 +330,25 @@ func (self *worker) wait() {
|
|||
if stat == core.CanonStatTy {
|
||||
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
|
||||
self.unconfirmed.Insert(block.NumberU64(), block.Hash())
|
||||
w.unconfirmed.Insert(block.NumberU64(), block.Hash())
|
||||
|
||||
if mustCommitNewWork {
|
||||
self.commitNewWork()
|
||||
w.commitNewWork()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// push sends a new work task to currently live miner agents.
|
||||
func (self *worker) push(work *Work) {
|
||||
if atomic.LoadInt32(&self.mining) != 1 {
|
||||
func (w *worker) push(work *Work) {
|
||||
if atomic.LoadInt32(&w.mining) != 1 {
|
||||
return
|
||||
}
|
||||
for agent := range self.agents {
|
||||
atomic.AddInt32(&self.atWork, 1)
|
||||
for agent := range w.agents {
|
||||
atomic.AddInt32(&w.atWork, 1)
|
||||
if ch := agent.Work(); ch != nil {
|
||||
ch <- work
|
||||
}
|
||||
|
|
@ -356,14 +356,14 @@ func (self *worker) push(work *Work) {
|
|||
}
|
||||
|
||||
// makeCurrent creates a new environment for the current cycle.
|
||||
func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error {
|
||||
state, err := self.chain.StateAt(parent.Root())
|
||||
func (w *worker) makeCurrent(parent *types.Block, header *types.Header) error {
|
||||
state, err := w.chain.StateAt(parent.Root())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
work := &Work{
|
||||
config: self.config,
|
||||
signer: types.NewEIP155Signer(self.config.ChainId),
|
||||
config: w.config,
|
||||
signer: types.NewEIP155Signer(w.config.ChainId),
|
||||
state: state,
|
||||
ancestors: 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)
|
||||
for _, ancestor := range self.chain.GetBlocksFromHash(parent.Hash(), 7) {
|
||||
for _, ancestor := range w.chain.GetBlocksFromHash(parent.Hash(), 7) {
|
||||
for _, uncle := range ancestor.Uncles() {
|
||||
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
|
||||
work.tcount = 0
|
||||
self.current = work
|
||||
w.current = work
|
||||
return nil
|
||||
}
|
||||
|
||||
func (self *worker) commitNewWork() {
|
||||
self.mu.Lock()
|
||||
defer self.mu.Unlock()
|
||||
self.uncleMu.Lock()
|
||||
defer self.uncleMu.Unlock()
|
||||
self.currentMu.Lock()
|
||||
defer self.currentMu.Unlock()
|
||||
func (w *worker) commitNewWork() {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
w.uncleMu.Lock()
|
||||
defer w.uncleMu.Unlock()
|
||||
w.currentMu.Lock()
|
||||
defer w.currentMu.Unlock()
|
||||
|
||||
tstart := time.Now()
|
||||
parent := self.chain.CurrentBlock()
|
||||
parent := w.chain.CurrentBlock()
|
||||
|
||||
tstamp := tstart.Unix()
|
||||
if parent.Time().Cmp(new(big.Int).SetInt64(tstamp)) >= 0 {
|
||||
|
|
@ -414,24 +414,24 @@ func (self *worker) commitNewWork() {
|
|||
ParentHash: parent.Hash(),
|
||||
Number: num.Add(num, common.Big1),
|
||||
GasLimit: core.CalcGasLimit(parent),
|
||||
Extra: self.extra,
|
||||
Extra: w.extra,
|
||||
Time: big.NewInt(tstamp),
|
||||
}
|
||||
// Only set the coinbase if we are mining (avoid spurious block rewards)
|
||||
if atomic.LoadInt32(&self.mining) == 1 {
|
||||
header.Coinbase = self.coinbase
|
||||
if atomic.LoadInt32(&w.mining) == 1 {
|
||||
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)
|
||||
return
|
||||
}
|
||||
// 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
|
||||
limit := new(big.Int).Add(daoBlock, params.DAOForkExtraRange)
|
||||
if header.Number.Cmp(daoBlock) >= 0 && header.Number.Cmp(limit) < 0 {
|
||||
// Depending whether we support or oppose the fork, override differently
|
||||
if self.config.DAOForkSupport {
|
||||
if w.config.DAOForkSupport {
|
||||
header.Extra = common.CopyBytes(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
|
||||
|
|
@ -439,34 +439,34 @@ func (self *worker) commitNewWork() {
|
|||
}
|
||||
}
|
||||
// Could potentially happen if starting to mine in an odd state.
|
||||
err := self.makeCurrent(parent, header)
|
||||
err := w.makeCurrent(parent, header)
|
||||
if err != nil {
|
||||
log.Error("Failed to create mining context", "err", err)
|
||||
return
|
||||
}
|
||||
// Create the current work task and check any fork transitions needed
|
||||
work := self.current
|
||||
if self.config.DAOForkSupport && self.config.DAOForkBlock != nil && self.config.DAOForkBlock.Cmp(header.Number) == 0 {
|
||||
work := w.current
|
||||
if w.config.DAOForkSupport && w.config.DAOForkBlock != nil && w.config.DAOForkBlock.Cmp(header.Number) == 0 {
|
||||
misc.ApplyDAOHardFork(work.state)
|
||||
}
|
||||
pending, err := self.eth.TxPool().Pending()
|
||||
pending, err := w.eth.TxPool().Pending()
|
||||
if err != nil {
|
||||
log.Error("Failed to fetch pending transactions", "err", err)
|
||||
return
|
||||
}
|
||||
txs := types.NewTransactionsByPriceAndNonce(self.current.signer, pending)
|
||||
work.commitTransactions(self.mux, txs, self.chain, self.coinbase)
|
||||
txs := types.NewTransactionsByPriceAndNonce(w.current.signer, pending)
|
||||
work.commitTransactions(w.mux, txs, w.chain, w.coinbase)
|
||||
|
||||
// compute uncles for the new block.
|
||||
var (
|
||||
uncles []*types.Header
|
||||
badUncles []common.Hash
|
||||
)
|
||||
for hash, uncle := range self.possibleUncles {
|
||||
for hash, uncle := range w.possibleUncles {
|
||||
if len(uncles) == 2 {
|
||||
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(fmt.Sprint(uncle))
|
||||
|
||||
|
|
@ -477,23 +477,23 @@ func (self *worker) commitNewWork() {
|
|||
}
|
||||
}
|
||||
for _, hash := range badUncles {
|
||||
delete(self.possibleUncles, hash)
|
||||
delete(w.possibleUncles, hash)
|
||||
}
|
||||
// Create the new block to seal with the consensus engine
|
||||
if work.Block, err = self.engine.Finalize(self.chain, header, work.state, work.txs, uncles, work.receipts); err != nil {
|
||||
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)
|
||||
return
|
||||
}
|
||||
// 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)))
|
||||
self.unconfirmed.Shift(work.Block.NumberU64() - 1)
|
||||
w.unconfirmed.Shift(work.Block.NumberU64() - 1)
|
||||
}
|
||||
self.push(work)
|
||||
self.updateSnapshot()
|
||||
w.push(work)
|
||||
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()
|
||||
if work.uncles.Has(hash) {
|
||||
return fmt.Errorf("uncle not unique")
|
||||
|
|
@ -508,17 +508,17 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (self *worker) updateSnapshot() {
|
||||
self.snapshotMu.Lock()
|
||||
defer self.snapshotMu.Unlock()
|
||||
func (w *worker) updateSnapshot() {
|
||||
w.snapshotMu.Lock()
|
||||
defer w.snapshotMu.Unlock()
|
||||
|
||||
self.snapshotBlock = types.NewBlock(
|
||||
self.current.header,
|
||||
self.current.txs,
|
||||
w.snapshotBlock = types.NewBlock(
|
||||
w.current.header,
|
||||
w.current.txs,
|
||||
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) {
|
||||
|
|
|
|||
|
|
@ -562,7 +562,7 @@ type preminedTestnet struct {
|
|||
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
|
||||
// fmt.Println("findnode query at dist", toaddr.Port)
|
||||
if toaddr.Port == 0 {
|
||||
|
|
@ -570,7 +570,7 @@ func (tn *preminedTestnet) findnode(toid NodeID, toaddr *net.UDPAddr, target Nod
|
|||
}
|
||||
next := uint16(toaddr.Port) - 1
|
||||
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)))
|
||||
}
|
||||
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
|
||||
// various distances to the given target.
|
||||
func (n *preminedTestnet) mine(target NodeID) {
|
||||
n.target = target
|
||||
n.targetSha = crypto.Keccak256Hash(n.target[:])
|
||||
func (net *preminedTestnet) mine(target NodeID) {
|
||||
net.target = target
|
||||
net.targetSha = crypto.Keccak256Hash(net.target[:])
|
||||
found := 0
|
||||
for found < bucketSize*10 {
|
||||
k := newkey()
|
||||
id := PubkeyID(&k.PublicKey)
|
||||
sha := crypto.Keccak256Hash(id[:])
|
||||
ld := logdist(n.targetSha, sha)
|
||||
if len(n.dists[ld]) < bucketSize {
|
||||
n.dists[ld] = append(n.dists[ld], id)
|
||||
ld := logdist(net.targetSha, sha)
|
||||
if len(net.dists[ld]) < bucketSize {
|
||||
net.dists[ld] = append(net.dists[ld], id)
|
||||
fmt.Println("found ID with ld", ld)
|
||||
found++
|
||||
}
|
||||
}
|
||||
fmt.Println("&preminedTestnet{")
|
||||
fmt.Printf(" target: %#v,\n", n.target)
|
||||
fmt.Printf(" targetSha: %#v,\n", n.targetSha)
|
||||
fmt.Printf(" dists: [%d][]NodeID{\n", len(n.dists))
|
||||
for ld, ns := range n.dists {
|
||||
fmt.Printf(" target: %#v,\n", net.target)
|
||||
fmt.Printf(" targetSha: %#v,\n", net.targetSha)
|
||||
fmt.Printf(" dists: [%d][]NodeID{\n", len(net.dists))
|
||||
for ld, ns := range net.dists {
|
||||
if len(ns) == 0 {
|
||||
continue
|
||||
}
|
||||
|
|
|
|||
|
|
@ -265,11 +265,11 @@ type preminedTestnet struct {
|
|||
net *Network
|
||||
}
|
||||
|
||||
func (tn *preminedTestnet) sendFindnode(to *Node, target NodeID) {
|
||||
func (net *preminedTestnet) sendFindnode(to *Node, target NodeID) {
|
||||
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
|
||||
// fmt.Println("findnode query at dist", toaddr.Port)
|
||||
if to.UDP <= lowPort {
|
||||
|
|
@ -277,21 +277,21 @@ func (tn *preminedTestnet) sendFindnodeHash(to *Node, target common.Hash) {
|
|||
}
|
||||
next := to.UDP - 1
|
||||
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)))
|
||||
}
|
||||
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 {
|
||||
injectResponse(tn.net, to, pongPacket, &pong{ReplyTok: []byte{1}})
|
||||
func (net *preminedTestnet) sendPing(to *Node, addr *net.UDPAddr, topics []Topic) []byte {
|
||||
injectResponse(net.net, to, pongPacket, &pong{ReplyTok: []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 {
|
||||
case pingPacket:
|
||||
injectResponse(tn.net, to, pongPacket, &pong{ReplyTok: []byte{1}})
|
||||
injectResponse(net.net, to, pongPacket, &pong{ReplyTok: []byte{1}})
|
||||
case pongPacket:
|
||||
// ignored
|
||||
case findnodeHashPacket:
|
||||
|
|
@ -302,29 +302,29 @@ func (tn *preminedTestnet) send(to *Node, ptype nodeEvent, data interface{}) (ha
|
|||
}
|
||||
next := to.UDP - 1
|
||||
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)))
|
||||
}
|
||||
injectResponse(tn.net, to, neighborsPacket, &neighbors{Nodes: result})
|
||||
injectResponse(net.net, to, neighborsPacket, &neighbors{Nodes: result})
|
||||
default:
|
||||
panic("send(" + ptype.String() + ")")
|
||||
}
|
||||
return []byte{2}
|
||||
}
|
||||
|
||||
func (tn *preminedTestnet) sendNeighbours(to *Node, nodes []*Node) {
|
||||
func (net *preminedTestnet) sendNeighbours(to *Node, nodes []*Node) {
|
||||
panic("sendNeighbours called")
|
||||
}
|
||||
|
||||
func (tn *preminedTestnet) sendTopicQuery(to *Node, topic Topic) {
|
||||
func (net *preminedTestnet) sendTopicQuery(to *Node, topic Topic) {
|
||||
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")
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
|
||||
|
|
@ -336,26 +336,26 @@ func (*preminedTestnet) localAddr() *net.UDPAddr {
|
|||
|
||||
// mine generates a testnet struct literal with nodes at
|
||||
// various distances to the given target.
|
||||
func (n *preminedTestnet) mine(target NodeID) {
|
||||
n.target = target
|
||||
n.targetSha = crypto.Keccak256Hash(n.target[:])
|
||||
func (net *preminedTestnet) mine(target NodeID) {
|
||||
net.target = target
|
||||
net.targetSha = crypto.Keccak256Hash(net.target[:])
|
||||
found := 0
|
||||
for found < bucketSize*10 {
|
||||
k := newkey()
|
||||
id := PubkeyID(&k.PublicKey)
|
||||
sha := crypto.Keccak256Hash(id[:])
|
||||
ld := logdist(n.targetSha, sha)
|
||||
if len(n.dists[ld]) < bucketSize {
|
||||
n.dists[ld] = append(n.dists[ld], id)
|
||||
ld := logdist(net.targetSha, sha)
|
||||
if len(net.dists[ld]) < bucketSize {
|
||||
net.dists[ld] = append(net.dists[ld], id)
|
||||
fmt.Println("found ID with ld", ld)
|
||||
found++
|
||||
}
|
||||
}
|
||||
fmt.Println("&preminedTestnet{")
|
||||
fmt.Printf(" target: %#v,\n", n.target)
|
||||
fmt.Printf(" targetSha: %#v,\n", n.targetSha)
|
||||
fmt.Printf(" dists: [%d][]NodeID{\n", len(n.dists))
|
||||
for ld, ns := range n.dists {
|
||||
fmt.Println("&preminedTesnetet{")
|
||||
fmt.Printf(" target: %#v,\n", net.target)
|
||||
fmt.Printf(" targetSha: %#v,\n", net.targetSha)
|
||||
fmt.Printf(" dists: [%d][]NodeID{\n", len(net.dists))
|
||||
for ld, ns := range net.dists {
|
||||
if len(ns) == 0 {
|
||||
continue
|
||||
}
|
||||
|
|
|
|||
|
|
@ -315,11 +315,11 @@ func PubkeyID(pub *ecdsa.PublicKey) NodeID {
|
|||
|
||||
// Pubkey returns the public key represented by the node ID.
|
||||
// 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)}
|
||||
half := len(id) / 2
|
||||
p.X.SetBytes(id[:half])
|
||||
p.Y.SetBytes(id[half:])
|
||||
half := len(n) / 2
|
||||
p.X.SetBytes(n[:half])
|
||||
p.Y.SetBytes(n[half:])
|
||||
if !p.Curve.IsOnCurve(p.X, p.Y) {
|
||||
return nil, errors.New("id is invalid secp256k1 curve point")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -304,8 +304,8 @@ func (s ticketRefByWaitTime) Len() int {
|
|||
return len(s)
|
||||
}
|
||||
|
||||
func (r ticketRef) waitTime() mclock.AbsTime {
|
||||
return r.t.regTime[r.idx] - r.t.issueTime
|
||||
func (ref ticketRef) waitTime() mclock.AbsTime {
|
||||
return ref.t.regTime[ref.idx] - ref.t.issueTime
|
||||
}
|
||||
|
||||
// Less reports whether the element with
|
||||
|
|
|
|||
|
|
@ -271,15 +271,15 @@ func (t *topicTable) useTicket(node *Node, serialNo uint32, topics []Topic, idx
|
|||
return false
|
||||
}
|
||||
|
||||
func (topictab *topicTable) getTicket(node *Node, topics []Topic) *ticket {
|
||||
topictab.collectGarbage()
|
||||
func (t *topicTable) getTicket(node *Node, topics []Topic) *ticket {
|
||||
t.collectGarbage()
|
||||
|
||||
now := mclock.Now()
|
||||
n := topictab.getOrNewNode(node)
|
||||
n := t.getOrNewNode(node)
|
||||
n.lastIssuedTicket++
|
||||
topictab.storeTicketCounters(node)
|
||||
t.storeTicketCounters(node)
|
||||
|
||||
t := &ticket{
|
||||
tic := &ticket{
|
||||
issueTime: now,
|
||||
topics: topics,
|
||||
serial: n.lastIssuedTicket,
|
||||
|
|
@ -287,15 +287,15 @@ func (topictab *topicTable) getTicket(node *Node, topics []Topic) *ticket {
|
|||
}
|
||||
for i, topic := range topics {
|
||||
var waitPeriod time.Duration
|
||||
if topic := topictab.topics[topic]; topic != nil {
|
||||
if topic := t.topics[topic]; topic != nil {
|
||||
waitPeriod = topic.wcl.waitPeriod
|
||||
} else {
|
||||
waitPeriod = minWaitPeriod
|
||||
}
|
||||
|
||||
t.regTime[i] = now + mclock.AbsTime(waitPeriod)
|
||||
tic.regTime[i] = now + mclock.AbsTime(waitPeriod)
|
||||
}
|
||||
return t
|
||||
return tic
|
||||
}
|
||||
|
||||
const gcInterval = time.Minute
|
||||
|
|
|
|||
|
|
@ -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
|
||||
// "message received" event
|
||||
func (self *msgEventer) ReadMsg() (Msg, error) {
|
||||
msg, err := self.MsgReadWriter.ReadMsg()
|
||||
func (ev *msgEventer) ReadMsg() (Msg, error) {
|
||||
msg, err := ev.MsgReadWriter.ReadMsg()
|
||||
if err != nil {
|
||||
return msg, err
|
||||
}
|
||||
self.feed.Send(&PeerEvent{
|
||||
ev.feed.Send(&PeerEvent{
|
||||
Type: PeerEventTypeMsgRecv,
|
||||
Peer: self.peerID,
|
||||
Protocol: self.Protocol,
|
||||
Peer: ev.peerID,
|
||||
Protocol: ev.Protocol,
|
||||
MsgCode: &msg.Code,
|
||||
MsgSize: &msg.Size,
|
||||
})
|
||||
|
|
@ -288,15 +288,15 @@ func (self *msgEventer) ReadMsg() (Msg, error) {
|
|||
|
||||
// WriteMsg writes a message to the underlying MsgReadWriter and emits a
|
||||
// "message sent" event
|
||||
func (self *msgEventer) WriteMsg(msg Msg) error {
|
||||
err := self.MsgReadWriter.WriteMsg(msg)
|
||||
func (ev *msgEventer) WriteMsg(msg Msg) error {
|
||||
err := ev.MsgReadWriter.WriteMsg(msg)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
self.feed.Send(&PeerEvent{
|
||||
ev.feed.Send(&PeerEvent{
|
||||
Type: PeerEventTypeMsgSend,
|
||||
Peer: self.peerID,
|
||||
Protocol: self.Protocol,
|
||||
Peer: ev.peerID,
|
||||
Protocol: ev.Protocol,
|
||||
MsgCode: &msg.Code,
|
||||
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
|
||||
// interface
|
||||
func (self *msgEventer) Close() error {
|
||||
if v, ok := self.MsgReadWriter.(io.Closer); ok {
|
||||
func (ev *msgEventer) Close() error {
|
||||
if v, ok := ev.MsgReadWriter.(io.Closer); ok {
|
||||
return v.Close()
|
||||
}
|
||||
return nil
|
||||
|
|
|
|||
|
|
@ -48,8 +48,8 @@ func newPeerError(code int, format string, v ...interface{}) *peerError {
|
|||
return err
|
||||
}
|
||||
|
||||
func (self *peerError) Error() string {
|
||||
return self.message
|
||||
func (pe *peerError) Error() string {
|
||||
return pe.message
|
||||
}
|
||||
|
||||
var errProtocolReturned = errors.New("protocol returned")
|
||||
|
|
|
|||
|
|
@ -154,30 +154,30 @@ type SimNode struct {
|
|||
}
|
||||
|
||||
// Addr returns the node's discovery address
|
||||
func (self *SimNode) Addr() []byte {
|
||||
return []byte(self.Node().String())
|
||||
func (sn *SimNode) Addr() []byte {
|
||||
return []byte(sn.Node().String())
|
||||
}
|
||||
|
||||
// Node returns a discover.Node representing the SimNode
|
||||
func (self *SimNode) Node() *discover.Node {
|
||||
return discover.NewNode(self.ID, net.IP{127, 0, 0, 1}, 30303, 30303)
|
||||
func (sn *SimNode) Node() *discover.Node {
|
||||
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
|
||||
// underlying services (it is set once the node has started)
|
||||
func (self *SimNode) Client() (*rpc.Client, error) {
|
||||
self.lock.RLock()
|
||||
defer self.lock.RUnlock()
|
||||
if self.client == nil {
|
||||
func (sn *SimNode) Client() (*rpc.Client, error) {
|
||||
sn.lock.RLock()
|
||||
defer sn.lock.RUnlock()
|
||||
if sn.client == nil {
|
||||
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
|
||||
// in-memory client to the node's RPC server
|
||||
func (self *SimNode) ServeRPC(conn net.Conn) error {
|
||||
handler, err := self.node.RPCHandler()
|
||||
func (sn *SimNode) ServeRPC(conn net.Conn) error {
|
||||
handler, err := sn.node.RPCHandler()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -187,13 +187,13 @@ func (self *SimNode) ServeRPC(conn net.Conn) error {
|
|||
|
||||
// Snapshots creates snapshots of the services by calling the
|
||||
// simulation_snapshot RPC method
|
||||
func (self *SimNode) Snapshots() (map[string][]byte, error) {
|
||||
self.lock.RLock()
|
||||
services := make(map[string]node.Service, len(self.running))
|
||||
for name, service := range self.running {
|
||||
func (sn *SimNode) Snapshots() (map[string][]byte, error) {
|
||||
sn.lock.RLock()
|
||||
services := make(map[string]node.Service, len(sn.running))
|
||||
for name, service := range sn.running {
|
||||
services[name] = service
|
||||
}
|
||||
self.lock.RUnlock()
|
||||
sn.lock.RUnlock()
|
||||
if len(services) == 0 {
|
||||
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
|
||||
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) {
|
||||
return func(nodeCtx *node.ServiceContext) (node.Service, error) {
|
||||
ctx := &ServiceContext{
|
||||
RPCDialer: self.adapter,
|
||||
RPCDialer: sn.adapter,
|
||||
NodeContext: nodeCtx,
|
||||
Config: self.config,
|
||||
Config: sn.config,
|
||||
}
|
||||
if snapshots != nil {
|
||||
ctx.Snapshot = snapshots[name]
|
||||
}
|
||||
serviceFunc := self.adapter.services[name]
|
||||
serviceFunc := sn.adapter.services[name]
|
||||
service, err := serviceFunc(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
self.running[name] = service
|
||||
sn.running[name] = service
|
||||
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
|
||||
// being stopped and then started again
|
||||
var regErr error
|
||||
self.registerOnce.Do(func() {
|
||||
for _, name := range self.config.Services {
|
||||
if err := self.node.Register(newService(name)); err != nil {
|
||||
sn.registerOnce.Do(func() {
|
||||
for _, name := range sn.config.Services {
|
||||
if err := sn.node.Register(newService(name)); err != nil {
|
||||
regErr = err
|
||||
return
|
||||
}
|
||||
|
|
@ -249,54 +249,55 @@ func (self *SimNode) Start(snapshots map[string][]byte) error {
|
|||
return regErr
|
||||
}
|
||||
|
||||
if err := self.node.Start(); err != nil {
|
||||
if err := sn.node.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// create an in-process RPC client
|
||||
handler, err := self.node.RPCHandler()
|
||||
handler, err := sn.node.RPCHandler()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
self.lock.Lock()
|
||||
self.client = rpc.DialInProc(handler)
|
||||
self.lock.Unlock()
|
||||
sn.lock.Lock()
|
||||
sn.client = rpc.DialInProc(handler)
|
||||
sn.lock.Unlock()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
// Stop closes the RPC client and stops the underlying devp2p node
|
||||
func (self *SimNode) Stop() error {
|
||||
self.lock.Lock()
|
||||
if self.client != nil {
|
||||
self.client.Close()
|
||||
self.client = nil
|
||||
func (sn *SimNode) Stop() error {
|
||||
sn.lock.Lock()
|
||||
if sn.client != nil {
|
||||
sn.client.Close()
|
||||
sn.client = nil
|
||||
}
|
||||
self.lock.Unlock()
|
||||
return self.node.Stop()
|
||||
sn.lock.Unlock()
|
||||
return sn.node.Stop()
|
||||
}
|
||||
|
||||
// Services returns a copy of the underlying services
|
||||
func (self *SimNode) Services() []node.Service {
|
||||
self.lock.RLock()
|
||||
defer self.lock.RUnlock()
|
||||
services := make([]node.Service, 0, len(self.running))
|
||||
for _, service := range self.running {
|
||||
func (sn *SimNode) Services() []node.Service {
|
||||
sn.lock.RLock()
|
||||
defer sn.lock.RUnlock()
|
||||
services := make([]node.Service, 0, len(sn.running))
|
||||
for _, service := range sn.running {
|
||||
services = append(services, service)
|
||||
}
|
||||
return services
|
||||
}
|
||||
|
||||
// Server returns the underlying p2p.Server
|
||||
func (self *SimNode) Server() *p2p.Server {
|
||||
return self.node.Server()
|
||||
func (sn *SimNode) Server() *p2p.Server {
|
||||
return sn.node.Server()
|
||||
}
|
||||
|
||||
// SubscribeEvents subscribes the given channel to peer events from the
|
||||
// underlying p2p.Server
|
||||
func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
|
||||
srv := self.Server()
|
||||
func (sn *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription {
|
||||
srv := sn.Server()
|
||||
if srv == nil {
|
||||
panic("node not running")
|
||||
}
|
||||
|
|
@ -304,12 +305,12 @@ func (self *SimNode) SubscribeEvents(ch chan *p2p.PeerEvent) event.Subscription
|
|||
}
|
||||
|
||||
// NodeInfo returns information about the node
|
||||
func (self *SimNode) NodeInfo() *p2p.NodeInfo {
|
||||
server := self.Server()
|
||||
func (sn *SimNode) NodeInfo() *p2p.NodeInfo {
|
||||
server := sn.Server()
|
||||
if server == nil {
|
||||
return &p2p.NodeInfo{
|
||||
ID: self.ID.String(),
|
||||
Enode: self.Node().String(),
|
||||
ID: sn.ID.String(),
|
||||
Enode: sn.Node().String(),
|
||||
}
|
||||
}
|
||||
return server.NodeInfo()
|
||||
|
|
|
|||
|
|
@ -20,12 +20,12 @@ type SimStateStore struct {
|
|||
m map[string][]byte
|
||||
}
|
||||
|
||||
func (self *SimStateStore) Load(s string) ([]byte, error) {
|
||||
return self.m[s], nil
|
||||
func (st *SimStateStore) Load(s string) ([]byte, error) {
|
||||
return st.m[s], nil
|
||||
}
|
||||
|
||||
func (self *SimStateStore) Save(s string, data []byte) error {
|
||||
self.m[s] = data
|
||||
func (st *SimStateStore) Save(s string, data []byte) error {
|
||||
st.m[s] = data
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -74,22 +74,22 @@ func NewNetwork(nodeAdapter adapters.NodeAdapter, conf *NetworkConfig) *Network
|
|||
}
|
||||
|
||||
// Events returns the output event feed of the Network.
|
||||
func (self *Network) Events() *event.Feed {
|
||||
return &self.events
|
||||
func (net *Network) Events() *event.Feed {
|
||||
return &net.events
|
||||
}
|
||||
|
||||
// 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.Services = []string{self.DefaultService}
|
||||
return self.NewNodeWithConfig(conf)
|
||||
conf.Services = []string{net.DefaultService}
|
||||
return net.NewNodeWithConfig(conf)
|
||||
}
|
||||
|
||||
// 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
|
||||
func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (net *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error) {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
|
||||
// create a random ID and PrivateKey if not set
|
||||
if conf.ID == (discover.NodeID{}) {
|
||||
|
|
@ -100,31 +100,31 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
|
|||
id := conf.ID
|
||||
if conf.Reachable == nil {
|
||||
conf.Reachable = func(otherID discover.NodeID) bool {
|
||||
_, err := self.InitConn(conf.ID, otherID)
|
||||
_, err := net.InitConn(conf.ID, otherID)
|
||||
return err == nil
|
||||
}
|
||||
}
|
||||
|
||||
// assign a name to the node if not set
|
||||
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
|
||||
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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
// if no services are configured, use the default service
|
||||
if len(conf.Services) == 0 {
|
||||
conf.Services = []string{self.DefaultService}
|
||||
conf.Services = []string{net.DefaultService}
|
||||
}
|
||||
|
||||
// use the NodeAdapter to create the node
|
||||
adapterNode, err := self.nodeAdapter.NewNode(conf)
|
||||
adapterNode, err := net.nodeAdapter.NewNode(conf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -133,27 +133,27 @@ func (self *Network) NewNodeWithConfig(conf *adapters.NodeConfig) (*Node, error)
|
|||
Config: conf,
|
||||
}
|
||||
log.Trace(fmt.Sprintf("node %v created", id))
|
||||
self.nodeMap[id] = len(self.Nodes)
|
||||
self.Nodes = append(self.Nodes, node)
|
||||
net.nodeMap[id] = len(net.Nodes)
|
||||
net.Nodes = append(net.Nodes, node)
|
||||
|
||||
// emit a "control" event
|
||||
self.events.Send(ControlEvent(node))
|
||||
net.events.Send(ControlEvent(node))
|
||||
|
||||
return node, nil
|
||||
}
|
||||
|
||||
// Config returns the network configuration
|
||||
func (self *Network) Config() *NetworkConfig {
|
||||
return &self.NetworkConfig
|
||||
func (net *Network) Config() *NetworkConfig {
|
||||
return &net.NetworkConfig
|
||||
}
|
||||
|
||||
// StartAll starts all nodes in the network
|
||||
func (self *Network) StartAll() error {
|
||||
for _, node := range self.Nodes {
|
||||
func (net *Network) StartAll() error {
|
||||
for _, node := range net.Nodes {
|
||||
if node.Up {
|
||||
continue
|
||||
}
|
||||
if err := self.Start(node.ID()); err != nil {
|
||||
if err := net.Start(node.ID()); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
@ -161,12 +161,12 @@ func (self *Network) StartAll() error {
|
|||
}
|
||||
|
||||
// StopAll stops all nodes in the network
|
||||
func (self *Network) StopAll() error {
|
||||
for _, node := range self.Nodes {
|
||||
func (net *Network) StopAll() error {
|
||||
for _, node := range net.Nodes {
|
||||
if !node.Up {
|
||||
continue
|
||||
}
|
||||
if err := self.Stop(node.ID()); err != nil {
|
||||
if err := net.Stop(node.ID()); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
@ -174,21 +174,21 @@ func (self *Network) StopAll() error {
|
|||
}
|
||||
|
||||
// Start starts the node with the given ID
|
||||
func (self *Network) Start(id discover.NodeID) error {
|
||||
return self.startWithSnapshots(id, nil)
|
||||
func (net *Network) Start(id discover.NodeID) error {
|
||||
return net.startWithSnapshots(id, nil)
|
||||
}
|
||||
|
||||
// startWithSnapshots starts the node with the given ID using the give
|
||||
// snapshots
|
||||
func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error {
|
||||
node := self.GetNode(id)
|
||||
func (net *Network) startWithSnapshots(id discover.NodeID, snapshots map[string][]byte) error {
|
||||
node := net.GetNode(id)
|
||||
if node == nil {
|
||||
return fmt.Errorf("node %v does not exist", id)
|
||||
}
|
||||
if node.Up {
|
||||
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 {
|
||||
log.Warn(fmt.Sprintf("start up failed: %v", err))
|
||||
return err
|
||||
|
|
@ -196,7 +196,7 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string
|
|||
node.Up = true
|
||||
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
|
||||
client, err := node.Client()
|
||||
|
|
@ -208,22 +208,22 @@ func (self *Network) startWithSnapshots(id discover.NodeID, snapshots map[string
|
|||
if err != nil {
|
||||
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
|
||||
}
|
||||
|
||||
// watchPeerEvents reads peer events from the given channel and emits
|
||||
// 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() {
|
||||
sub.Unsubscribe()
|
||||
|
||||
// assume the node is now down
|
||||
self.lock.Lock()
|
||||
node := self.getNode(id)
|
||||
net.lock.Lock()
|
||||
node := net.getNode(id)
|
||||
node.Up = false
|
||||
self.lock.Unlock()
|
||||
self.events.Send(NewEvent(node))
|
||||
net.lock.Unlock()
|
||||
net.events.Send(NewEvent(node))
|
||||
}()
|
||||
for {
|
||||
select {
|
||||
|
|
@ -235,16 +235,16 @@ func (self *Network) watchPeerEvents(id discover.NodeID, events chan *p2p.PeerEv
|
|||
switch event.Type {
|
||||
|
||||
case p2p.PeerEventTypeAdd:
|
||||
self.DidConnect(id, peer)
|
||||
net.DidConnect(id, peer)
|
||||
|
||||
case p2p.PeerEventTypeDrop:
|
||||
self.DidDisconnect(id, peer)
|
||||
net.DidDisconnect(id, peer)
|
||||
|
||||
case p2p.PeerEventTypeMsgSend:
|
||||
self.DidSend(id, peer, event.Protocol, *event.MsgCode)
|
||||
net.DidSend(id, peer, event.Protocol, *event.MsgCode)
|
||||
|
||||
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
|
||||
func (self *Network) Stop(id discover.NodeID) error {
|
||||
node := self.GetNode(id)
|
||||
func (net *Network) Stop(id discover.NodeID) error {
|
||||
node := net.GetNode(id)
|
||||
if node == nil {
|
||||
return fmt.Errorf("node %v does not exist", id)
|
||||
}
|
||||
|
|
@ -272,15 +272,15 @@ func (self *Network) Stop(id discover.NodeID) error {
|
|||
node.Up = false
|
||||
log.Info(fmt.Sprintf("stop node %v: %v", id, node.Up))
|
||||
|
||||
self.events.Send(ControlEvent(node))
|
||||
net.events.Send(ControlEvent(node))
|
||||
return nil
|
||||
}
|
||||
|
||||
// Connect connects two nodes together by calling the "admin_addPeer" RPC
|
||||
// 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))
|
||||
conn, err := self.InitConn(oneID, otherID)
|
||||
conn, err := net.InitConn(oneID, otherID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -288,14 +288,14 @@ func (self *Network) Connect(oneID, otherID discover.NodeID) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
self.events.Send(ControlEvent(conn))
|
||||
net.events.Send(ControlEvent(conn))
|
||||
return client.Call(nil, "admin_addPeer", string(conn.other.Addr()))
|
||||
}
|
||||
|
||||
// Disconnect disconnects two nodes by calling the "admin_removePeer" RPC
|
||||
// method on the "one" node so that it disconnects from the "other" node
|
||||
func (self *Network) Disconnect(oneID, otherID discover.NodeID) error {
|
||||
conn := self.GetConn(oneID, otherID)
|
||||
func (net *Network) Disconnect(oneID, otherID discover.NodeID) error {
|
||||
conn := net.GetConn(oneID, otherID)
|
||||
if conn == nil {
|
||||
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 {
|
||||
return err
|
||||
}
|
||||
self.events.Send(ControlEvent(conn))
|
||||
net.events.Send(ControlEvent(conn))
|
||||
return client.Call(nil, "admin_removePeer", string(conn.other.Addr()))
|
||||
}
|
||||
|
||||
// DidConnect tracks the fact that the "one" node connected to the "other" node
|
||||
func (self *Network) DidConnect(one, other discover.NodeID) error {
|
||||
conn, err := self.GetOrCreateConn(one, other)
|
||||
func (net *Network) DidConnect(one, other discover.NodeID) error {
|
||||
conn, err := net.GetOrCreateConn(one, other)
|
||||
if err != nil {
|
||||
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)
|
||||
}
|
||||
conn.Up = true
|
||||
self.events.Send(NewEvent(conn))
|
||||
net.events.Send(NewEvent(conn))
|
||||
return nil
|
||||
}
|
||||
|
||||
// DidDisconnect tracks the fact that the "one" node disconnected from the
|
||||
// "other" node
|
||||
func (self *Network) DidDisconnect(one, other discover.NodeID) error {
|
||||
conn := self.GetConn(one, other)
|
||||
func (net *Network) DidDisconnect(one, other discover.NodeID) error {
|
||||
conn := net.GetConn(one, other)
|
||||
if conn == nil {
|
||||
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.initiated = time.Now().Add(-dialBanTimeout)
|
||||
self.events.Send(NewEvent(conn))
|
||||
net.events.Send(NewEvent(conn))
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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{
|
||||
One: sender,
|
||||
Other: receiver,
|
||||
|
|
@ -349,12 +349,12 @@ func (self *Network) DidSend(sender, receiver discover.NodeID, proto string, cod
|
|||
Code: code,
|
||||
Received: false,
|
||||
}
|
||||
self.events.Send(NewEvent(msg))
|
||||
net.events.Send(NewEvent(msg))
|
||||
return nil
|
||||
}
|
||||
|
||||
// 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{
|
||||
One: sender,
|
||||
Other: receiver,
|
||||
|
|
@ -362,36 +362,36 @@ func (self *Network) DidReceive(sender, receiver discover.NodeID, proto string,
|
|||
Code: code,
|
||||
Received: true,
|
||||
}
|
||||
self.events.Send(NewEvent(msg))
|
||||
net.events.Send(NewEvent(msg))
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetNode gets the node with the given ID, returning nil if the node does not
|
||||
// exist
|
||||
func (self *Network) GetNode(id discover.NodeID) *Node {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
return self.getNode(id)
|
||||
func (net *Network) GetNode(id discover.NodeID) *Node {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
return net.getNode(id)
|
||||
}
|
||||
|
||||
// GetNode gets the node with the given name, returning nil if the node does
|
||||
// not exist
|
||||
func (self *Network) GetNodeByName(name string) *Node {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
return self.getNodeByName(name)
|
||||
func (net *Network) GetNodeByName(name string) *Node {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
return net.getNodeByName(name)
|
||||
}
|
||||
|
||||
func (self *Network) getNode(id discover.NodeID) *Node {
|
||||
i, found := self.nodeMap[id]
|
||||
func (net *Network) getNode(id discover.NodeID) *Node {
|
||||
i, found := net.nodeMap[id]
|
||||
if !found {
|
||||
return nil
|
||||
}
|
||||
return self.Nodes[i]
|
||||
return net.Nodes[i]
|
||||
}
|
||||
|
||||
func (self *Network) getNodeByName(name string) *Node {
|
||||
for _, node := range self.Nodes {
|
||||
func (net *Network) getNodeByName(name string) *Node {
|
||||
for _, node := range net.Nodes {
|
||||
if node.Config.Name == name {
|
||||
return node
|
||||
}
|
||||
|
|
@ -400,40 +400,40 @@ func (self *Network) getNodeByName(name string) *Node {
|
|||
}
|
||||
|
||||
// GetNodes returns the existing nodes
|
||||
func (self *Network) GetNodes() (nodes []*Node) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (net *Network) GetNodes() (nodes []*Node) {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
|
||||
nodes = append(nodes, self.Nodes...)
|
||||
nodes = append(nodes, net.Nodes...)
|
||||
return nodes
|
||||
}
|
||||
|
||||
// GetConn returns the connection which exists between "one" and "other"
|
||||
// regardless of which node initiated the connection
|
||||
func (self *Network) GetConn(oneID, otherID discover.NodeID) *Conn {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
return self.getConn(oneID, otherID)
|
||||
func (net *Network) GetConn(oneID, otherID discover.NodeID) *Conn {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
return net.getConn(oneID, otherID)
|
||||
}
|
||||
|
||||
// GetOrCreateConn is like GetConn but creates the connection if it doesn't
|
||||
// already exist
|
||||
func (self *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
return self.getOrCreateConn(oneID, otherID)
|
||||
func (net *Network) GetOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
return net.getOrCreateConn(oneID, otherID)
|
||||
}
|
||||
|
||||
func (self *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
|
||||
if conn := self.getConn(oneID, otherID); conn != nil {
|
||||
func (net *Network) getOrCreateConn(oneID, otherID discover.NodeID) (*Conn, error) {
|
||||
if conn := net.getConn(oneID, otherID); conn != nil {
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
one := self.getNode(oneID)
|
||||
one := net.getNode(oneID)
|
||||
if one == nil {
|
||||
return nil, fmt.Errorf("node %v does not exist", oneID)
|
||||
}
|
||||
other := self.getNode(otherID)
|
||||
other := net.getNode(otherID)
|
||||
if other == nil {
|
||||
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,
|
||||
}
|
||||
label := ConnLabel(oneID, otherID)
|
||||
self.connMap[label] = len(self.Conns)
|
||||
self.Conns = append(self.Conns, conn)
|
||||
net.connMap[label] = len(net.Conns)
|
||||
net.Conns = append(net.Conns, conn)
|
||||
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)
|
||||
i, found := self.connMap[label]
|
||||
i, found := net.connMap[label]
|
||||
if !found {
|
||||
return nil
|
||||
}
|
||||
return self.Conns[i]
|
||||
return net.Conns[i]
|
||||
}
|
||||
|
||||
// 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
|
||||
// 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
|
||||
func (self *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (net *Network) InitConn(oneID, otherID discover.NodeID) (*Conn, error) {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
if oneID == otherID {
|
||||
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 {
|
||||
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
|
||||
func (self *Network) Shutdown() {
|
||||
for _, node := range self.Nodes {
|
||||
func (net *Network) Shutdown() {
|
||||
for _, node := range net.Nodes {
|
||||
log.Debug(fmt.Sprintf("stopping node %s", node.ID().TerminalString()))
|
||||
if err := node.Stop(); err != nil {
|
||||
log.Warn(fmt.Sprintf("error stopping node %s", node.ID().TerminalString()), "err", err)
|
||||
}
|
||||
}
|
||||
close(self.quitc)
|
||||
close(net.quitc)
|
||||
}
|
||||
|
||||
//Reset resets all network properties:
|
||||
//emtpies the nodes and the connection list
|
||||
func (self *Network) Reset() {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (net *Network) Reset() {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
|
||||
//re-initialize the maps
|
||||
self.connMap = make(map[string]int)
|
||||
self.nodeMap = make(map[discover.NodeID]int)
|
||||
net.connMap = make(map[string]int)
|
||||
net.nodeMap = make(map[discover.NodeID]int)
|
||||
|
||||
self.Nodes = nil
|
||||
self.Conns = nil
|
||||
net.Nodes = nil
|
||||
net.Conns = nil
|
||||
}
|
||||
|
||||
// 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
|
||||
func (self *Node) ID() discover.NodeID {
|
||||
return self.Config.ID
|
||||
func (n *Node) ID() discover.NodeID {
|
||||
return n.Config.ID
|
||||
}
|
||||
|
||||
// String returns a log-friendly string
|
||||
func (self *Node) String() string {
|
||||
return fmt.Sprintf("Node %v", self.ID().TerminalString())
|
||||
func (n *Node) String() string {
|
||||
return fmt.Sprintf("Node %v", n.ID().TerminalString())
|
||||
}
|
||||
|
||||
// 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
|
||||
if self.Node == nil {
|
||||
if n.Node == nil {
|
||||
return nil
|
||||
}
|
||||
info := self.Node.NodeInfo()
|
||||
info.Name = self.Config.Name
|
||||
info := n.Node.NodeInfo()
|
||||
info.Name = n.Config.Name
|
||||
return info
|
||||
}
|
||||
|
||||
// MarshalJSON implements the json.Marshaler interface so that the encoded
|
||||
// JSON includes the NodeInfo
|
||||
func (self *Node) MarshalJSON() ([]byte, error) {
|
||||
func (n *Node) MarshalJSON() ([]byte, error) {
|
||||
return json.Marshal(struct {
|
||||
Info *p2p.NodeInfo `json:"info,omitempty"`
|
||||
Config *adapters.NodeConfig `json:"config,omitempty"`
|
||||
Up bool `json:"up"`
|
||||
}{
|
||||
Info: self.NodeInfo(),
|
||||
Config: self.Config,
|
||||
Up: self.Up,
|
||||
Info: n.NodeInfo(),
|
||||
Config: n.Config,
|
||||
Up: n.Up,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -580,19 +580,19 @@ type Conn struct {
|
|||
}
|
||||
|
||||
// nodesUp returns whether both nodes are currently up
|
||||
func (self *Conn) nodesUp() error {
|
||||
if !self.one.Up {
|
||||
return fmt.Errorf("one %v is not up", self.One)
|
||||
func (c *Conn) nodesUp() error {
|
||||
if !c.one.Up {
|
||||
return fmt.Errorf("one %v is not up", c.One)
|
||||
}
|
||||
if !self.other.Up {
|
||||
return fmt.Errorf("other %v is not up", self.Other)
|
||||
if !c.other.Up {
|
||||
return fmt.Errorf("other %v is not up", c.Other)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// String returns a log-friendly string
|
||||
func (self *Conn) String() string {
|
||||
return fmt.Sprintf("Conn %v->%v", self.One.TerminalString(), self.Other.TerminalString())
|
||||
func (c *Conn) String() string {
|
||||
return fmt.Sprintf("Conn %v->%v", c.One.TerminalString(), c.Other.TerminalString())
|
||||
}
|
||||
|
||||
// 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
|
||||
func (self *Msg) String() string {
|
||||
return fmt.Sprintf("Msg(%d) %v->%v", self.Code, self.One.TerminalString(), self.Other.TerminalString())
|
||||
func (m *Msg) String() string {
|
||||
return fmt.Sprintf("Msg(%d) %v->%v", m.Code, m.One.TerminalString(), m.Other.TerminalString())
|
||||
}
|
||||
|
||||
// ConnLabel generates a deterministic string which represents a connection
|
||||
|
|
@ -640,14 +640,14 @@ type NodeSnapshot struct {
|
|||
}
|
||||
|
||||
// Snapshot creates a network snapshot
|
||||
func (self *Network) Snapshot() (*Snapshot, error) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
func (net *Network) Snapshot() (*Snapshot, error) {
|
||||
net.lock.Lock()
|
||||
defer net.lock.Unlock()
|
||||
snap := &Snapshot{
|
||||
Nodes: make([]NodeSnapshot, len(self.Nodes)),
|
||||
Conns: make([]Conn, len(self.Conns)),
|
||||
Nodes: make([]NodeSnapshot, len(net.Nodes)),
|
||||
Conns: make([]Conn, len(net.Conns)),
|
||||
}
|
||||
for i, node := range self.Nodes {
|
||||
for i, node := range net.Nodes {
|
||||
snap.Nodes[i] = NodeSnapshot{Node: *node}
|
||||
if !node.Up {
|
||||
continue
|
||||
|
|
@ -658,33 +658,33 @@ func (self *Network) Snapshot() (*Snapshot, error) {
|
|||
}
|
||||
snap.Nodes[i].Snapshots = snapshots
|
||||
}
|
||||
for i, conn := range self.Conns {
|
||||
for i, conn := range net.Conns {
|
||||
snap.Conns[i] = *conn
|
||||
}
|
||||
return snap, nil
|
||||
}
|
||||
|
||||
// Load loads a network snapshot
|
||||
func (self *Network) Load(snap *Snapshot) error {
|
||||
func (net *Network) Load(snap *Snapshot) error {
|
||||
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
|
||||
}
|
||||
if !n.Node.Up {
|
||||
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
|
||||
}
|
||||
}
|
||||
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,
|
||||
//so it would result in the snapshot `Load` to fail
|
||||
continue
|
||||
}
|
||||
if err := self.Connect(conn.One, conn.Other); err != nil {
|
||||
if err := net.Connect(conn.One, conn.Other); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
|
@ -692,7 +692,7 @@ func (self *Network) Load(snap *Snapshot) error {
|
|||
}
|
||||
|
||||
// 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 {
|
||||
select {
|
||||
case event, ok := <-events:
|
||||
|
|
@ -700,23 +700,23 @@ func (self *Network) Subscribe(events chan *Event) {
|
|||
return
|
||||
}
|
||||
if event.Control {
|
||||
self.executeControlEvent(event)
|
||||
net.executeControlEvent(event)
|
||||
}
|
||||
case <-self.quitc:
|
||||
case <-net.quitc:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (self *Network) executeControlEvent(event *Event) {
|
||||
func (net *Network) executeControlEvent(event *Event) {
|
||||
log.Trace("execute control event", "type", event.Type, "event", event)
|
||||
switch event.Type {
|
||||
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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
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 {
|
||||
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 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 {
|
||||
return self.Connect(e.Conn.One, e.Conn.Other)
|
||||
return net.Connect(e.Conn.One, e.Conn.Other)
|
||||
} else {
|
||||
return self.Disconnect(e.Conn.One, e.Conn.Other)
|
||||
return net.Disconnect(e.Conn.One, e.Conn.Other)
|
||||
}
|
||||
}
|
||||
|
|
@ -39,29 +39,29 @@ func NewTestPeerPool() *TestPeerPool {
|
|||
return &TestPeerPool{peers: make(map[discover.NodeID]TestPeer)}
|
||||
}
|
||||
|
||||
func (self *TestPeerPool) Add(p TestPeer) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
log.Trace(fmt.Sprintf("pp add peer %v", p.ID()))
|
||||
self.peers[p.ID()] = p
|
||||
func (p *TestPeerPool) Add(peer TestPeer) {
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
log.Trace(fmt.Sprintf("pp add peer %v", peer.ID()))
|
||||
p.peers[peer.ID()] = peer
|
||||
|
||||
}
|
||||
|
||||
func (self *TestPeerPool) Remove(p TestPeer) {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
delete(self.peers, p.ID())
|
||||
func (p *TestPeerPool) Remove(peer TestPeer) {
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
delete(p.peers, peer.ID())
|
||||
}
|
||||
|
||||
func (self *TestPeerPool) Has(id discover.NodeID) bool {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
_, ok := self.peers[id]
|
||||
func (p *TestPeerPool) Has(id discover.NodeID) bool {
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
_, ok := p.peers[id]
|
||||
return ok
|
||||
}
|
||||
|
||||
func (self *TestPeerPool) Get(id discover.NodeID) TestPeer {
|
||||
self.lock.Lock()
|
||||
defer self.lock.Unlock()
|
||||
return self.peers[id]
|
||||
func (p *TestPeerPool) Get(id discover.NodeID) TestPeer {
|
||||
p.lock.Lock()
|
||||
defer p.lock.Unlock()
|
||||
return p.peers[id]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -78,10 +78,10 @@ type Disconnect struct {
|
|||
}
|
||||
|
||||
// trigger sends messages from peers
|
||||
func (self *ProtocolSession) trigger(trig Trigger) error {
|
||||
simNode, ok := self.adapter.GetNode(trig.Peer)
|
||||
func (s *ProtocolSession) trigger(trig Trigger) error {
|
||||
simNode, ok := s.adapter.GetNode(trig.Peer)
|
||||
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)
|
||||
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
|
||||
func (self *ProtocolSession) expect(exps []Expect) error {
|
||||
func (s *ProtocolSession) expect(exps []Expect) error {
|
||||
// construct a map of expectations for each node
|
||||
peerExpects := make(map[discover.NodeID][]Expect)
|
||||
for _, exp := range exps {
|
||||
|
|
@ -120,9 +120,9 @@ func (self *ProtocolSession) expect(exps []Expect) error {
|
|||
// construct a map of mockNodes for each node
|
||||
mockNodes := make(map[discover.NodeID]*mockNode)
|
||||
for nodeID := range peerExpects {
|
||||
simNode, ok := self.adapter.GetNode(nodeID)
|
||||
simNode, ok := s.adapter.GetNode(nodeID)
|
||||
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)
|
||||
if !ok {
|
||||
|
|
@ -202,9 +202,9 @@ func (self *ProtocolSession) expect(exps []Expect) error {
|
|||
}
|
||||
|
||||
// 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 {
|
||||
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)
|
||||
}
|
||||
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.
|
||||
// Default timeout value is 2 seconds.
|
||||
func (self *ProtocolSession) testExchange(e Exchange) error {
|
||||
func (s *ProtocolSession) testExchange(e Exchange) error {
|
||||
errc := make(chan error)
|
||||
done := make(chan struct{})
|
||||
defer close(done)
|
||||
|
||||
go func() {
|
||||
for _, trig := range e.Triggers {
|
||||
err := self.trigger(trig)
|
||||
err := s.trigger(trig)
|
||||
if err != nil {
|
||||
errc <- err
|
||||
return
|
||||
|
|
@ -229,7 +229,7 @@ func (self *ProtocolSession) testExchange(e Exchange) error {
|
|||
}
|
||||
|
||||
select {
|
||||
case errc <- self.expect(e.Expects):
|
||||
case errc <- s.expect(e.Expects):
|
||||
case <-done:
|
||||
}
|
||||
}()
|
||||
|
|
@ -250,7 +250,7 @@ func (self *ProtocolSession) testExchange(e Exchange) error {
|
|||
|
||||
// TestDisconnected tests the disconnections given as arguments
|
||||
// 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)
|
||||
for _, disconnect := range disconnects {
|
||||
expects[disconnect.Peer] = disconnect.Error
|
||||
|
|
@ -259,7 +259,7 @@ func (self *ProtocolSession) TestDisconnected(disconnects ...*Disconnect) error
|
|||
timeout := time.After(time.Second)
|
||||
for len(expects) > 0 {
|
||||
select {
|
||||
case event := <-self.events:
|
||||
case event := <-s.events:
|
||||
if event.Type != p2p.PeerEventTypeDrop {
|
||||
continue
|
||||
}
|
||||
|
|
|
|||
|
|
@ -101,24 +101,24 @@ func NewProtocolTester(t *testing.T, id discover.NodeID, n int, run func(*p2p.Pe
|
|||
}
|
||||
|
||||
// Stop stops the p2p server
|
||||
func (self *ProtocolTester) Stop() error {
|
||||
self.Server.Stop()
|
||||
func (t *ProtocolTester) Stop() error {
|
||||
t.Server.Stop()
|
||||
return nil
|
||||
}
|
||||
|
||||
// Connect brings up the remote peer node and connects it using the
|
||||
// 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 {
|
||||
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))
|
||||
}
|
||||
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))
|
||||
}
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue