roysc/plugeth
1
0
Fork 0
forked from cerc-io/plugeth
Code Pull Requests Activity

Merge remote-tracking branch 'ethereum/conversion' into conversion

Browse Source
This commit is contained in:
Felix Lange 2015-03-20 22:47:27 +01:00
commit 81800ca39e
33 changed files with 1499 additions and 1256 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

250
blockpool/blockpool.go
View File

@ -7,8 +7,10 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/errs" "github.com/ethereum/go-ethereum/errs"
"github.com/ethereum/go-ethereum/event"
ethlogger "github.com/ethereum/go-ethereum/logger" ethlogger "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/pow" "github.com/ethereum/go-ethereum/pow"
) )
@ -32,12 +34,15 @@ var (
blockHashesTimeout = 60 * time.Second blockHashesTimeout = 60 * time.Second
// timeout interval: max time allowed for peer without sending a block // timeout interval: max time allowed for peer without sending a block
blocksTimeout = 60 * time.Second blocksTimeout = 60 * time.Second
// // timeout interval: max time allowed for best peer to remain idle (not send new block after sync complete)
idleBestPeerTimeout = 120 * time.Second idleBestPeerTimeout = 120 * time.Second
// duration of suspension after peer fatal error during which peer is not allowed to reconnect
peerSuspensionInterval = 300 * time.Second
// status is logged every statusUpdateInterval
statusUpdateInterval = 3 * time.Second
) )
// config embedded in components, by default fall back to constants // blockpool config, values default to constants
// by default all resolved to local
type Config struct { type Config struct {
BlockHashesBatchSize int BlockHashesBatchSize int
BlockBatchSize int BlockBatchSize int
@ -48,26 +53,41 @@ type Config struct {
BlockHashesTimeout time.Duration BlockHashesTimeout time.Duration
BlocksTimeout time.Duration BlocksTimeout time.Duration
IdleBestPeerTimeout time.Duration IdleBestPeerTimeout time.Duration
PeerSuspensionInterval time.Duration
StatusUpdateInterval time.Duration
} }
// blockpool errors // blockpool errors
const ( const (
ErrInvalidBlock = iota ErrInvalidBlock = iota
ErrInvalidPoW ErrInvalidPoW
ErrUnrequestedBlock
ErrInsufficientChainInfo ErrInsufficientChainInfo
ErrIdleTooLong ErrIdleTooLong
ErrIncorrectTD
ErrUnrequestedBlock
) )
// error descriptions
var errorToString = map[int]string{ var errorToString = map[int]string{
ErrInvalidBlock: "Invalid block", ErrInvalidBlock: "Invalid block", // fatal
ErrInvalidPoW: "Invalid PoW", ErrInvalidPoW: "Invalid PoW", // fatal
ErrInsufficientChainInfo: "Insufficient chain info", // fatal
ErrIdleTooLong: "Idle too long", // fatal
ErrIncorrectTD: "Incorrect Total Difficulty", // fatal
ErrUnrequestedBlock: "Unrequested block", ErrUnrequestedBlock: "Unrequested block",
ErrInsufficientChainInfo: "Insufficient chain info",
ErrIdleTooLong: "Idle too long",
} }
// init initialises all your laundry // error severity
func severity(code int) ethlogger.LogLevel {
switch code {
case ErrUnrequestedBlock:
return ethlogger.WarnLevel
default:
return ethlogger.ErrorLevel
}
}
// init initialises the Config, zero values fall back to constants
func (self *Config) init() { func (self *Config) init() {
if self.BlockHashesBatchSize == 0 { if self.BlockHashesBatchSize == 0 {
self.BlockHashesBatchSize = blockHashesBatchSize self.BlockHashesBatchSize = blockHashesBatchSize
@ -96,6 +116,12 @@ func (self *Config) init() {
if self.IdleBestPeerTimeout == 0 { if self.IdleBestPeerTimeout == 0 {
self.IdleBestPeerTimeout = idleBestPeerTimeout self.IdleBestPeerTimeout = idleBestPeerTimeout
} }
if self.PeerSuspensionInterval == 0 {
self.PeerSuspensionInterval = peerSuspensionInterval
}
if self.StatusUpdateInterval == 0 {
self.StatusUpdateInterval = statusUpdateInterval
}
} }
// node is the basic unit of the internal model of block chain/tree in the blockpool // node is the basic unit of the internal model of block chain/tree in the blockpool
@ -122,31 +148,41 @@ type entry struct {
type BlockPool struct { type BlockPool struct {
Config *Config Config *Config
// the minimal interface with blockchain // the minimal interface with blockchain manager
hasBlock func(hash common.Hash) bool hasBlock func(hash common.Hash) bool // query if block is known
insertChain func(types.Blocks) error insertChain func(types.Blocks) error // add section to blockchain
verifyPoW func(pow.Block) bool verifyPoW func(pow.Block) bool // soft PoW verification
chainEvents *event.TypeMux // ethereum eventer for chainEvents
pool map[string]*entry tdSub event.Subscription // subscription to core.ChainHeadEvent
peers *peers td *big.Int // our own total difficulty
pool map[common.Hash]*entry // the actual blockpool
peers *peers // peers manager in peers.go
status *status // info about blockpool (UI interface) in status.go
lock sync.RWMutex lock sync.RWMutex
chainLock sync.RWMutex chainLock sync.RWMutex
// alloc-easy pool of hash slices // alloc-easy pool of hash slices
hashSlicePool chan []common.Hash hashSlicePool chan []common.Hash
status *status // waitgroup is used in tests to wait for result-critical routines
// as well as in determining idle / syncing status
quit chan bool wg sync.WaitGroup //
wg sync.WaitGroup quit chan bool // chan used for quitting parallel routines
running bool running bool //
} }
// public constructor // public constructor
// after blockpool returned, config can be set
// BlockPool.Start will call Config.init to set missing values
func New( func New(
hasBlock func(hash common.Hash) bool, hasBlock func(hash common.Hash) bool,
insertChain func(types.Blocks) error, insertChain func(types.Blocks) error,
verifyPoW func(pow.Block) bool, verifyPoW func(pow.Block) bool,
chainEvents *event.TypeMux,
td *big.Int,
) *BlockPool { ) *BlockPool {
return &BlockPool{ return &BlockPool{
@ -154,15 +190,8 @@ func New(
hasBlock: hasBlock, hasBlock: hasBlock,
insertChain: insertChain, insertChain: insertChain,
verifyPoW: verifyPoW, verifyPoW: verifyPoW,
} chainEvents: chainEvents,
} td: td,
func severity(code int) ethlogger.LogLevel {
switch code {
case ErrUnrequestedBlock:
return ethlogger.WarnLevel
default:
return ethlogger.ErrorLevel
} }
} }
@ -175,11 +204,13 @@ func (self *BlockPool) Start() {
return return
} }
// set missing values
self.Config.init() self.Config.init()
self.hashSlicePool = make(chan []common.Hash, 150) self.hashSlicePool = make(chan []common.Hash, 150)
self.status = newStatus() self.status = newStatus()
self.quit = make(chan bool) self.quit = make(chan bool)
self.pool = make(map[string]*entry) self.pool = make(map[common.Hash]*entry)
self.running = true self.running = true
self.peers = &peers{ self.peers = &peers{
@ -188,16 +219,37 @@ func (self *BlockPool) Start() {
Errors: errorToString, Errors: errorToString,
Level: severity, Level: severity,
}, },
peers: make(map[string]*peer), peers: make(map[string]*peer),
status: self.status, blacklist: make(map[string]time.Time),
bp: self, status: self.status,
bp: self,
} }
timer := time.NewTicker(3 * time.Second)
// subscribe and listen to core.ChainHeadEvent{} for uptodate TD
self.tdSub = self.chainEvents.Subscribe(core.ChainHeadEvent{})
// status update interval
timer := time.NewTicker(self.Config.StatusUpdateInterval)
go func() { go func() {
for { for {
select { select {
case <-self.quit: case <-self.quit:
return return
case event := <-self.tdSub.Chan():
if ev, ok := event.(core.ChainHeadEvent); ok {
td := ev.Block.Td
plog.DebugDetailf("td: %v", td)
self.setTD(td)
self.peers.lock.Lock()
if best := self.peers.best; best != nil {
if td.Cmp(best.td) >= 0 {
self.peers.best = nil
self.switchPeer(best, nil)
}
}
self.peers.lock.Unlock()
}
case <-timer.C: case <-timer.C:
plog.DebugDetailf("status:\n%v", self.Status()) plog.DebugDetailf("status:\n%v", self.Status())
} }
@ -218,6 +270,7 @@ func (self *BlockPool) Stop() {
plog.Infoln("Stopping...") plog.Infoln("Stopping...")
self.tdSub.Unsubscribe()
close(self.quit) close(self.quit)
self.lock.Lock() self.lock.Lock()
@ -255,9 +308,14 @@ func (self *BlockPool) Wait(t time.Duration) {
/* /*
AddPeer is called by the eth protocol instance running on the peer after AddPeer is called by the eth protocol instance running on the peer after
the status message has been received with total difficulty and current block hash the status message has been received with total difficulty and current block hash
Called a second time with the same peer id, it is used to update chain info for a peer. This is used when a new (mined) block message is received.
Called a second time with the same peer id, it is used to update chain info for a peer.
This is used when a new (mined) block message is received.
RemovePeer needs to be called when the peer disconnects. RemovePeer needs to be called when the peer disconnects.
Peer info is currently not persisted across disconnects (or sessions)
Peer info is currently not persisted across disconnects (or sessions) except for suspension
*/ */
func (self *BlockPool) AddPeer( func (self *BlockPool) AddPeer(
@ -267,7 +325,8 @@ func (self *BlockPool) AddPeer(
requestBlocks func([]common.Hash) error, requestBlocks func([]common.Hash) error,
peerError func(*errs.Error), peerError func(*errs.Error),
) (best bool) { ) (best bool, suspended bool) {
return self.peers.addPeer(td, currentBlockHash, peerId, requestBlockHashes, requestBlocks, peerError) return self.peers.addPeer(td, currentBlockHash, peerId, requestBlockHashes, requestBlocks, peerError)
} }
@ -281,12 +340,12 @@ AddBlockHashes
Entry point for eth protocol to add block hashes received via BlockHashesMsg Entry point for eth protocol to add block hashes received via BlockHashesMsg
only hashes from the best peer are handled Only hashes from the best peer are handled
initiates further hash requests until a known parent is reached (unless cancelled by a peerSwitch event, i.e., when a better peer becomes best peer) Initiates further hash requests until a known parent is reached (unless cancelled by a peerSwitch event, i.e., when a better peer becomes best peer)
launches all block request processes on each chain section Launches all block request processes on each chain section
the first argument is an iterator function. Using this block hashes are decoded from the rlp message payload on demand. As a result, AddBlockHashes needs to run synchronously for one peer since the message is discarded if the caller thread returns. The first argument is an iterator function. Using this block hashes are decoded from the rlp message payload on demand. As a result, AddBlockHashes needs to run synchronously for one peer since the message is discarded if the caller thread returns.
*/ */
func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId string) { func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId string) {
@ -297,7 +356,6 @@ func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId st
// bestpeer is still the best peer // bestpeer is still the best peer
self.wg.Add(1) self.wg.Add(1)
defer func() { self.wg.Done() }() defer func() { self.wg.Done() }()
self.status.lock.Lock() self.status.lock.Lock()
@ -322,11 +380,11 @@ func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId st
return return
} }
/* /*
when peer is promoted in switchPeer, a new header section process is launched When peer is promoted in switchPeer, a new header section process is launched.
as the head section skeleton is actually created here, it is signaled to the process Once the head section skeleton is actually created here, it is signaled to the process
so that it can quit so that it can quit.
in the special case that the node for parent of the head block is found in the blockpool In the special case that the node for parent of the head block is found in the blockpool
(with or without fetched block) (with or without fetched block), a singleton section containing only the head block node is created.
*/ */
headSection = true headSection = true
if entry := self.get(bestpeer.currentBlockHash); entry == nil { if entry := self.get(bestpeer.currentBlockHash); entry == nil {
@ -337,6 +395,7 @@ func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId st
block: bestpeer.currentBlock, block: bestpeer.currentBlock,
hashBy: peerId, hashBy: peerId,
blockBy: peerId, blockBy: peerId,
td: bestpeer.td,
} }
// nodes is a list of nodes in one section ordered top-bottom (old to young) // nodes is a list of nodes in one section ordered top-bottom (old to young)
nodes = append(nodes, node) nodes = append(nodes, node)
@ -344,7 +403,7 @@ func (self *BlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId st
} else { } else {
// otherwise set child section iff found node is the root of a section // otherwise set child section iff found node is the root of a section
// this is a possible scenario when a singleton head section was created // this is a possible scenario when a singleton head section was created
// on an earlier occasion this peer or another with the same block was best peer // on an earlier occasion when this peer or another with the same block was best peer
if entry.node == entry.section.bottom { if entry.node == entry.section.bottom {
child = entry.section child = entry.section
plog.DebugDetailf("AddBlockHashes: peer <%s>: connects to child section root %s", peerId, hex(bestpeer.currentBlockHash)) plog.DebugDetailf("AddBlockHashes: peer <%s>: connects to child section root %s", peerId, hex(bestpeer.currentBlockHash))
@ -375,7 +434,7 @@ LOOP:
default: default:
} }
// if we reach the blockchain we stop reading more // if we reach the blockchain we stop reading further blockhashes
if self.hasBlock(hash) { if self.hasBlock(hash) {
// check if known block connecting the downloaded chain to our blockchain // check if known block connecting the downloaded chain to our blockchain
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found block %s in the blockchain", peerId, hex(bestpeer.currentBlockHash), hex(hash)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found block %s in the blockchain", peerId, hex(bestpeer.currentBlockHash), hex(hash))
@ -412,10 +471,11 @@ LOOP:
// reached a known chain in the pool // reached a known chain in the pool
if entry.node == entry.section.bottom && n == 1 { if entry.node == entry.section.bottom && n == 1 {
/* /*
the first block hash received is an orphan in the pool The first block hash received is an orphan node in the pool
this also supports clients that (despite the spec) include <from> hash in their
This also supports clients that (despite the spec) include <from> hash in their
response to hashes request. Note that by providing <from> we can link sections response to hashes request. Note that by providing <from> we can link sections
without having to wait for the root block of the child section to arrive, so it allows for superior performance without having to wait for the root block of the child section to arrive, so it allows for superior performance.
*/ */
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found head block [%s] as root of connecting child section [%s] skipping", peerId, hex(bestpeer.currentBlockHash), hex(hash), sectionhex(entry.section)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) found head block [%s] as root of connecting child section [%s] skipping", peerId, hex(bestpeer.currentBlockHash), hex(hash), sectionhex(entry.section))
// record the entry's chain section as child section // record the entry's chain section as child section
@ -447,9 +507,8 @@ LOOP:
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): %v nodes in new section", peerId, hex(bestpeer.currentBlockHash), len(nodes)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): %v nodes in new section", peerId, hex(bestpeer.currentBlockHash), len(nodes))
/* /*
handle forks where connecting node is mid-section Handle forks where connecting node is mid-section by splitting section at fork.
by splitting section at fork No splitting needed if connecting node is head of a section.
no splitting needed if connecting node is head of a section
*/ */
if parent != nil && entry != nil && entry.node != parent.top && len(nodes) > 0 { if parent != nil && entry != nil && entry.node != parent.top && len(nodes) > 0 {
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): fork after %s", peerId, hex(bestpeer.currentBlockHash), hex(hash)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): fork after %s", peerId, hex(bestpeer.currentBlockHash), hex(hash))
@ -461,10 +520,7 @@ LOOP:
self.status.lock.Unlock() self.status.lock.Unlock()
} }
/* // If new section is created, link it to parent/child sections.
if new section is created, link it to parent/child sections
and launch section process fetching blocks and further hashes
*/
sec = self.linkSections(nodes, parent, child) sec = self.linkSections(nodes, parent, child)
if sec != nil { if sec != nil {
@ -477,11 +533,12 @@ LOOP:
self.chainLock.Unlock() self.chainLock.Unlock()
/* /*
if a blockpool node is reached (parent section is not nil), If a blockpool node is reached (parent section is not nil),
activate section (unless our peer is demoted by now). activate section (unless our peer is demoted by now).
this can be the bottom half of a newly split section in case of a fork. This can be the bottom half of a newly split section in case of a fork.
bestPeer is nil if we got here after our peer got demoted while processing. bestPeer is nil if we got here after our peer got demoted while processing.
in this case no activation should happen In this case no activation should happen
*/ */
if parent != nil && !peerswitch { if parent != nil && !peerswitch {
self.activateChain(parent, bestpeer, nil) self.activateChain(parent, bestpeer, nil)
@ -489,9 +546,8 @@ LOOP:
} }
/* /*
if a new section was created, If a new section was created, register section iff head section or no child known
register section iff head section or no child known Activate it with this peer.
activate it with this peer
*/ */
if sec != nil { if sec != nil {
// switch on section process (it is paused by switchC) // switch on section process (it is paused by switchC)
@ -502,9 +558,9 @@ LOOP:
bestpeer.lock.Unlock() bestpeer.lock.Unlock()
} }
/* /*
request next block hashes for parent section here. Request another batch of older block hashes for parent section here.
but only once, repeating only when bottom block arrives, But only once, repeating only when the section's root block arrives.
otherwise no way to check if it arrived Otherwise no way to check if it arrived.
*/ */
bestpeer.requestBlockHashes(sec.bottom.hash) bestpeer.requestBlockHashes(sec.bottom.hash)
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): start requesting blocks for section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s): start requesting blocks for section [%s]", peerId, hex(bestpeer.currentBlockHash), sectionhex(sec))
@ -515,7 +571,7 @@ LOOP:
} }
} }
// if we are processing peer's head section, signal it to headSection process that it is created // If we are processing peer's head section, signal it to headSection process that it is created.
if headSection { if headSection {
plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) head section registered on head section process", peerId, hex(bestpeer.currentBlockHash)) plog.DebugDetailf("AddBlockHashes: peer <%s> (head: %s) head section registered on head section process", peerId, hex(bestpeer.currentBlockHash))
@ -539,11 +595,13 @@ LOOP:
/* /*
AddBlock is the entry point for the eth protocol to call when blockMsg is received. AddBlock is the entry point for the eth protocol to call when blockMsg is received.
It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error It has a strict interpretation of the protocol in that if the block received has not been requested, it results in an error.
At the same time it is opportunistic in that if a requested block may be provided by any peer. At the same time it is opportunistic in that if a requested block may be provided by any peer.
The received block is checked for PoW. Only the first PoW-valid block for a hash is considered legit. The received block is checked for PoW. Only the first PoW-valid block for a hash is considered legit.
If the block received is the head block of the current best peer, signal it to the head section process
*/ */
func (self *BlockPool) AddBlock(block *types.Block, peerId string) { func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
hash := block.Hash() hash := block.Hash()
@ -564,7 +622,6 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
if sender.currentBlock == nil { if sender.currentBlock == nil {
plog.Debugf("AddBlock: add head block %s for peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash)) plog.Debugf("AddBlock: add head block %s for peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash))
sender.setChainInfoFromBlock(block) sender.setChainInfoFromBlock(block)
// sender.currentBlockC <- block
self.status.lock.Lock() self.status.lock.Lock()
self.status.values.BlockHashes++ self.status.values.BlockHashes++
@ -573,6 +630,7 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
self.status.lock.Unlock() self.status.lock.Unlock()
} else { } else {
plog.DebugDetailf("AddBlock: head block %s for peer <%s> (head: %s) already known", hex(hash), peerId, hex(sender.currentBlockHash)) plog.DebugDetailf("AddBlock: head block %s for peer <%s> (head: %s) already known", hex(hash), peerId, hex(sender.currentBlockHash))
// signal to head section process
sender.currentBlockC <- block sender.currentBlockC <- block
} }
} else { } else {
@ -591,7 +649,6 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
sender.lock.Unlock() sender.lock.Unlock()
if entry == nil { if entry == nil {
// penalise peer for sending what we have not asked
plog.DebugDetailf("AddBlock: unrequested block %s received from peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash)) plog.DebugDetailf("AddBlock: unrequested block %s received from peer <%s> (head: %s)", hex(hash), peerId, hex(sender.currentBlockHash))
sender.addError(ErrUnrequestedBlock, "%x", hash) sender.addError(ErrUnrequestedBlock, "%x", hash)
@ -609,7 +666,7 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
node.lock.Lock() node.lock.Lock()
defer node.lock.Unlock() defer node.lock.Unlock()
// check if block already present // check if block already received
if node.block != nil { if node.block != nil {
plog.DebugDetailf("AddBlock: block %s from peer <%s> (head: %s) already sent by <%s> ", hex(hash), peerId, hex(sender.currentBlockHash), node.blockBy) plog.DebugDetailf("AddBlock: block %s from peer <%s> (head: %s) already sent by <%s> ", hex(hash), peerId, hex(sender.currentBlockHash), node.blockBy)
return return
@ -645,9 +702,9 @@ func (self *BlockPool) AddBlock(block *types.Block, peerId string) {
} }
/* /*
iterates down a chain section by section activateChain iterates down a chain section by section.
activating section process on incomplete sections with peer It activates the section process on incomplete sections with peer.
relinking orphaned sections with their parent if root block (and its parent hash) is known) It relinks orphaned sections with their parent if root block (and its parent hash) is known.
*/ */
func (self *BlockPool) activateChain(sec *section, p *peer, connected map[string]*section) { func (self *BlockPool) activateChain(sec *section, p *peer, connected map[string]*section) {
@ -666,8 +723,8 @@ LOOP:
connected[sec.top.hash.Str()] = sec connected[sec.top.hash.Str()] = sec
} }
/* /*
we need to relink both complete and incomplete sections Need to relink both complete and incomplete sections
the latter could have been blockHashesRequestsComplete before being delinked from its parent An incomplete section could have been blockHashesRequestsComplete before being delinked from its parent.
*/ */
if parent == nil { if parent == nil {
if sec.bottom.block != nil { if sec.bottom.block != nil {
@ -682,7 +739,7 @@ LOOP:
} }
sec = parent sec = parent
// stop if peer got demoted // stop if peer got demoted or global quit
select { select {
case <-switchC: case <-switchC:
break LOOP break LOOP
@ -693,7 +750,19 @@ LOOP:
} }
} }
// must run in separate go routine, otherwise // check if block's actual TD (calculated after successful insertChain) is identical to TD advertised for peer's head block.
func (self *BlockPool) checkTD(nodes ...*node) {
for _, n := range nodes {
if n.td != nil {
plog.DebugDetailf("peer td %v =?= block td %v", n.td, n.block.Td)
if n.td.Cmp(n.block.Td) != 0 {
self.peers.peerError(n.blockBy, ErrIncorrectTD, "on block %x", n.hash)
}
}
}
}
// requestBlocks must run in separate go routine, otherwise
// switchpeer -> activateChain -> activate deadlocks on section process select and peers.lock // switchpeer -> activateChain -> activate deadlocks on section process select and peers.lock
func (self *BlockPool) requestBlocks(attempts int, hashes []common.Hash) { func (self *BlockPool) requestBlocks(attempts int, hashes []common.Hash) {
self.wg.Add(1) self.wg.Add(1)
@ -720,13 +789,26 @@ func (self *BlockPool) getChild(sec *section) *section {
func (self *BlockPool) get(hash common.Hash) *entry { func (self *BlockPool) get(hash common.Hash) *entry {
self.lock.RLock() self.lock.RLock()
defer self.lock.RUnlock() defer self.lock.RUnlock()
return self.pool[hash.Str()] return self.pool[hash]
} }
func (self *BlockPool) set(hash common.Hash, e *entry) { func (self *BlockPool) set(hash common.Hash, e *entry) {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
self.pool[hash.Str()] = e self.pool[hash] = e
}
// accessor and setter for total difficulty
func (self *BlockPool) getTD() *big.Int {
self.lock.RLock()
defer self.lock.RUnlock()
return self.td
}
func (self *BlockPool) setTD(td *big.Int) {
self.lock.Lock()
defer self.lock.Unlock()
self.td = td
} }
func (self *BlockPool) remove(sec *section) { func (self *BlockPool) remove(sec *section) {
@ -735,7 +817,7 @@ func (self *BlockPool) remove(sec *section) {
defer self.lock.Unlock() defer self.lock.Unlock()
for _, node := range sec.nodes { for _, node := range sec.nodes {
delete(self.pool, node.hash.Str()) delete(self.pool, node.hash)
} }
if sec.initialised && sec.poolRootIndex != 0 { if sec.initialised && sec.poolRootIndex != 0 {
self.status.lock.Lock() self.status.lock.Lock()
@ -744,6 +826,7 @@ func (self *BlockPool) remove(sec *section) {
} }
} }
// get/put for optimised allocation similar to sync.Pool
func (self *BlockPool) getHashSlice() (s []common.Hash) { func (self *BlockPool) getHashSlice() (s []common.Hash) {
select { select {
case s = <-self.hashSlicePool: case s = <-self.hashSlicePool:
@ -753,7 +836,6 @@ func (self *BlockPool) getHashSlice() (s []common.Hash) {
return return
} }
// Return returns a Client to the pool.
func (self *BlockPool) putHashSlice(s []common.Hash) { func (self *BlockPool) putHashSlice(s []common.Hash) {
if len(s) == self.Config.BlockBatchSize { if len(s) == self.Config.BlockBatchSize {
select { select {

124
blockpool/blockpool_test.go
View File

@ -5,10 +5,11 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/blockpool/test" "github.com/ethereum/go-ethereum/blockpool/test"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/common"
) )
// using the mock framework in blockpool_util_test
// we test various scenarios here
func TestPeerWithKnownBlock(t *testing.T) { func TestPeerWithKnownBlock(t *testing.T) {
test.LogInit() test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
@ -44,48 +45,6 @@ func TestPeerWithKnownParentBlock(t *testing.T) {
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain) blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
} }
func TestPeerPromotionByOptionalTdOnBlock(t *testing.T) {
test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(4)
peer0 := blockPoolTester.newPeer("peer0", 2, 2)
peer1 := blockPoolTester.newPeer("peer1", 1, 1)
peer2 := blockPoolTester.newPeer("peer2", 3, 4)
blockPool.Start()
// pool
peer0.AddPeer()
peer0.serveBlocks(1, 2)
best := peer1.AddPeer()
// this tests that peer1 is not promoted over peer0 yet
if best {
t.Errorf("peer1 (TD=1) should not be set as best")
}
best = peer2.AddPeer()
peer2.serveBlocks(3, 4)
peer2.serveBlockHashes(4, 3, 2, 1)
hashes := blockPoolTester.hashPool.IndexesToHashes([]int{2, 3})
peer1.waitBlocksRequests(3)
blockPool.AddBlock(&types.Block{
HeaderHash: common.Bytes(hashes[1]),
ParentHeaderHash: common.Bytes(hashes[0]),
Td: common.Big3,
}, "peer1")
blockPool.RemovePeer("peer2")
if blockPool.peers.best.id != "peer1" {
t.Errorf("peer1 (TD=3) should be set as best")
}
peer1.serveBlocks(0, 1, 2)
blockPool.Wait(waitTimeout)
blockPool.Stop()
blockPoolTester.refBlockChain[4] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
}
func TestSimpleChain(t *testing.T) { func TestSimpleChain(t *testing.T) {
test.LogInit() test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
@ -94,7 +53,7 @@ func TestSimpleChain(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 2) peer1 := blockPoolTester.newPeer("peer1", 2, 2)
peer1.AddPeer() peer1.AddPeer()
peer1.serveBlocks(1, 2) peer1.serveBlocks(1, 2)
go peer1.serveBlockHashes(2, 1, 0) go peer1.serveBlockHashes(2, 1, 0)
@ -114,7 +73,7 @@ func TestChainConnectingWithParentHash(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3) peer1 := blockPoolTester.newPeer("peer1", 3, 3)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(2, 3) go peer1.serveBlocks(2, 3)
go peer1.serveBlockHashes(3, 2, 1) go peer1.serveBlockHashes(3, 2, 1)
@ -134,7 +93,7 @@ func TestMultiSectionChain(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 5) peer1 := blockPoolTester.newPeer("peer1", 5, 5)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(4, 5) go peer1.serveBlocks(4, 5)
@ -156,14 +115,17 @@ func TestNewBlocksOnPartialChain(t *testing.T) {
blockPoolTester.initRefBlockChain(7) blockPoolTester.initRefBlockChain(7)
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 5) peer1 := blockPoolTester.newPeer("peer1", 5, 5)
blockPoolTester.tds = make(map[int]int)
blockPoolTester.tds[5] = 5
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(4, 5) // partially complete section go peer1.serveBlocks(4, 5) // partially complete section
go peer1.serveBlockHashes(5, 4, 3) go peer1.serveBlockHashes(5, 4, 3)
peer1.serveBlocks(3, 4) // partially complete section peer1.serveBlocks(3, 4) // partially complete section
// peer1 found new blocks // peer1 found new blocks
peer1.td = 2 peer1.td = 7
peer1.currentBlock = 7 peer1.currentBlock = 7
peer1.AddPeer() peer1.AddPeer()
peer1.sendBlocks(6, 7) peer1.sendBlocks(6, 7)
@ -172,7 +134,6 @@ func TestNewBlocksOnPartialChain(t *testing.T) {
go peer1.serveBlocks(5, 6) go peer1.serveBlocks(5, 6)
go peer1.serveBlockHashes(3, 2, 1) // tests that hash request from known chain root is remembered go peer1.serveBlockHashes(3, 2, 1) // tests that hash request from known chain root is remembered
peer1.serveBlocks(0, 1, 2) peer1.serveBlocks(0, 1, 2)
// blockPool.RemovePeer("peer1")
blockPool.Wait(waitTimeout) blockPool.Wait(waitTimeout)
blockPool.Stop() blockPool.Stop()
@ -188,16 +149,15 @@ func TestPeerSwitchUp(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 6) peer1 := blockPoolTester.newPeer("peer1", 6, 6)
peer2 := blockPoolTester.newPeer("peer2", 2, 7) peer2 := blockPoolTester.newPeer("peer2", 7, 7)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(5, 6) go peer1.serveBlocks(5, 6)
go peer1.serveBlockHashes(6, 5, 4, 3) // go peer1.serveBlockHashes(6, 5, 4, 3) //
peer1.serveBlocks(2, 3) // section partially complete, block 3 will be preserved after peer demoted peer1.serveBlocks(2, 3) // section partially complete, block 3 will be preserved after peer demoted
peer2.AddPeer() // peer2 is promoted as best peer, peer1 is demoted peer2.AddPeer() // peer2 is promoted as best peer, peer1 is demoted
go peer2.serveBlocks(6, 7) go peer2.serveBlocks(6, 7) //
// go peer2.serveBlockHashes(7, 6) //
go peer2.serveBlocks(4, 5) // tests that block request for earlier section is remembered go peer2.serveBlocks(4, 5) // tests that block request for earlier section is remembered
go peer1.serveBlocks(3, 4) // tests that connecting section by demoted peer is remembered and blocks are accepted from demoted peer go peer1.serveBlocks(3, 4) // tests that connecting section by demoted peer is remembered and blocks are accepted from demoted peer
go peer2.serveBlockHashes(3, 2, 1, 0) // tests that known chain section is activated, hash requests from 3 is remembered go peer2.serveBlockHashes(3, 2, 1, 0) // tests that known chain section is activated, hash requests from 3 is remembered
@ -216,8 +176,8 @@ func TestPeerSwitchDownOverlapSectionWithoutRootBlock(t *testing.T) {
blockPoolTester.initRefBlockChain(6) blockPoolTester.initRefBlockChain(6)
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 4) peer1 := blockPoolTester.newPeer("peer1", 4, 4)
peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer2 := blockPoolTester.newPeer("peer2", 6, 6)
peer2.AddPeer() peer2.AddPeer()
peer2.serveBlocks(5, 6) // partially complete, section will be preserved peer2.serveBlocks(5, 6) // partially complete, section will be preserved
@ -242,8 +202,8 @@ func TestPeerSwitchDownOverlapSectionWithRootBlock(t *testing.T) {
blockPoolTester.initRefBlockChain(6) blockPoolTester.initRefBlockChain(6)
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 4) peer1 := blockPoolTester.newPeer("peer1", 4, 4)
peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer2 := blockPoolTester.newPeer("peer2", 6, 6)
peer2.AddPeer() peer2.AddPeer()
peer2.serveBlocks(5, 6) // partially complete, section will be preserved peer2.serveBlocks(5, 6) // partially complete, section will be preserved
@ -269,8 +229,8 @@ func TestPeerSwitchDownDisjointSection(t *testing.T) {
blockPoolTester.initRefBlockChain(3) blockPoolTester.initRefBlockChain(3)
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3) peer1 := blockPoolTester.newPeer("peer1", 3, 3)
peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer2 := blockPoolTester.newPeer("peer2", 6, 6)
peer2.AddPeer() peer2.AddPeer()
peer2.serveBlocks(5, 6) // partially complete, section will be preserved peer2.serveBlocks(5, 6) // partially complete, section will be preserved
@ -297,8 +257,8 @@ func TestPeerSwitchBack(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 2, 11) peer1 := blockPoolTester.newPeer("peer1", 11, 11)
peer2 := blockPoolTester.newPeer("peer2", 1, 8) peer2 := blockPoolTester.newPeer("peer2", 8, 8)
peer2.AddPeer() peer2.AddPeer()
go peer2.serveBlocks(7, 8) go peer2.serveBlocks(7, 8)
@ -328,9 +288,10 @@ func TestForkSimple(t *testing.T) {
delete(blockPoolTester.refBlockChain, 6) delete(blockPoolTester.refBlockChain, 6)
blockPool.Start() blockPool.Start()
blockPoolTester.tds = make(map[int]int)
peer1 := blockPoolTester.newPeer("peer1", 1, 9) blockPoolTester.tds[6] = 10
peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer1 := blockPoolTester.newPeer("peer1", 9, 9)
peer2 := blockPoolTester.newPeer("peer2", 10, 6)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(8, 9) go peer1.serveBlocks(8, 9)
@ -363,9 +324,10 @@ func TestForkSwitchBackByNewBlocks(t *testing.T) {
delete(blockPoolTester.refBlockChain, 6) delete(blockPoolTester.refBlockChain, 6)
blockPool.Start() blockPool.Start()
blockPoolTester.tds = make(map[int]int)
peer1 := blockPoolTester.newPeer("peer1", 1, 9) blockPoolTester.tds[6] = 10
peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer1 := blockPoolTester.newPeer("peer1", 9, 9)
peer2 := blockPoolTester.newPeer("peer2", 10, 6)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(8, 9) // go peer1.serveBlocks(8, 9) //
@ -378,7 +340,7 @@ func TestForkSwitchBackByNewBlocks(t *testing.T) {
peer2.serveBlocks(1, 2, 3, 4, 5) // peer2.serveBlocks(1, 2, 3, 4, 5) //
// peer1 finds new blocks // peer1 finds new blocks
peer1.td = 3 peer1.td = 11
peer1.currentBlock = 11 peer1.currentBlock = 11
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(10, 11) go peer1.serveBlocks(10, 11)
@ -410,8 +372,14 @@ func TestForkSwitchBackByPeerSwitchBack(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 9) blockPoolTester.tds = make(map[int]int)
peer2 := blockPoolTester.newPeer("peer2", 2, 6) blockPoolTester.tds[6] = 10
blockPoolTester.tds = make(map[int]int)
blockPoolTester.tds[6] = 10
peer1 := blockPoolTester.newPeer("peer1", 9, 9)
peer2 := blockPoolTester.newPeer("peer2", 10, 6)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(8, 9) go peer1.serveBlocks(8, 9)
@ -448,14 +416,17 @@ func TestForkCompleteSectionSwitchBackByPeerSwitchBack(t *testing.T) {
blockPool.Start() blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 9) blockPoolTester.tds = make(map[int]int)
peer2 := blockPoolTester.newPeer("peer2", 2, 6) blockPoolTester.tds[6] = 10
peer1 := blockPoolTester.newPeer("peer1", 9, 9)
peer2 := blockPoolTester.newPeer("peer2", 10, 6)
peer1.AddPeer() peer1.AddPeer()
go peer1.serveBlocks(8, 9) go peer1.serveBlocks(8, 9)
go peer1.serveBlockHashes(9, 8, 7) go peer1.serveBlockHashes(9, 8, 7)
peer1.serveBlocks(3, 7, 8) // make sure this section is complete peer1.serveBlocks(3, 7, 8) // make sure this section is complete
time.Sleep(1 * time.Second) time.Sleep(1 * time.Second) //
go peer1.serveBlockHashes(7, 3, 2) // block 3/7 is section boundary go peer1.serveBlockHashes(7, 3, 2) // block 3/7 is section boundary
peer1.serveBlocks(2, 3) // partially complete sections block 2 missing peer1.serveBlocks(2, 3) // partially complete sections block 2 missing
peer2.AddPeer() // peer2.AddPeer() //
@ -463,8 +434,7 @@ func TestForkCompleteSectionSwitchBackByPeerSwitchBack(t *testing.T) {
go peer2.serveBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3 go peer2.serveBlockHashes(6, 5, 4, 3, 2) // peer2 forks on block 3
peer2.serveBlocks(2, 3, 4, 5) // block 2 still missing. peer2.serveBlocks(2, 3, 4, 5) // block 2 still missing.
blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer blockPool.RemovePeer("peer2") // peer2 disconnects, peer1 is promoted again as best peer
// peer1.serveBlockHashes(7, 3) // tests that hash request from fork root is remembered even though section process completed go peer1.serveBlockHashes(2, 1, 0) //
go peer1.serveBlockHashes(2, 1, 0) //
peer1.serveBlocks(0, 1, 2) peer1.serveBlocks(0, 1, 2)
blockPool.Wait(waitTimeout) blockPool.Wait(waitTimeout)

117
blockpool/blockpool_util_test.go
View File

@ -8,9 +8,10 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/blockpool/test" "github.com/ethereum/go-ethereum/blockpool/test"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/errs" "github.com/ethereum/go-ethereum/errs"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/pow" "github.com/ethereum/go-ethereum/pow"
) )
@ -38,6 +39,8 @@ type blockPoolTester struct {
blockChain blockChain blockChain blockChain
blockPool *BlockPool blockPool *BlockPool
t *testing.T t *testing.T
chainEvents *event.TypeMux
tds map[int]int
} }
func newTestBlockPool(t *testing.T) (hashPool *test.TestHashPool, blockPool *BlockPool, b *blockPoolTester) { func newTestBlockPool(t *testing.T) (hashPool *test.TestHashPool, blockPool *BlockPool, b *blockPoolTester) {
@ -48,8 +51,9 @@ func newTestBlockPool(t *testing.T) (hashPool *test.TestHashPool, blockPool *Blo
blockChain: make(blockChain), blockChain: make(blockChain),
refBlockChain: make(blockChain), refBlockChain: make(blockChain),
blocksRequestsMap: make(map[int]bool), blocksRequestsMap: make(map[int]bool),
chainEvents: &event.TypeMux{},
} }
b.blockPool = New(b.hasBlock, b.insertChain, b.verifyPoW) b.blockPool = New(b.hasBlock, b.insertChain, b.verifyPoW, b.chainEvents, common.Big0)
blockPool = b.blockPool blockPool = b.blockPool
blockPool.Config.BlockHashesRequestInterval = testBlockHashesRequestInterval blockPool.Config.BlockHashesRequestInterval = testBlockHashesRequestInterval
blockPool.Config.BlocksRequestInterval = testBlocksRequestInterval blockPool.Config.BlocksRequestInterval = testBlocksRequestInterval
@ -57,22 +61,24 @@ func newTestBlockPool(t *testing.T) (hashPool *test.TestHashPool, blockPool *Blo
} }
func (self *blockPoolTester) Errorf(format string, params ...interface{}) { func (self *blockPoolTester) Errorf(format string, params ...interface{}) {
fmt.Printf(format+"\n", params...) // fmt.Printf(format+"\n", params...)
self.t.Errorf(format, params...) self.t.Errorf(format, params...)
} }
// blockPoolTester implements the 3 callbacks needed by the blockPool: // blockPoolTester implements the 3 callbacks needed by the blockPool:
// hasBlock, insetChain, verifyPoW // hasBlock, insetChain, verifyPoW as well as provides the eventer
func (self *blockPoolTester) hasBlock(block []byte) (ok bool) { // to subscribe to head insertions
func (self *blockPoolTester) hasBlock(block common.Hash) (ok bool) {
self.lock.RLock() self.lock.RLock()
defer self.lock.RUnlock() defer self.lock.RUnlock()
indexes := self.hashPool.HashesToIndexes([][]byte{block}) indexes := self.hashPool.HashesToIndexes([]common.Hash{block})
i := indexes[0] i := indexes[0]
_, ok = self.blockChain[i] _, ok = self.blockChain[i]
fmt.Printf("has block %v (%x...): %v\n", i, block[0:4], ok) // fmt.Printf("has block %v (%x...): %v\n", i, block[0:4], ok)
return return
} }
// mock insertChain relies on refBlockChain to determine block validity
func (self *blockPoolTester) insertChain(blocks types.Blocks) error { func (self *blockPoolTester) insertChain(blocks types.Blocks) error {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
@ -80,13 +86,21 @@ func (self *blockPoolTester) insertChain(blocks types.Blocks) error {
var children, refChildren []int var children, refChildren []int
var ok bool var ok bool
for _, block := range blocks { for _, block := range blocks {
child = self.hashPool.HashesToIndexes([][]byte{block.Hash()})[0] child = self.hashPool.HashesToIndexes([]common.Hash{block.Hash()})[0]
var td int
if self.tds != nil {
td, ok = self.tds[child]
}
if !ok {
td = child
}
block.Td = big.NewInt(int64(td))
_, ok = self.blockChain[child] _, ok = self.blockChain[child]
if ok { if ok {
fmt.Printf("block %v already in blockchain\n", child) // fmt.Printf("block %v already in blockchain\n", child)
continue // already in chain continue // already in chain
} }
parent = self.hashPool.HashesToIndexes([][]byte{block.ParentHeaderHash})[0] parent = self.hashPool.HashesToIndexes([]common.Hash{block.ParentHeaderHash})[0]
children, ok = self.blockChain[parent] children, ok = self.blockChain[parent]
if !ok { if !ok {
return fmt.Errorf("parent %v not in blockchain ", parent) return fmt.Errorf("parent %v not in blockchain ", parent)
@ -108,7 +122,6 @@ func (self *blockPoolTester) insertChain(blocks types.Blocks) error {
} }
if ok { if ok {
// accept any blocks if parent not in refBlockChain // accept any blocks if parent not in refBlockChain
fmt.Errorf("blockchain insert %v -> %v\n", parent, child)
self.blockChain[parent] = append(children, child) self.blockChain[parent] = append(children, child)
self.blockChain[child] = nil self.blockChain[child] = nil
} }
@ -116,6 +129,7 @@ func (self *blockPoolTester) insertChain(blocks types.Blocks) error {
return nil return nil
} }
// mock soft block validation always succeeds
func (self *blockPoolTester) verifyPoW(pblock pow.Block) bool { func (self *blockPoolTester) verifyPoW(pblock pow.Block) bool {
return true return true
} }
@ -124,12 +138,12 @@ func (self *blockPoolTester) verifyPoW(pblock pow.Block) bool {
func (self *blockPoolTester) checkBlockChain(blockChain map[int][]int) { func (self *blockPoolTester) checkBlockChain(blockChain map[int][]int) {
self.lock.RLock() self.lock.RLock()
defer self.lock.RUnlock() defer self.lock.RUnlock()
for k, v := range self.blockChain { // for k, v := range self.blockChain {
fmt.Printf("got: %v -> %v\n", k, v) // fmt.Printf("got: %v -> %v\n", k, v)
} // }
for k, v := range blockChain { // for k, v := range blockChain {
fmt.Printf("expected: %v -> %v\n", k, v) // fmt.Printf("expected: %v -> %v\n", k, v)
} // }
if len(blockChain) != len(self.blockChain) { if len(blockChain) != len(self.blockChain) {
self.Errorf("blockchain incorrect (zlength differ)") self.Errorf("blockchain incorrect (zlength differ)")
} }
@ -141,24 +155,24 @@ func (self *blockPoolTester) checkBlockChain(blockChain map[int][]int) {
} }
} }
//
// peerTester provides the peer callbacks for the blockPool // peerTester provides the peer callbacks for the blockPool
// it registers actual callbacks so that the result can be compared to desired behaviour // it registers actual callbacks so that the result can be compared to desired behaviour
// provides helper functions to mock the protocol calls to the blockPool // provides helper functions to mock the protocol calls to the blockPool
type peerTester struct { type peerTester struct {
// containers to record request and error callbacks
blockHashesRequests []int blockHashesRequests []int
blocksRequests [][]int blocksRequests [][]int
blocksRequestsMap map[int]bool blocksRequestsMap map[int]bool
peerErrors []int peerErrors []int
blockPool *BlockPool
hashPool *test.TestHashPool blockPool *BlockPool
lock sync.RWMutex hashPool *test.TestHashPool
bt *blockPoolTester lock sync.RWMutex
id string bt *blockPoolTester
td int id string
currentBlock int td int
t *testing.T currentBlock int
t *testing.T
} }
// peerTester constructor takes hashPool and blockPool from the blockPoolTester // peerTester constructor takes hashPool and blockPool from the blockPoolTester
@ -176,7 +190,7 @@ func (self *blockPoolTester) newPeer(id string, td int, cb int) *peerTester {
} }
func (self *peerTester) Errorf(format string, params ...interface{}) { func (self *peerTester) Errorf(format string, params ...interface{}) {
fmt.Printf(format+"\n", params...) // fmt.Printf(format+"\n", params...)
self.t.Errorf(format, params...) self.t.Errorf(format, params...)
} }
@ -211,6 +225,7 @@ func (self *peerTester) checkBlockHashesRequests(blocksHashesRequests ...int) {
// waiter function used by peer.serveBlocks // waiter function used by peer.serveBlocks
// blocking until requests appear // blocking until requests appear
// this mocks proper wire protocol behaviour
// since block requests are sent to any random peers // since block requests are sent to any random peers
// block request map is shared between peers // block request map is shared between peers
// times out after waitTimeout // times out after waitTimeout
@ -220,7 +235,7 @@ func (self *peerTester) waitBlocksRequests(blocksRequest ...int) {
for { for {
self.lock.RLock() self.lock.RLock()
r := self.blocksRequestsMap r := self.blocksRequestsMap
fmt.Printf("[%s] blocks request check %v (%v)\n", self.id, rr, r) // fmt.Printf("[%s] blocks request check %v (%v)\n", self.id, rr, r)
i := 0 i := 0
for i = 0; i < len(rr); i++ { for i = 0; i < len(rr); i++ {
_, ok := r[rr[i]] _, ok := r[rr[i]]
@ -243,6 +258,7 @@ func (self *peerTester) waitBlocksRequests(blocksRequest ...int) {
// waiter function used by peer.serveBlockHashes // waiter function used by peer.serveBlockHashes
// blocking until requests appear // blocking until requests appear
// this mocks proper wire protocol behaviour
// times out after a period // times out after a period
func (self *peerTester) waitBlockHashesRequests(blocksHashesRequest int) { func (self *peerTester) waitBlockHashesRequests(blocksHashesRequest int) {
timeout := time.After(waitTimeout) timeout := time.After(waitTimeout)
@ -251,7 +267,7 @@ func (self *peerTester) waitBlockHashesRequests(blocksHashesRequest int) {
self.lock.RLock() self.lock.RLock()
r := self.blockHashesRequests r := self.blockHashesRequests
self.lock.RUnlock() self.lock.RUnlock()
fmt.Printf("[%s] block hash request check %v (%v)\n", self.id, rr, r) // fmt.Printf("[%s] block hash request check %v (%v)\n", self.id, rr, r)
for ; i < len(r); i++ { for ; i < len(r); i++ {
if rr == r[i] { if rr == r[i] {
return return
@ -274,24 +290,26 @@ func (self *blockPoolTester) initRefBlockChain(n int) {
// peerTester functions that mimic protocol calls to the blockpool // peerTester functions that mimic protocol calls to the blockpool
// registers the peer with the blockPool // registers the peer with the blockPool
func (self *peerTester) AddPeer() bool { func (self *peerTester) AddPeer() (best bool) {
hash := self.hashPool.IndexesToHashes([]int{self.currentBlock})[0] hash := self.hashPool.IndexesToHashes([]int{self.currentBlock})[0]
return self.blockPool.AddPeer(big.NewInt(int64(self.td)), hash, self.id, self.requestBlockHashes, self.requestBlocks, self.peerError) best, _ = self.blockPool.AddPeer(big.NewInt(int64(self.td)), hash, self.id, self.requestBlockHashes, self.requestBlocks, self.peerError)
return
} }
// peer sends blockhashes if and when gets a request // peer sends blockhashes if and when gets a request
func (self *peerTester) serveBlockHashes(indexes ...int) { func (self *peerTester) serveBlockHashes(indexes ...int) {
fmt.Printf("ready to serve block hashes %v\n", indexes) // fmt.Printf("ready to serve block hashes %v\n", indexes)
self.waitBlockHashesRequests(indexes[0]) self.waitBlockHashesRequests(indexes[0])
self.sendBlockHashes(indexes...) self.sendBlockHashes(indexes...)
} }
// peer sends blockhashes not waiting for request
func (self *peerTester) sendBlockHashes(indexes ...int) { func (self *peerTester) sendBlockHashes(indexes ...int) {
fmt.Printf("adding block hashes %v\n", indexes) // fmt.Printf("adding block hashes %v\n", indexes)
hashes := self.hashPool.IndexesToHashes(indexes) hashes := self.hashPool.IndexesToHashes(indexes)
i := 1 i := 1
next := func() (hash []byte, ok bool) { next := func() (hash common.Hash, ok bool) {
if i < len(hashes) { if i < len(hashes) {
hash = hashes[i] hash = hashes[i]
ok = true ok = true
@ -303,28 +321,30 @@ func (self *peerTester) sendBlockHashes(indexes ...int) {
} }
// peer sends blocks if and when there is a request // peer sends blocks if and when there is a request
// (in the shared request store, not necessarily to a person) // (in the shared request store, not necessarily to a specific peer)
func (self *peerTester) serveBlocks(indexes ...int) { func (self *peerTester) serveBlocks(indexes ...int) {
fmt.Printf("ready to serve blocks %v\n", indexes[1:]) // fmt.Printf("ready to serve blocks %v\n", indexes[1:])
self.waitBlocksRequests(indexes[1:]...) self.waitBlocksRequests(indexes[1:]...)
self.sendBlocks(indexes...) self.sendBlocks(indexes...)
} }
// peer sends blocks not waiting for request
func (self *peerTester) sendBlocks(indexes ...int) { func (self *peerTester) sendBlocks(indexes ...int) {
fmt.Printf("adding blocks %v \n", indexes) // fmt.Printf("adding blocks %v \n", indexes)
hashes := self.hashPool.IndexesToHashes(indexes) hashes := self.hashPool.IndexesToHashes(indexes)
for i := 1; i < len(hashes); i++ { for i := 1; i < len(hashes); i++ {
fmt.Printf("adding block %v %x\n", indexes[i], hashes[i][:4]) // fmt.Printf("adding block %v %x\n", indexes[i], hashes[i][:4])
self.blockPool.AddBlock(&types.Block{HeaderHash: common.Bytes(hashes[i]), ParentHeaderHash: common.Bytes(hashes[i-1])}, self.id) self.blockPool.AddBlock(&types.Block{HeaderHash: hashes[i], ParentHeaderHash: hashes[i-1]}, self.id)
} }
} }
// peer callbacks // the 3 mock peer callbacks
// -1 is special: not found (a hash never seen)
// records block hashes requests by the blockPool // records block hashes requests by the blockPool
func (self *peerTester) requestBlockHashes(hash []byte) error { // -1 is special: not found (a hash never seen)
indexes := self.hashPool.HashesToIndexes([][]byte{hash}) func (self *peerTester) requestBlockHashes(hash common.Hash) error {
fmt.Printf("[%s] block hash request %v %x\n", self.id, indexes[0], hash[:4]) indexes := self.hashPool.HashesToIndexes([]common.Hash{hash})
// fmt.Printf("[%s] block hash request %v %x\n", self.id, indexes[0], hash[:4])
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
self.blockHashesRequests = append(self.blockHashesRequests, indexes[0]) self.blockHashesRequests = append(self.blockHashesRequests, indexes[0])
@ -332,9 +352,9 @@ func (self *peerTester) requestBlockHashes(hash []byte) error {
} }
// records block requests by the blockPool // records block requests by the blockPool
func (self *peerTester) requestBlocks(hashes [][]byte) error { func (self *peerTester) requestBlocks(hashes []common.Hash) error {
indexes := self.hashPool.HashesToIndexes(hashes) indexes := self.hashPool.HashesToIndexes(hashes)
fmt.Printf("blocks request %v %x...\n", indexes, hashes[0][:4]) // fmt.Printf("blocks request %v %x...\n", indexes, hashes[0][:4])
self.bt.reqlock.Lock() self.bt.reqlock.Lock()
defer self.bt.reqlock.Unlock() defer self.bt.reqlock.Unlock()
self.blocksRequests = append(self.blocksRequests, indexes) self.blocksRequests = append(self.blocksRequests, indexes)
@ -347,4 +367,7 @@ func (self *peerTester) requestBlocks(hashes [][]byte) error {
// records the error codes of all the peerErrors found the blockPool // records the error codes of all the peerErrors found the blockPool
func (self *peerTester) peerError(err *errs.Error) { func (self *peerTester) peerError(err *errs.Error) {
self.peerErrors = append(self.peerErrors, err.Code) self.peerErrors = append(self.peerErrors, err.Code)
if err.Fatal() {
self.blockPool.RemovePeer(self.id)
}
} }

11
blockpool/config_test.go
View File

@ -5,11 +5,12 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/blockpool/test" "github.com/ethereum/go-ethereum/blockpool/test"
"github.com/ethereum/go-ethereum/event"
) )
func TestBlockPoolConfig(t *testing.T) { func TestBlockPoolConfig(t *testing.T) {
test.LogInit() test.LogInit()
blockPool := &BlockPool{Config: &Config{}} blockPool := &BlockPool{Config: &Config{}, chainEvents: &event.TypeMux{}}
blockPool.Start() blockPool.Start()
c := blockPool.Config c := blockPool.Config
test.CheckInt("BlockHashesBatchSize", c.BlockHashesBatchSize, blockHashesBatchSize, t) test.CheckInt("BlockHashesBatchSize", c.BlockHashesBatchSize, blockHashesBatchSize, t)
@ -21,12 +22,14 @@ func TestBlockPoolConfig(t *testing.T) {
test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, blockHashesTimeout, t) test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, blockHashesTimeout, t)
test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t) test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t)
test.CheckDuration("IdleBestPeerTimeout", c.IdleBestPeerTimeout, idleBestPeerTimeout, t) test.CheckDuration("IdleBestPeerTimeout", c.IdleBestPeerTimeout, idleBestPeerTimeout, t)
test.CheckDuration("PeerSuspensionInterval", c.PeerSuspensionInterval, peerSuspensionInterval, t)
test.CheckDuration("StatusUpdateInterval", c.StatusUpdateInterval, statusUpdateInterval, t)
} }
func TestBlockPoolOverrideConfig(t *testing.T) { func TestBlockPoolOverrideConfig(t *testing.T) {
test.LogInit() test.LogInit()
blockPool := &BlockPool{Config: &Config{}} blockPool := &BlockPool{Config: &Config{}, chainEvents: &event.TypeMux{}}
c := &Config{128, 32, 1, 0, 300 * time.Millisecond, 100 * time.Millisecond, 90 * time.Second, 0, 30 * time.Second} c := &Config{128, 32, 1, 0, 300 * time.Millisecond, 100 * time.Millisecond, 90 * time.Second, 0, 30 * time.Second, 30 * time.Second, 4 * time.Second}
blockPool.Config = c blockPool.Config = c
blockPool.Start() blockPool.Start()
@ -39,4 +42,6 @@ func TestBlockPoolOverrideConfig(t *testing.T) {
test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, 90*time.Second, t) test.CheckDuration("BlockHashesTimeout", c.BlockHashesTimeout, 90*time.Second, t)
test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t) test.CheckDuration("BlocksTimeout", c.BlocksTimeout, blocksTimeout, t)
test.CheckDuration("IdleBestPeerTimeout", c.IdleBestPeerTimeout, 30*time.Second, t) test.CheckDuration("IdleBestPeerTimeout", c.IdleBestPeerTimeout, 30*time.Second, t)
test.CheckDuration("PeerSuspensionInterval", c.PeerSuspensionInterval, 30*time.Second, t)
test.CheckDuration("StatusUpdateInterval", c.StatusUpdateInterval, 4*time.Second, t)
} }

61
blockpool/errors_test.go
View File

@ -5,6 +5,7 @@ import (
"time" "time"
"github.com/ethereum/go-ethereum/blockpool/test" "github.com/ethereum/go-ethereum/blockpool/test"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/pow" "github.com/ethereum/go-ethereum/pow"
) )
@ -45,7 +46,7 @@ func TestVerifyPoW(t *testing.T) {
first := false first := false
blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool { blockPoolTester.blockPool.verifyPoW = func(b pow.Block) bool {
bb, _ := b.(*types.Block) bb, _ := b.(*types.Block)
indexes := blockPoolTester.hashPool.HashesToIndexes([][]byte{bb.Hash()}) indexes := blockPoolTester.hashPool.HashesToIndexes([]common.Hash{bb.Hash()})
if indexes[0] == 2 && !first { if indexes[0] == 2 && !first {
first = true first = true
return false return false
@ -92,7 +93,6 @@ func TestUnrequestedBlock(t *testing.T) {
peer1.AddPeer() peer1.AddPeer()
peer1.sendBlocks(1, 2) peer1.sendBlocks(1, 2)
// blockPool.Wait(waitTimeout)
blockPool.Stop() blockPool.Stop()
if len(peer1.peerErrors) == 1 { if len(peer1.peerErrors) == 1 {
if peer1.peerErrors[0] != ErrUnrequestedBlock { if peer1.peerErrors[0] != ErrUnrequestedBlock {
@ -122,3 +122,60 @@ func TestErrInsufficientChainInfo(t *testing.T) {
t.Errorf("expected %v error, got %v", ErrInsufficientChainInfo, peer1.peerErrors) t.Errorf("expected %v error, got %v", ErrInsufficientChainInfo, peer1.peerErrors)
} }
} }
func TestIncorrectTD(t *testing.T) {
test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(3)
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer()
go peer1.serveBlocks(2, 3)
go peer1.serveBlockHashes(3, 2, 1, 0)
peer1.serveBlocks(0, 1, 2)
blockPool.Wait(waitTimeout)
blockPool.Stop()
blockPoolTester.refBlockChain[3] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
if len(peer1.peerErrors) == 1 {
if peer1.peerErrors[0] != ErrIncorrectTD {
t.Errorf("wrong error, got %v, expected %v", peer1.peerErrors[0], ErrIncorrectTD)
}
} else {
t.Errorf("expected %v error, got %v", ErrIncorrectTD, peer1.peerErrors)
}
}
func TestPeerSuspension(t *testing.T) {
test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPool.Config.PeerSuspensionInterval = 100 * time.Millisecond
blockPool.Start()
peer1 := blockPoolTester.newPeer("peer1", 1, 3)
peer1.AddPeer()
blockPool.peers.peerError("peer1", 0, "")
bestpeer, _ := blockPool.peers.getPeer("peer1")
if bestpeer != nil {
t.Errorf("peer1 not removed on error")
}
peer1.AddPeer()
bestpeer, _ = blockPool.peers.getPeer("peer1")
if bestpeer != nil {
t.Errorf("peer1 not removed on reconnect")
}
time.Sleep(100 * time.Millisecond)
peer1.AddPeer()
bestpeer, _ = blockPool.peers.getPeer("peer1")
if bestpeer == nil {
t.Errorf("peer1 not connected after PeerSuspensionInterval")
}
// blockPool.Wait(waitTimeout)
blockPool.Stop()
}

90
blockpool/peers.go
View File

@ -12,6 +12,7 @@ import (
"github.com/ethereum/go-ethereum/errs" "github.com/ethereum/go-ethereum/errs"
) )
// the blockpool's model of a peer
type peer struct { type peer struct {
lock sync.RWMutex lock sync.RWMutex
@ -47,6 +48,8 @@ type peer struct {
blocksRequestTimer <-chan time.Time blocksRequestTimer <-chan time.Time
suicideC <-chan time.Time suicideC <-chan time.Time
addToBlacklist func(id string)
idle bool idle bool
} }
@ -55,11 +58,12 @@ type peer struct {
type peers struct { type peers struct {
lock sync.RWMutex lock sync.RWMutex
bp *BlockPool bp *BlockPool
errors *errs.Errors errors *errs.Errors
peers map[string]*peer peers map[string]*peer
best *peer best *peer
status *status status *status
blacklist map[string]time.Time
} }
// peer constructor // peer constructor
@ -84,26 +88,48 @@ func (self *peers) newPeer(
headSectionC: make(chan *section), headSectionC: make(chan *section),
bp: self.bp, bp: self.bp,
idle: true, idle: true,
addToBlacklist: self.addToBlacklist,
} }
// at creation the peer is recorded in the peer pool // at creation the peer is recorded in the peer pool
self.peers[id] = p self.peers[id] = p
return return
} }
// dispatches an error to a peer if still connected // dispatches an error to a peer if still connected, adds it to the blacklist
func (self *peers) peerError(id string, code int, format string, params ...interface{}) { func (self *peers) peerError(id string, code int, format string, params ...interface{}) {
self.lock.RLock() self.lock.RLock()
defer self.lock.RUnlock()
peer, ok := self.peers[id] peer, ok := self.peers[id]
self.lock.RUnlock()
if ok { if ok {
peer.addError(code, format, params) peer.addError(code, format, params)
} }
// blacklisting comes here self.addToBlacklist(id)
}
// record time of offence in blacklist to implement suspension for PeerSuspensionInterval
func (self *peers) addToBlacklist(id string) {
self.lock.Lock()
defer self.lock.Unlock()
self.blacklist[id] = time.Now()
}
// suspended checks if peer is still suspended
func (self *peers) suspended(id string) (s bool) {
self.lock.Lock()
defer self.lock.Unlock()
if suspendedAt, ok := self.blacklist[id]; ok {
if s = suspendedAt.Add(self.bp.Config.PeerSuspensionInterval).After(time.Now()); !s {
// no longer suspended, delete entry
delete(self.blacklist, id)
}
}
return
} }
func (self *peer) addError(code int, format string, params ...interface{}) { func (self *peer) addError(code int, format string, params ...interface{}) {
err := self.errors.New(code, format, params...) err := self.errors.New(code, format, params...)
self.peerError(err) self.peerError(err)
self.addToBlacklist(self.id)
} }
func (self *peer) setChainInfo(td *big.Int, c common.Hash) { func (self *peer) setChainInfo(td *big.Int, c common.Hash) {
@ -138,8 +164,8 @@ func (self *peer) setChainInfoFromBlock(block *types.Block) {
}() }()
} }
// distribute block request among known peers
func (self *peers) requestBlocks(attempts int, hashes []common.Hash) { func (self *peers) requestBlocks(attempts int, hashes []common.Hash) {
// distribute block request among known peers
self.lock.RLock() self.lock.RLock()
defer self.lock.RUnlock() defer self.lock.RUnlock()
peerCount := len(self.peers) peerCount := len(self.peers)
@ -174,7 +200,9 @@ func (self *peers) requestBlocks(attempts int, hashes []common.Hash) {
} }
// addPeer implements the logic for blockpool.AddPeer // addPeer implements the logic for blockpool.AddPeer
// returns true iff peer is promoted as best peer in the pool // returns 2 bool values
// 1. true iff peer is promoted as best peer in the pool
// 2. true iff peer is still suspended
func (self *peers) addPeer( func (self *peers) addPeer(
td *big.Int, td *big.Int,
currentBlockHash common.Hash, currentBlockHash common.Hash,
@ -182,16 +210,23 @@ func (self *peers) addPeer(
requestBlockHashes func(common.Hash) error, requestBlockHashes func(common.Hash) error,
requestBlocks func([]common.Hash) error, requestBlocks func([]common.Hash) error,
peerError func(*errs.Error), peerError func(*errs.Error),
) (best bool) { ) (best bool, suspended bool) {
var previousBlockHash common.Hash var previousBlockHash common.Hash
if self.suspended(id) {
suspended = true
return
}
self.lock.Lock() self.lock.Lock()
p, found := self.peers[id] p, found := self.peers[id]
if found { if found {
// when called on an already connected peer, it means a newBlockMsg is received
// peer head info is updated
if p.currentBlockHash != currentBlockHash { if p.currentBlockHash != currentBlockHash {
previousBlockHash = p.currentBlockHash previousBlockHash = p.currentBlockHash
plog.Debugf("addPeer: Update peer <%s> with td %v and current block %s (was %v)", id, td, hex(currentBlockHash), hex(previousBlockHash)) plog.Debugf("addPeer: Update peer <%s> with td %v and current block %s (was %v)", id, td, hex(currentBlockHash), hex(previousBlockHash))
p.setChainInfo(td, currentBlockHash) p.setChainInfo(td, currentBlockHash)
self.status.lock.Lock() self.status.lock.Lock()
self.status.values.NewBlocks++ self.status.values.NewBlocks++
self.status.lock.Unlock() self.status.lock.Unlock()
@ -209,11 +244,11 @@ func (self *peers) addPeer(
} }
self.lock.Unlock() self.lock.Unlock()
// check peer current head // check if peer's current head block is known
if self.bp.hasBlock(currentBlockHash) { if self.bp.hasBlock(currentBlockHash) {
// peer not ahead // peer not ahead
plog.Debugf("addPeer: peer <%v> with td %v and current block %s is behind", id, td, hex(currentBlockHash)) plog.Debugf("addPeer: peer <%v> with td %v and current block %s is behind", id, td, hex(currentBlockHash))
return false return false, false
} }
if self.best == p { if self.best == p {
@ -229,7 +264,8 @@ func (self *peers) addPeer(
} }
best = true best = true
} else { } else {
currentTD := common.Big0 // baseline is our own TD
currentTD := self.bp.getTD()
if self.best != nil { if self.best != nil {
currentTD = self.best.td currentTD = self.best.td
} }
@ -237,7 +273,7 @@ func (self *peers) addPeer(
self.status.lock.Lock() self.status.lock.Lock()
self.status.bestPeers[p.id]++ self.status.bestPeers[p.id]++
self.status.lock.Unlock() self.status.lock.Unlock()
plog.Debugf("addPeer: peer <%v> promoted best peer", id) plog.Debugf("addPeer: peer <%v> (td: %v > current td %v) promoted best peer", id, td, currentTD)
self.bp.switchPeer(self.best, p) self.bp.switchPeer(self.best, p)
self.best = p self.best = p
best = true best = true
@ -257,13 +293,13 @@ func (self *peers) removePeer(id string) {
} }
delete(self.peers, id) delete(self.peers, id)
plog.Debugf("addPeer: remove peer <%v>", id) plog.Debugf("addPeer: remove peer <%v> (td: %v)", id, p.td)
// if current best peer is removed, need to find a better one // if current best peer is removed, need to find a better one
if self.best == p { if self.best == p {
var newp *peer var newp *peer
// FIXME: own TD // only peers that are ahead of us are considered
max := common.Big0 max := self.bp.getTD()
// peer with the highest self-acclaimed TD is chosen // peer with the highest self-acclaimed TD is chosen
for _, pp := range self.peers { for _, pp := range self.peers {
if pp.td.Cmp(max) > 0 { if pp.td.Cmp(max) > 0 {
@ -275,7 +311,7 @@ func (self *peers) removePeer(id string) {
self.status.lock.Lock() self.status.lock.Lock()
self.status.bestPeers[p.id]++ self.status.bestPeers[p.id]++
self.status.lock.Unlock() self.status.lock.Unlock()
plog.Debugf("addPeer: peer <%v> with td %v promoted best peer", newp.id, newp.td) plog.Debugf("addPeer: peer <%v> (td: %v) promoted best peer", newp.id, newp.td)
} else { } else {
plog.Warnln("addPeer: no suitable peers found") plog.Warnln("addPeer: no suitable peers found")
} }
@ -288,6 +324,7 @@ func (self *peers) removePeer(id string) {
func (self *BlockPool) switchPeer(oldp, newp *peer) { func (self *BlockPool) switchPeer(oldp, newp *peer) {
// first quit AddBlockHashes, requestHeadSection and activateChain // first quit AddBlockHashes, requestHeadSection and activateChain
// by closing the old peer's switchC channel
if oldp != nil { if oldp != nil {
plog.DebugDetailf("<%s> quit peer processes", oldp.id) plog.DebugDetailf("<%s> quit peer processes", oldp.id)
close(oldp.switchC) close(oldp.switchC)
@ -340,11 +377,12 @@ func (self *BlockPool) switchPeer(oldp, newp *peer) {
// newp activating section process changes the quit channel for this reason // newp activating section process changes the quit channel for this reason
if oldp != nil { if oldp != nil {
plog.DebugDetailf("<%s> quit section processes", oldp.id) plog.DebugDetailf("<%s> quit section processes", oldp.id)
//
close(oldp.idleC) close(oldp.idleC)
} }
} }
// getPeer looks up peer by id, returns peer and a bool value
// that is true iff peer is current best peer
func (self *peers) getPeer(id string) (p *peer, best bool) { func (self *peers) getPeer(id string) (p *peer, best bool) {
self.lock.RLock() self.lock.RLock()
defer self.lock.RUnlock() defer self.lock.RUnlock()
@ -355,6 +393,8 @@ func (self *peers) getPeer(id string) (p *peer, best bool) {
return return
} }
// head section process
func (self *peer) handleSection(sec *section) { func (self *peer) handleSection(sec *section) {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
@ -426,6 +466,11 @@ func (self *peer) getBlockHashes() {
self.addError(ErrInvalidBlock, "%v", err) self.addError(ErrInvalidBlock, "%v", err)
self.bp.status.badPeers[self.id]++ self.bp.status.badPeers[self.id]++
} else { } else {
if self.currentBlock.Td != nil {
if self.td.Cmp(self.currentBlock.Td) != 0 {
self.addError(ErrIncorrectTD, "on block %x", self.currentBlockHash)
}
}
headKey := self.parentHash.Str() headKey := self.parentHash.Str()
height := self.bp.status.chain[headKey] + 1 height := self.bp.status.chain[headKey] + 1
self.bp.status.chain[self.currentBlockHash.Str()] = height self.bp.status.chain[self.currentBlockHash.Str()] = height
@ -445,6 +490,7 @@ func (self *peer) getBlockHashes() {
block: self.currentBlock, block: self.currentBlock,
hashBy: self.id, hashBy: self.id,
blockBy: self.id, blockBy: self.id,
td: self.td,
} }
self.bp.newSection([]*node{n}).activate(self) self.bp.newSection([]*node{n}).activate(self)
} else { } else {
@ -485,7 +531,7 @@ func (self *peer) run() {
LOOP: LOOP:
for { for {
select { select {
// to minitor section process behaviou // to minitor section process behaviour
case <-ping.C: case <-ping.C:
plog.Debugf("HeadSection: <%s> section with head %s, idle: %v", self.id, hex(self.currentBlockHash), self.idle) plog.Debugf("HeadSection: <%s> section with head %s, idle: %v", self.id, hex(self.currentBlockHash), self.idle)
@ -538,7 +584,7 @@ LOOP:
// quit // quit
case <-quit: case <-quit:
self.peerError(self.bp.peers.errors.New(ErrIdleTooLong, "timed out without providing new blocks (td: %v, head: %s)...quitting", self.td, self.currentBlockHash)) self.peerError(self.bp.peers.errors.New(ErrIdleTooLong, "timed out without providing new blocks (td: %v, head: %s)...quitting", self.td, hex(self.currentBlockHash)))
self.bp.status.lock.Lock() self.bp.status.lock.Lock()
self.bp.status.badPeers[self.id]++ self.bp.status.badPeers[self.id]++

88
blockpool/peers_test.go
View File

@ -3,17 +3,21 @@ package blockpool
import ( import (
"math/big" "math/big"
"testing" "testing"
"time"
"github.com/ethereum/go-ethereum/blockpool/test" "github.com/ethereum/go-ethereum/blockpool/test"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
) )
// the actual tests // the actual tests
func TestAddPeer(t *testing.T) { func TestAddPeer(t *testing.T) {
test.LogInit() test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
peer0 := blockPoolTester.newPeer("peer0", 1, 0) peer0 := blockPoolTester.newPeer("peer0", 1, 1)
peer1 := blockPoolTester.newPeer("peer1", 2, 1) peer1 := blockPoolTester.newPeer("peer1", 2, 2)
peer2 := blockPoolTester.newPeer("peer2", 3, 2) peer2 := blockPoolTester.newPeer("peer2", 3, 3)
var bestpeer *peer var bestpeer *peer
blockPool.Start() blockPool.Start()
@ -34,7 +38,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peers.best.id != "peer2" { if blockPool.peers.best.id != "peer2" {
t.Errorf("peer2 (TD=3) not set as best") t.Errorf("peer2 (TD=3) not set as best")
} }
peer2.waitBlocksRequests(2) peer2.waitBlocksRequests(3)
best = peer1.AddPeer() best = peer1.AddPeer()
if best { if best {
@ -48,7 +52,7 @@ func TestAddPeer(t *testing.T) {
} }
peer2.td = 4 peer2.td = 4
peer2.currentBlock = 3 peer2.currentBlock = 4
best = peer2.AddPeer() best = peer2.AddPeer()
if !best { if !best {
t.Errorf("peer2 (TD=4) not accepted as best") t.Errorf("peer2 (TD=4) not accepted as best")
@ -59,10 +63,10 @@ func TestAddPeer(t *testing.T) {
if blockPool.peers.best.td.Cmp(big.NewInt(int64(4))) != 0 { if blockPool.peers.best.td.Cmp(big.NewInt(int64(4))) != 0 {
t.Errorf("peer2 TD not updated") t.Errorf("peer2 TD not updated")
} }
peer2.waitBlocksRequests(3) peer2.waitBlocksRequests(4)
peer1.td = 3 peer1.td = 3
peer1.currentBlock = 2 peer1.currentBlock = 3
best = peer1.AddPeer() best = peer1.AddPeer()
if best { if best {
t.Errorf("peer1 (TD=3) should not be set as best") t.Errorf("peer1 (TD=3) should not be set as best")
@ -84,7 +88,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peers.best.id != "peer1" { if blockPool.peers.best.id != "peer1" {
t.Errorf("existing peer1 (TD=3) should be set as best peer") t.Errorf("existing peer1 (TD=3) should be set as best peer")
} }
peer1.waitBlocksRequests(2) peer1.waitBlocksRequests(3)
blockPool.RemovePeer("peer1") blockPool.RemovePeer("peer1")
bestpeer, best = blockPool.peers.getPeer("peer1") bestpeer, best = blockPool.peers.getPeer("peer1")
@ -95,7 +99,7 @@ func TestAddPeer(t *testing.T) {
if blockPool.peers.best.id != "peer0" { if blockPool.peers.best.id != "peer0" {
t.Errorf("existing peer0 (TD=1) should be set as best peer") t.Errorf("existing peer0 (TD=1) should be set as best peer")
} }
peer0.waitBlocksRequests(0) peer0.waitBlocksRequests(1)
blockPool.RemovePeer("peer0") blockPool.RemovePeer("peer0")
bestpeer, best = blockPool.peers.getPeer("peer0") bestpeer, best = blockPool.peers.getPeer("peer0")
@ -115,6 +119,70 @@ func TestAddPeer(t *testing.T) {
} }
peer0.waitBlocksRequests(3) peer0.waitBlocksRequests(3)
blockPool.Stop() newblock := &types.Block{Td: common.Big3}
blockPool.chainEvents.Post(core.ChainHeadEvent{newblock})
time.Sleep(100 * time.Millisecond)
if blockPool.peers.best != nil {
t.Errorf("no peer should be ahead of self")
}
best = peer1.AddPeer()
if blockPool.peers.best != nil {
t.Errorf("still no peer should be ahead of self")
}
best = peer2.AddPeer()
if !best {
t.Errorf("peer2 (TD=4) not accepted as best")
}
blockPool.RemovePeer("peer2")
if blockPool.peers.best != nil {
t.Errorf("no peer should be ahead of self")
}
blockPool.Stop()
}
func TestPeerPromotionByOptionalTdOnBlock(t *testing.T) {
test.LogInit()
_, blockPool, blockPoolTester := newTestBlockPool(t)
blockPoolTester.blockChain[0] = nil
blockPoolTester.initRefBlockChain(4)
peer0 := blockPoolTester.newPeer("peer0", 2, 2)
peer1 := blockPoolTester.newPeer("peer1", 1, 1)
peer2 := blockPoolTester.newPeer("peer2", 4, 4)
blockPool.Start()
blockPoolTester.tds = make(map[int]int)
blockPoolTester.tds[3] = 3
// pool
peer0.AddPeer()
peer0.serveBlocks(1, 2)
best := peer1.AddPeer()
// this tests that peer1 is not promoted over peer0 yet
if best {
t.Errorf("peer1 (TD=1) should not be set as best")
}
best = peer2.AddPeer()
peer2.serveBlocks(3, 4)
peer2.serveBlockHashes(4, 3, 2, 1)
hashes := blockPoolTester.hashPool.IndexesToHashes([]int{2, 3})
peer1.waitBlocksRequests(3)
blockPool.AddBlock(&types.Block{
HeaderHash: common.Hash(hashes[1]),
ParentHeaderHash: common.Hash(hashes[0]),
Td: common.Big3,
}, "peer1")
blockPool.RemovePeer("peer2")
if blockPool.peers.best.id != "peer1" {
t.Errorf("peer1 (TD=3) should be set as best")
}
peer1.serveBlocks(0, 1, 2)
blockPool.Wait(waitTimeout)
blockPool.Stop()
blockPoolTester.refBlockChain[4] = []int{}
blockPoolTester.checkBlockChain(blockPoolTester.refBlockChain)
} }

37
blockpool/section.go
View File

@ -83,9 +83,9 @@ func (self *BlockPool) newSection(nodes []*node) *section {
offC: make(chan bool), offC: make(chan bool),
} }
for i, node := range nodes { for i, n := range nodes {
entry := &entry{node: node, section: sec, index: &index{i}} entry := &entry{node: n, section: sec, index: &index{i}}
self.set(node.hash, entry) self.set(n.hash, entry)
} }
plog.DebugDetailf("[%s] setup section process", sectionhex(sec)) plog.DebugDetailf("[%s] setup section process", sectionhex(sec))
@ -104,20 +104,22 @@ func (self *section) addSectionToBlockChain(p *peer) {
self.bp.wg.Done() self.bp.wg.Done()
}() }()
var node *node var nodes []*node
var keys []string var n *node
var keys []common.Hash
var blocks []*types.Block var blocks []*types.Block
for self.poolRootIndex > 0 { for self.poolRootIndex > 0 {
node = self.nodes[self.poolRootIndex-1] n = self.nodes[self.poolRootIndex-1]
node.lock.RLock() n.lock.RLock()
block := node.block block := n.block
node.lock.RUnlock() n.lock.RUnlock()
if block == nil { if block == nil {
break break
} }
self.poolRootIndex-- self.poolRootIndex--
keys = append(keys, node.hash.Str()) keys = append(keys, n.hash)
blocks = append(blocks, block) blocks = append(blocks, block)
nodes = append(nodes, n)
} }
if len(blocks) == 0 { if len(blocks) == 0 {
@ -134,13 +136,20 @@ func (self *section) addSectionToBlockChain(p *peer) {
err := self.bp.insertChain(blocks) err := self.bp.insertChain(blocks)
if err != nil { if err != nil {
self.invalid = true self.invalid = true
self.bp.peers.peerError(node.blockBy, ErrInvalidBlock, "%v", err) self.bp.peers.peerError(n.blockBy, ErrInvalidBlock, "%v", err)
plog.Warnf("invalid block %x", node.hash) plog.Warnf("invalid block %x", n.hash)
plog.Warnf("penalise peers %v (hash), %v (block)", node.hashBy, node.blockBy) plog.Warnf("penalise peers %v (hash), %v (block)", n.hashBy, n.blockBy)
// or invalid block and the entire chain needs to be removed // or invalid block and the entire chain needs to be removed
self.removeChain() self.removeChain()
} else { } else {
// check tds
self.bp.wg.Add(1)
go func() {
plog.DebugDetailf("checking td")
self.bp.checkTD(nodes...)
self.bp.wg.Done()
}()
// if all blocks inserted in this section // if all blocks inserted in this section
// then need to try to insert blocks in child section // then need to try to insert blocks in child section
if self.poolRootIndex == 0 { if self.poolRootIndex == 0 {
@ -178,7 +187,7 @@ func (self *section) addSectionToBlockChain(p *peer) {
self.bp.status.values.BlocksInChain += len(blocks) self.bp.status.values.BlocksInChain += len(blocks)
self.bp.status.values.BlocksInPool -= len(blocks) self.bp.status.values.BlocksInPool -= len(blocks)
if err != nil { if err != nil {
self.bp.status.badPeers[node.blockBy]++ self.bp.status.badPeers[n.blockBy]++
} }
self.bp.status.lock.Unlock() self.bp.status.lock.Unlock()

308
blockpool/status_test.go
View File

@ -1,9 +1,9 @@
package blockpool package blockpool
import ( import (
"fmt" // "fmt"
"testing" "testing"
// "time" "time"
"github.com/ethereum/go-ethereum/blockpool/test" "github.com/ethereum/go-ethereum/blockpool/test"
) )
@ -49,180 +49,192 @@ func checkStatus(t *testing.T, bp *BlockPool, syncing bool, expected []int) (err
} }
got := getStatusValues(s) got := getStatusValues(s)
for i, v := range expected { for i, v := range expected {
if i == 0 || i == 7 {
continue //hack
}
err = test.CheckInt(statusFields[i], got[i], v, t) err = test.CheckInt(statusFields[i], got[i], v, t)
// fmt.Printf("%v: %v (%v)\n", statusFields[i], got[i], v)
if err != nil { if err != nil {
return err return err
} }
fmt.Printf("%v: %v (%v)\n", statusFields[i], got[i], v)
} }
return return
} }
// func TestBlockPoolStatus(t *testing.T) { func TestBlockPoolStatus(t *testing.T) {
// test.LogInit() test.LogInit()
// _, blockPool, blockPoolTester := newTestBlockPool(t) _, blockPool, blockPoolTester := newTestBlockPool(t)
// blockPoolTester.blockChain[0] = nil blockPoolTester.blockChain[0] = nil
// blockPoolTester.initRefBlockChain(12) blockPoolTester.initRefBlockChain(12)
// blockPoolTester.refBlockChain[3] = []int{4, 7} blockPoolTester.refBlockChain[3] = []int{4, 7}
// delete(blockPoolTester.refBlockChain, 6) delete(blockPoolTester.refBlockChain, 6)
// blockPool.Start() blockPool.Start()
blockPoolTester.tds = make(map[int]int)
blockPoolTester.tds[9] = 1
blockPoolTester.tds[11] = 3
blockPoolTester.tds[6] = 2
// peer1 := blockPoolTester.newPeer("peer1", 1, 9) peer1 := blockPoolTester.newPeer("peer1", 1, 9)
// peer2 := blockPoolTester.newPeer("peer2", 2, 6) peer2 := blockPoolTester.newPeer("peer2", 2, 6)
// peer3 := blockPoolTester.newPeer("peer3", 3, 11) peer3 := blockPoolTester.newPeer("peer3", 3, 11)
// peer4 := blockPoolTester.newPeer("peer4", 1, 9) peer4 := blockPoolTester.newPeer("peer4", 1, 9)
// peer2.blocksRequestsMap = peer1.blocksRequestsMap // peer1 := blockPoolTester.newPeer("peer1", 1, 9)
// peer2 := blockPoolTester.newPeer("peer2", 2, 6)
// peer3 := blockPoolTester.newPeer("peer3", 3, 11)
// peer4 := blockPoolTester.newPeer("peer4", 1, 9)
peer2.blocksRequestsMap = peer1.blocksRequestsMap
// var expected []int var expected []int
// var err error var err error
// expected = []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0} expected = []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}
// err = checkStatus(t, blockPool, false, expected) err = checkStatus(t, blockPool, false, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.AddPeer() peer1.AddPeer()
// expected = []int{0, 0, 0, 0, 0, 1, 0, 0, 1, 1, 0, 1, 0} expected = []int{0, 0, 0, 0, 0, 1, 0, 0, 1, 1, 0, 1, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.serveBlocks(8, 9) peer1.serveBlocks(8, 9)
// expected = []int{0, 0, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0} expected = []int{0, 0, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0}
// err = checkStatus(t, blockPool, true, expected) // err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.serveBlockHashes(9, 8, 7, 3, 2) peer1.serveBlockHashes(9, 8, 7, 3, 2)
// expected = []int{5, 5, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0} expected = []int{6, 5, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0}
// err = checkStatus(t, blockPool, true, expected) // expected = []int{5, 5, 1, 1, 0, 1, 0, 0, 1, 1, 1, 1, 0}
// if err != nil { err = checkStatus(t, blockPool, true, expected)
// return if err != nil {
// } return
}
// peer1.serveBlocks(3, 7, 8) peer1.serveBlocks(3, 7, 8)
// expected = []int{5, 5, 3, 3, 0, 1, 0, 0, 1, 1, 1, 1, 0} expected = []int{6, 5, 3, 3, 0, 1, 0, 0, 1, 1, 1, 1, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.serveBlocks(2, 3) peer1.serveBlocks(2, 3)
// expected = []int{5, 5, 4, 4, 0, 1, 0, 0, 1, 1, 1, 1, 0} expected = []int{6, 5, 4, 4, 0, 1, 0, 0, 1, 1, 1, 1, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer4.AddPeer() peer4.AddPeer()
// expected = []int{5, 5, 4, 4, 0, 2, 0, 0, 2, 2, 1, 1, 0} expected = []int{6, 5, 4, 4, 0, 2, 0, 0, 2, 2, 1, 1, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer4.sendBlockHashes(12, 11) peer4.sendBlockHashes(12, 11)
// expected = []int{5, 5, 4, 4, 0, 2, 0, 0, 2, 2, 1, 1, 0} expected = []int{6, 5, 4, 4, 0, 2, 0, 0, 2, 2, 1, 1, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer2.AddPeer() peer2.AddPeer()
// expected = []int{5, 5, 4, 4, 0, 3, 0, 0, 3, 3, 1, 2, 0} expected = []int{6, 5, 4, 4, 0, 3, 0, 0, 3, 3, 1, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer2.serveBlocks(5, 6) peer2.serveBlocks(5, 6)
// peer2.serveBlockHashes(6, 5, 4, 3, 2) peer2.serveBlockHashes(6, 5, 4, 3, 2)
// expected = []int{8, 8, 5, 5, 0, 3, 1, 0, 3, 3, 2, 2, 0} expected = []int{10, 8, 5, 5, 0, 3, 1, 0, 3, 3, 2, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer2.serveBlocks(2, 3, 4) peer2.serveBlocks(2, 3, 4)
// expected = []int{8, 8, 6, 6, 0, 3, 1, 0, 3, 3, 2, 2, 0} expected = []int{10, 8, 6, 6, 0, 3, 1, 0, 3, 3, 2, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// blockPool.RemovePeer("peer2") blockPool.RemovePeer("peer2")
// expected = []int{8, 8, 6, 6, 0, 3, 1, 0, 3, 2, 2, 2, 0} expected = []int{10, 8, 6, 6, 0, 3, 1, 0, 3, 2, 2, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.serveBlockHashes(2, 1, 0) peer1.serveBlockHashes(2, 1, 0)
// expected = []int{9, 9, 6, 6, 0, 3, 1, 0, 3, 2, 2, 2, 0} expected = []int{11, 9, 6, 6, 0, 3, 1, 0, 3, 2, 2, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.serveBlocks(1, 2) peer1.serveBlocks(1, 2)
// expected = []int{9, 9, 7, 7, 0, 3, 1, 0, 3, 2, 2, 2, 0} expected = []int{11, 9, 7, 7, 0, 3, 1, 0, 3, 2, 2, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer1.serveBlocks(4, 5) peer1.serveBlocks(4, 5)
// expected = []int{9, 9, 8, 8, 0, 3, 1, 0, 3, 2, 2, 2, 0} expected = []int{11, 9, 8, 8, 0, 3, 1, 0, 3, 2, 2, 2, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer3.AddPeer() peer3.AddPeer()
// expected = []int{9, 9, 8, 8, 0, 4, 1, 0, 4, 3, 2, 3, 0} expected = []int{11, 9, 8, 8, 0, 4, 1, 0, 4, 3, 2, 3, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer3.serveBlocks(10, 11) peer3.serveBlocks(10, 11)
// expected = []int{9, 9, 9, 9, 0, 4, 1, 0, 4, 3, 3, 3, 0} expected = []int{12, 9, 9, 9, 0, 4, 1, 0, 4, 3, 3, 3, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer3.serveBlockHashes(11, 10, 9) peer3.serveBlockHashes(11, 10, 9)
// expected = []int{11, 11, 9, 9, 0, 4, 1, 0, 4, 3, 3, 3, 0} expected = []int{14, 11, 9, 9, 0, 4, 1, 0, 4, 3, 3, 3, 0}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer4.sendBlocks(11, 12) peer4.sendBlocks(11, 12)
// expected = []int{11, 11, 9, 9, 0, 4, 1, 0, 4, 3, 4, 3, 1} expected = []int{14, 11, 9, 9, 0, 4, 1, 0, 4, 3, 4, 3, 1}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer3.serveBlocks(9, 10) peer3.serveBlocks(9, 10)
// expected = []int{11, 11, 10, 10, 0, 4, 1, 0, 4, 3, 4, 3, 1} expected = []int{14, 11, 10, 10, 0, 4, 1, 0, 4, 3, 4, 3, 1}
// err = checkStatus(t, blockPool, true, expected) err = checkStatus(t, blockPool, true, expected)
// if err != nil { if err != nil {
// return return
// } }
// peer3.serveBlocks(0, 1) peer3.serveBlocks(0, 1)
// blockPool.Wait(waitTimeout) blockPool.Wait(waitTimeout)
// time.Sleep(200 * time.Millisecond) time.Sleep(200 * time.Millisecond)
// expected = []int{11, 3, 11, 3, 8, 4, 1, 8, 4, 3, 4, 3, 1} expected = []int{14, 3, 11, 3, 8, 4, 1, 8, 4, 3, 4, 3, 1}
// err = checkStatus(t, blockPool, false, expected) err = checkStatus(t, blockPool, false, expected)
// if err != nil { if err != nil {
// return return
// } }
// blockPool.Stop() blockPool.Stop()
// } }

32
blockpool/test/hash_pool.go
View File

@ -3,20 +3,10 @@ package test
import ( import (
"sync" "sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
) )
// test helpers
// TODO: move into common test helper package (see p2p/crypto etc.)
func NewHashPool() *TestHashPool {
return &TestHashPool{intToHash: make(intToHash), hashToInt: make(hashToInt)}
}
type intToHash map[int][]byte
type hashToInt map[string]int
// hashPool is a test helper, that allows random hashes to be referred to by integers // hashPool is a test helper, that allows random hashes to be referred to by integers
type TestHashPool struct { type TestHashPool struct {
intToHash intToHash
@ -24,11 +14,19 @@ type TestHashPool struct {
lock sync.Mutex lock sync.Mutex
} }
func newHash(i int) []byte { func NewHashPool() *TestHashPool {
return crypto.Sha3([]byte(string(i))) return &TestHashPool{intToHash: make(intToHash), hashToInt: make(hashToInt)}
} }
func (self *TestHashPool) IndexesToHashes(indexes []int) (hashes [][]byte) { type intToHash map[int]common.Hash
type hashToInt map[common.Hash]int
func newHash(i int) common.Hash {
return common.BytesToHash(crypto.Sha3([]byte(string(i))))
}
func (self *TestHashPool) IndexesToHashes(indexes []int) (hashes []common.Hash) {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
for _, i := range indexes { for _, i := range indexes {
@ -36,18 +34,18 @@ func (self *TestHashPool) IndexesToHashes(indexes []int) (hashes [][]byte) {
if !found { if !found {
hash = newHash(i) hash = newHash(i)
self.intToHash[i] = hash self.intToHash[i] = hash
self.hashToInt[string(hash)] = i self.hashToInt[hash] = i
} }
hashes = append(hashes, hash) hashes = append(hashes, hash)
} }
return return
} }
func (self *TestHashPool) HashesToIndexes(hashes [][]byte) (indexes []int) { func (self *TestHashPool) HashesToIndexes(hashes []common.Hash) (indexes []int) {
self.lock.Lock() self.lock.Lock()
defer self.lock.Unlock() defer self.lock.Unlock()
for _, hash := range hashes { for _, hash := range hashes {
i, found := self.hashToInt[string(hash)] i, found := self.hashToInt[hash]
if !found { if !found {
i = -1 i = -1
} }

4
blockpool/test/logger.go
View File

@ -9,6 +9,8 @@ import (
"github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger"
) )
// logging in tests
var once sync.Once var once sync.Once
/* usage: /* usage:
@ -19,7 +21,7 @@ func TestFunc(t *testing.T) {
*/ */
func LogInit() { func LogInit() {
once.Do(func() { once.Do(func() {
var logsys = logger.NewStdLogSystem(os.Stdout, log.LstdFlags, logger.LogLevel(logger.DebugDetailLevel)) var logsys = logger.NewStdLogSystem(os.Stdout, log.LstdFlags, logger.LogLevel(logger.WarnLevel))
logger.AddLogSystem(logsys) logger.AddLogSystem(logsys)
}) })
} }

2
blockpool/test/util.go
View File

@ -6,6 +6,8 @@ import (
"time" "time"
) )
// miscellaneous test helpers
func CheckInt(name string, got int, expected int, t *testing.T) (err error) { func CheckInt(name string, got int, expected int, t *testing.T) (err error) {
if got != expected { if got != expected {
t.Errorf("status for %v incorrect. expected %v, got %v", name, expected, got) t.Errorf("status for %v incorrect. expected %v, got %v", name, expected, got)

68
core/block_cache.go Normal file
View File

@ -0,0 +1,68 @@
package core
import (
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
)
// BlockCache implements a caching mechanism specifically for blocks and uses FILO to pop
type BlockCache struct {
size int
hashes []common.Hash
blocks map[common.Hash]*types.Block
mu sync.RWMutex
}
// Creates and returns a `BlockCache` with `size`. If `size` is smaller than 1 it will panic
func NewBlockCache(size int) *BlockCache {
if size < 1 {
panic("block cache size not allowed to be smaller than 1")
}
bc := &BlockCache{size: size}
bc.Clear()
return bc
}
func (bc *BlockCache) Clear() {
bc.blocks = make(map[common.Hash]*types.Block)
bc.hashes = nil
}
func (bc *BlockCache) Push(block *types.Block) {
bc.mu.Lock()
defer bc.mu.Unlock()
if len(bc.hashes) == bc.size {
delete(bc.blocks, bc.hashes[0])
// XXX There are a few other options on solving this
// 1) use a poller / GC like mechanism to clean up untracked objects
// 2) copy as below
// re-use the slice and remove the reference to bc.hashes[0]
// this will allow the element to be garbage collected.
copy(bc.hashes, bc.hashes[1:])
} else {
bc.hashes = append(bc.hashes, common.Hash{})
}
hash := block.Hash()
bc.blocks[hash] = block
bc.hashes[len(bc.hashes)-1] = hash
}
func (bc *BlockCache) Get(hash common.Hash) *types.Block {
bc.mu.RLock()
defer bc.mu.RUnlock()
if block, haz := bc.blocks[hash]; haz {
return block
}
return nil
}

48
core/block_cache_test.go Normal file
View File

@ -0,0 +1,48 @@
package core
import (
"math/big"
"testing"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
)
func newChain(size int) (chain []*types.Block) {
var parentHash common.Hash
for i := 0; i < size; i++ {
block := types.NewBlock(parentHash, common.Address{}, common.Hash{}, new(big.Int), 0, "")
block.Header().Number = big.NewInt(int64(i))
chain = append(chain, block)
parentHash = block.Hash()
}
return
}
func insertChainCache(cache *BlockCache, chain []*types.Block) {
for _, block := range chain {
cache.Push(block)
}
}
func TestNewBlockCache(t *testing.T) {
chain := newChain(3)
cache := NewBlockCache(2)
insertChainCache(cache, chain)
if cache.hashes[0] != chain[1].Hash() {
t.Error("oldest block incorrect")
}
}
func TestInclusion(t *testing.T) {
chain := newChain(3)
cache := NewBlockCache(3)
insertChainCache(cache, chain)
for _, block := range chain {
if b := cache.Get(block.Hash()); b == nil {
t.Errorf("getting %x failed", block.Hash())
}
}
}

20
core/chain_manager.go
View File

@ -23,6 +23,8 @@ var (
blockNumPre = []byte("block-num-") blockNumPre = []byte("block-num-")
) )
const blockCacheLimit = 10000
type StateQuery interface { type StateQuery interface {
GetAccount(addr []byte) *state.StateObject GetAccount(addr []byte) *state.StateObject
} }
@ -92,15 +94,25 @@ type ChainManager struct {
transState *state.StateDB transState *state.StateDB
txState *state.ManagedState txState *state.ManagedState
cache *BlockCache
quit chan struct{} quit chan struct{}
} }
func NewChainManager(blockDb, stateDb common.Database, mux *event.TypeMux) *ChainManager { func NewChainManager(blockDb, stateDb common.Database, mux *event.TypeMux) *ChainManager {
bc := &ChainManager{blockDb: blockDb, stateDb: stateDb, genesisBlock: GenesisBlock(stateDb), eventMux: mux, quit: make(chan struct{})} bc := &ChainManager{blockDb: blockDb, stateDb: stateDb, genesisBlock: GenesisBlock(stateDb), eventMux: mux, quit: make(chan struct{}), cache: NewBlockCache(blockCacheLimit)}
bc.setLastBlock() bc.setLastBlock()
bc.transState = bc.State().Copy() bc.transState = bc.State().Copy()
// Take ownership of this particular state // Take ownership of this particular state
bc.txState = state.ManageState(bc.State().Copy()) bc.txState = state.ManageState(bc.State().Copy())
// load in last `blockCacheLimit` - 1 blocks. Last block is the current.
ancestors := bc.GetAncestors(bc.currentBlock, blockCacheLimit-1)
ancestors = append(ancestors, bc.currentBlock)
for _, block := range ancestors {
bc.cache.Push(block)
}
go bc.update() go bc.update()
return bc return bc
@ -275,6 +287,8 @@ func (bc *ChainManager) insert(block *types.Block) {
key := append(blockNumPre, block.Number().Bytes()...) key := append(blockNumPre, block.Number().Bytes()...)
bc.blockDb.Put(key, bc.lastBlockHash.Bytes()) bc.blockDb.Put(key, bc.lastBlockHash.Bytes())
// Push block to cache
bc.cache.Push(block)
} }
func (bc *ChainManager) write(block *types.Block) { func (bc *ChainManager) write(block *types.Block) {
@ -318,6 +332,10 @@ func (self *ChainManager) GetBlockHashesFromHash(hash common.Hash, max uint64) (
} }
func (self *ChainManager) GetBlock(hash common.Hash) *types.Block { func (self *ChainManager) GetBlock(hash common.Hash) *types.Block {
if block := self.cache.Get(hash); block != nil {
return block
}
data, _ := self.blockDb.Get(append(blockHashPre, hash[:]...)) data, _ := self.blockDb.Get(append(blockHashPre, hash[:]...))
if len(data) == 0 { if len(data) == 0 {
return nil return nil

2
core/chain_manager_test.go
View File

@ -69,7 +69,7 @@ func printChain(bc *ChainManager) {
func testChain(chainB types.Blocks, bman *BlockProcessor) (*big.Int, error) { func testChain(chainB types.Blocks, bman *BlockProcessor) (*big.Int, error) {
td := new(big.Int) td := new(big.Int)
for _, block := range chainB { for _, block := range chainB {
td2, err := bman.bc.processor.Process(block) td2, _, err := bman.bc.processor.Process(block)
if err != nil { if err != nil {
if IsKnownBlockErr(err) { if IsKnownBlockErr(err) {
continue continue

4
core/filter.go
View File

@ -34,7 +34,7 @@ type Filter struct {
topics [][]common.Hash topics [][]common.Hash
BlockCallback func(*types.Block, state.Logs) BlockCallback func(*types.Block, state.Logs)
PendingCallback func(*types.Block, state.Logs) PendingCallback func(*types.Transaction)
LogsCallback func(state.Logs) LogsCallback func(state.Logs)
} }
@ -46,7 +46,7 @@ func NewFilter(eth Backend) *Filter {
// SetOptions copies the filter options to the filter it self. The reason for this "silly" copy // SetOptions copies the filter options to the filter it self. The reason for this "silly" copy
// is simply because named arguments in this case is extremely nice and readable. // is simply because named arguments in this case is extremely nice and readable.
func (self *Filter) SetOptions(options FilterOptions) { func (self *Filter) SetOptions(options *FilterOptions) {
self.earliest = options.Earliest self.earliest = options.Earliest
self.latest = options.Latest self.latest = options.Latest
self.skip = options.Skip self.skip = options.Skip

5
eth/backend.go
View File

@ -195,7 +195,8 @@ func New(config *Config) (*Ethereum, error) {
hasBlock := eth.chainManager.HasBlock hasBlock := eth.chainManager.HasBlock
insertChain := eth.chainManager.InsertChain insertChain := eth.chainManager.InsertChain
eth.blockPool = blockpool.New(hasBlock, insertChain, eth.pow.Verify) td := eth.chainManager.Td()
eth.blockPool = blockpool.New(hasBlock, insertChain, eth.pow.Verify, eth.EventMux(), td)
netprv, err := config.nodeKey() netprv, err := config.nodeKey()
if err != nil { if err != nil {
@ -205,7 +206,7 @@ func New(config *Config) (*Ethereum, error) {
ethProto := EthProtocol(config.ProtocolVersion, config.NetworkId, eth.txPool, eth.chainManager, eth.blockPool) ethProto := EthProtocol(config.ProtocolVersion, config.NetworkId, eth.txPool, eth.chainManager, eth.blockPool)
protocols := []p2p.Protocol{ethProto} protocols := []p2p.Protocol{ethProto}
if config.Shh { if config.Shh {
protocols = append(protocols, eth.whisper.Protocol()) //protocols = append(protocols, eth.whisper.Protocol())
} }
eth.net = &p2p.Server{ eth.net = &p2p.Server{

16
eth/protocol.go
View File

@ -42,6 +42,7 @@ const (
ErrGenesisBlockMismatch ErrGenesisBlockMismatch
ErrNoStatusMsg ErrNoStatusMsg
ErrExtraStatusMsg ErrExtraStatusMsg
ErrSuspendedPeer
) )
var errorToString = map[int]string{ var errorToString = map[int]string{
@ -53,6 +54,7 @@ var errorToString = map[int]string{
ErrGenesisBlockMismatch: "Genesis block mismatch", ErrGenesisBlockMismatch: "Genesis block mismatch",
ErrNoStatusMsg: "No status message", ErrNoStatusMsg: "No status message",
ErrExtraStatusMsg: "Extra status message", ErrExtraStatusMsg: "Extra status message",
ErrSuspendedPeer: "Suspended peer",
} }
// ethProtocol represents the ethereum wire protocol // ethProtocol represents the ethereum wire protocol
@ -85,7 +87,7 @@ type chainManager interface {
type blockPool interface { type blockPool interface {
AddBlockHashes(next func() (common.Hash, bool), peerId string) AddBlockHashes(next func() (common.Hash, bool), peerId string)
AddBlock(block *types.Block, peerId string) AddBlock(block *types.Block, peerId string)
AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool)
RemovePeer(peerId string) RemovePeer(peerId string)
} }
@ -211,8 +213,7 @@ func (self *ethProtocol) handle() error {
var i int var i int
iter := func() (hash common.Hash, ok bool) { iter := func() (hash common.Hash, ok bool) {
var h common.Hash err := msgStream.Decode(&hash)
err := msgStream.Decode(&h)
if err == rlp.EOL { if err == rlp.EOL {
return common.Hash{}, false return common.Hash{}, false
} else if err != nil { } else if err != nil {
@ -288,7 +289,7 @@ func (self *ethProtocol) handle() error {
// to simplify backend interface adding a new block // to simplify backend interface adding a new block
// uses AddPeer followed by AddBlock only if peer is the best peer // uses AddPeer followed by AddBlock only if peer is the best peer
// (or selected as new best peer) // (or selected as new best peer)
if self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) { if best, _ := self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect); best {
self.blockPool.AddBlock(request.Block, self.id) self.blockPool.AddBlock(request.Block, self.id)
} }
@ -334,9 +335,12 @@ func (self *ethProtocol) handleStatus() error {
return self.protoError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, self.protocolVersion) return self.protoError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, self.protocolVersion)
} }
self.peer.Infof("Peer is [eth] capable (%d/%d). TD=%v H=%x\n", status.ProtocolVersion, status.NetworkId, status.TD, status.CurrentBlock[:4]) _, suspended := self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect)
if suspended {
return self.protoError(ErrSuspendedPeer, "")
}
self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) self.peer.Infof("Peer is [eth] capable (%d/%d). TD=%v H=%x\n", status.ProtocolVersion, status.NetworkId, status.TD, status.CurrentBlock[:4])
return nil return nil
} }

13
eth/protocol_test.go
View File

@ -41,17 +41,10 @@ type testChainManager struct {
type testBlockPool struct { type testBlockPool struct {
addBlockHashes func(next func() (common.Hash, bool), peerId string) addBlockHashes func(next func() (common.Hash, bool), peerId string)
addBlock func(block *types.Block, peerId string) (err error) addBlock func(block *types.Block, peerId string) (err error)
addPeer func(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) addPeer func(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool)
removePeer func(peerId string) removePeer func(peerId string)
} }
// func (self *testTxPool) GetTransactions() (txs []*types.Transaction) {
// if self.getTransactions != nil {
// txs = self.getTransactions()
// }
// return
// }
func (self *testTxPool) AddTransactions(txs []*types.Transaction) { func (self *testTxPool) AddTransactions(txs []*types.Transaction) {
if self.addTransactions != nil { if self.addTransactions != nil {
self.addTransactions(txs) self.addTransactions(txs)
@ -93,9 +86,9 @@ func (self *testBlockPool) AddBlock(block *types.Block, peerId string) {
} }
} }
func (self *testBlockPool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestBlockHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool) { func (self *testBlockPool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestBlockHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool) {
if self.addPeer != nil { if self.addPeer != nil {
best = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError) best, suspended = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError)
} }
return return
} }

80
eth/wallet.go
View File

@ -1,80 +0,0 @@
package eth
/*
import (
"crypto/ecdsa"
"errors"
"math/big"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
)
type Account struct {
w *Wallet
}
func (self *Account) Transact(to *Account, value, gas, price *big.Int, data []byte) error {
return self.w.transact(self, to, value, gas, price, data)
}
func (self *Account) Address() []byte {
return nil
}
func (self *Account) PrivateKey() *ecdsa.PrivateKey {
return nil
}
type Wallet struct{}
func NewWallet() *Wallet {
return &Wallet{}
}
func (self *Wallet) GetAccount(i int) *Account {
}
func (self *Wallet) transact(from, to *Account, value, gas, price *big.Int, data []byte) error {
if from.PrivateKey() == nil {
return errors.New("accounts is not owned (no private key available)")
}
var createsContract bool
if to == nil {
createsContract = true
}
var msg *types.Transaction
if contractCreation {
msg = types.NewContractCreationTx(value, gas, price, data)
} else {
msg = types.NewTransactionMessage(to.Address(), value, gas, price, data)
}
state := self.chainManager.TransState()
nonce := state.GetNonce(key.Address())
msg.SetNonce(nonce)
msg.SignECDSA(from.PriateKey())
// Do some pre processing for our "pre" events and hooks
block := self.chainManager.NewBlock(from.Address())
coinbase := state.GetOrNewStateObject(from.Address())
coinbase.SetGasPool(block.GasLimit())
self.blockManager.ApplyTransactions(coinbase, state, block, types.Transactions{tx}, true)
err := self.obj.TxPool().Add(tx)
if err != nil {
return nil, err
}
state.SetNonce(key.Address(), nonce+1)
if contractCreation {
addr := core.AddressFromMessage(tx)
pipelogger.Infof("Contract addr %x\n", addr)
}
return tx, nil
}
*/

11
event/filter/eth_filter.go
View File

@ -48,7 +48,9 @@ func (self *FilterManager) InstallFilter(filter *core.Filter) (id int) {
func (self *FilterManager) UninstallFilter(id int) { func (self *FilterManager) UninstallFilter(id int) {
self.filterMu.Lock() self.filterMu.Lock()
defer self.filterMu.Unlock() defer self.filterMu.Unlock()
delete(self.filters, id) if _, ok := self.filters[id]; ok {
delete(self.filters, id)
}
} }
// GetFilter retrieves a filter installed using InstallFilter. // GetFilter retrieves a filter installed using InstallFilter.
@ -62,8 +64,9 @@ func (self *FilterManager) GetFilter(id int) *core.Filter {
func (self *FilterManager) filterLoop() { func (self *FilterManager) filterLoop() {
// Subscribe to events // Subscribe to events
events := self.eventMux.Subscribe( events := self.eventMux.Subscribe(
core.PendingBlockEvent{}, //core.PendingBlockEvent{},
core.ChainEvent{}, core.ChainEvent{},
core.TxPreEvent{},
state.Logs(nil)) state.Logs(nil))
out: out:
@ -82,11 +85,11 @@ out:
} }
self.filterMu.RUnlock() self.filterMu.RUnlock()
case core.PendingBlockEvent: case core.TxPreEvent:
self.filterMu.RLock() self.filterMu.RLock()
for _, filter := range self.filters { for _, filter := range self.filters {
if filter.PendingCallback != nil { if filter.PendingCallback != nil {
filter.PendingCallback(event.Block, event.Logs) filter.PendingCallback(event.Tx)
} }
} }
self.filterMu.RUnlock() self.filterMu.RUnlock()

4
miner/agent.go
View File

@ -79,3 +79,7 @@ func (self *CpuMiner) mine(block *types.Block) {
self.returnCh <- Work{block.Number().Uint64(), nonce, mixDigest, seedHash} self.returnCh <- Work{block.Number().Uint64(), nonce, mixDigest, seedHash}
} }
} }
func (self *CpuMiner) GetHashRate() int64 {
return self.pow.GetHashrate()
}

4
miner/worker.go
View File

@ -57,7 +57,7 @@ type Agent interface {
SetWorkCh(chan<- Work) SetWorkCh(chan<- Work)
Stop() Stop()
Start() Start()
Pow() pow.PoW GetHashRate() int64
} }
type worker struct { type worker struct {
@ -272,7 +272,7 @@ func (self *worker) commitTransaction(tx *types.Transaction) error {
func (self *worker) HashRate() int64 { func (self *worker) HashRate() int64 {
var tot int64 var tot int64
for _, agent := range self.agents { for _, agent := range self.agents {
tot += agent.Pow().GetHashrate() tot += agent.GetHashRate()
} }
return tot return tot

743
rpc/api.go
View File

File diff suppressed because it is too large Load Diff

60
rpc/api_test.go
View File

@ -2,9 +2,9 @@ package rpc
import ( import (
"encoding/json" "encoding/json"
"sync" // "sync"
"testing" "testing"
"time" // "time"
) )
func TestWeb3Sha3(t *testing.T) { func TestWeb3Sha3(t *testing.T) {
@ -24,33 +24,33 @@ func TestWeb3Sha3(t *testing.T) {
} }
} }
func TestFilterClose(t *testing.T) { // func TestFilterClose(t *testing.T) {
t.Skip() // t.Skip()
api := &EthereumApi{ // api := &EthereumApi{
logs: make(map[int]*logFilter), // logs: make(map[int]*logFilter),
messages: make(map[int]*whisperFilter), // messages: make(map[int]*whisperFilter),
quit: make(chan struct{}), // quit: make(chan struct{}),
} // }
filterTickerTime = 1 // filterTickerTime = 1
api.logs[0] = &logFilter{} // api.logs[0] = &logFilter{}
api.messages[0] = &whisperFilter{} // api.messages[0] = &whisperFilter{}
var wg sync.WaitGroup // var wg sync.WaitGroup
wg.Add(1) // wg.Add(1)
go api.start() // go api.start()
go func() { // go func() {
select { // select {
case <-time.After(500 * time.Millisecond): // case <-time.After(500 * time.Millisecond):
api.stop() // api.stop()
wg.Done() // wg.Done()
} // }
}() // }()
wg.Wait() // wg.Wait()
if len(api.logs) != 0 { // if len(api.logs) != 0 {
t.Error("expected logs to be empty") // t.Error("expected logs to be empty")
} // }
if len(api.messages) != 0 { // if len(api.messages) != 0 {
t.Error("expected messages to be empty") // t.Error("expected messages to be empty")
} // }
} // }

26
rpc/args.go
View File

@ -35,8 +35,8 @@ func blockAge(raw interface{}, number *int64) (err error) {
} }
type GetBlockByHashArgs struct { type GetBlockByHashArgs struct {
BlockHash string BlockHash string
Transactions bool IncludeTxs bool
} }
func (args *GetBlockByHashArgs) UnmarshalJSON(b []byte) (err error) { func (args *GetBlockByHashArgs) UnmarshalJSON(b []byte) (err error) {
@ -57,15 +57,15 @@ func (args *GetBlockByHashArgs) UnmarshalJSON(b []byte) (err error) {
args.BlockHash = argstr args.BlockHash = argstr
if len(obj) > 1 { if len(obj) > 1 {
args.Transactions = obj[1].(bool) args.IncludeTxs = obj[1].(bool)
} }
return nil return nil
} }
type GetBlockByNumberArgs struct { type GetBlockByNumberArgs struct {
BlockNumber int64 BlockNumber int64
Transactions bool IncludeTxs bool
} }
func (args *GetBlockByNumberArgs) UnmarshalJSON(b []byte) (err error) { func (args *GetBlockByNumberArgs) UnmarshalJSON(b []byte) (err error) {
@ -86,7 +86,7 @@ func (args *GetBlockByNumberArgs) UnmarshalJSON(b []byte) (err error) {
} }
if len(obj) > 1 { if len(obj) > 1 {
args.Transactions = obj[1].(bool) args.IncludeTxs = obj[1].(bool)
} }
return nil return nil
@ -433,7 +433,7 @@ func (args *Sha3Args) UnmarshalJSON(b []byte) (err error) {
return nil return nil
} }
type FilterOptions struct { type BlockFilterArgs struct {
Earliest int64 Earliest int64
Latest int64 Latest int64
Address interface{} Address interface{}
@ -442,7 +442,7 @@ type FilterOptions struct {
Max int Max int
} }
func (args *FilterOptions) UnmarshalJSON(b []byte) (err error) { func (args *BlockFilterArgs) UnmarshalJSON(b []byte) (err error) {
var obj []struct { var obj []struct {
FromBlock interface{} `json:"fromBlock"` FromBlock interface{} `json:"fromBlock"`
ToBlock interface{} `json:"toBlock"` ToBlock interface{} `json:"toBlock"`
@ -609,6 +609,16 @@ func (args *FilterStringArgs) UnmarshalJSON(b []byte) (err error) {
return nil return nil
} }
func (args *FilterStringArgs) requirements() error {
switch args.Word {
case "latest", "pending":
break
default:
return NewValidationError("Word", "Must be `latest` or `pending`")
}
return nil
}
type FilterIdArgs struct { type FilterIdArgs struct {
Id int Id int
} }

32
rpc/args_test.go
View File

@ -82,7 +82,7 @@ func TestGetBlockByHashArgs(t *testing.T) {
input := `["0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331", true]` input := `["0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331", true]`
expected := new(GetBlockByHashArgs) expected := new(GetBlockByHashArgs)
expected.BlockHash = "0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331" expected.BlockHash = "0xe670ec64341771606e55d6b4ca35a1a6b75ee3d5145a99d05921026d1527331"
expected.Transactions = true expected.IncludeTxs = true
args := new(GetBlockByHashArgs) args := new(GetBlockByHashArgs)
if err := json.Unmarshal([]byte(input), &args); err != nil { if err := json.Unmarshal([]byte(input), &args); err != nil {
@ -93,8 +93,8 @@ func TestGetBlockByHashArgs(t *testing.T) {
t.Errorf("BlockHash should be %v but is %v", expected.BlockHash, args.BlockHash) t.Errorf("BlockHash should be %v but is %v", expected.BlockHash, args.BlockHash)
} }
if args.Transactions != expected.Transactions { if args.IncludeTxs != expected.IncludeTxs {
t.Errorf("Transactions should be %v but is %v", expected.Transactions, args.Transactions) t.Errorf("IncludeTxs should be %v but is %v", expected.IncludeTxs, args.IncludeTxs)
} }
} }
@ -112,7 +112,7 @@ func TestGetBlockByNumberArgs(t *testing.T) {
input := `["0x1b4", false]` input := `["0x1b4", false]`
expected := new(GetBlockByNumberArgs) expected := new(GetBlockByNumberArgs)
expected.BlockNumber = 436 expected.BlockNumber = 436
expected.Transactions = false expected.IncludeTxs = false
args := new(GetBlockByNumberArgs) args := new(GetBlockByNumberArgs)
if err := json.Unmarshal([]byte(input), &args); err != nil { if err := json.Unmarshal([]byte(input), &args); err != nil {
@ -123,8 +123,8 @@ func TestGetBlockByNumberArgs(t *testing.T) {
t.Errorf("BlockHash should be %v but is %v", expected.BlockNumber, args.BlockNumber) t.Errorf("BlockHash should be %v but is %v", expected.BlockNumber, args.BlockNumber)
} }
if args.Transactions != expected.Transactions { if args.IncludeTxs != expected.IncludeTxs {
t.Errorf("Transactions should be %v but is %v", expected.Transactions, args.Transactions) t.Errorf("IncludeTxs should be %v but is %v", expected.IncludeTxs, args.IncludeTxs)
} }
} }
@ -388,7 +388,7 @@ func TestGetDataEmptyArgs(t *testing.T) {
} }
} }
func TestFilterOptions(t *testing.T) { func TestBlockFilterArgs(t *testing.T) {
input := `[{ input := `[{
"fromBlock": "0x1", "fromBlock": "0x1",
"toBlock": "0x2", "toBlock": "0x2",
@ -396,7 +396,7 @@ func TestFilterOptions(t *testing.T) {
"offset": "0x0", "offset": "0x0",
"address": "0xd5677cf67b5aa051bb40496e68ad359eb97cfbf8", "address": "0xd5677cf67b5aa051bb40496e68ad359eb97cfbf8",
"topics": ["0x12341234"]}]` "topics": ["0x12341234"]}]`
expected := new(FilterOptions) expected := new(BlockFilterArgs)
expected.Earliest = 1 expected.Earliest = 1
expected.Latest = 2 expected.Latest = 2
expected.Max = 3 expected.Max = 3
@ -404,7 +404,7 @@ func TestFilterOptions(t *testing.T) {
expected.Address = "0xd5677cf67b5aa051bb40496e68ad359eb97cfbf8" expected.Address = "0xd5677cf67b5aa051bb40496e68ad359eb97cfbf8"
// expected.Topics = []string{"0x12341234"} // expected.Topics = []string{"0x12341234"}
args := new(FilterOptions) args := new(BlockFilterArgs)
if err := json.Unmarshal([]byte(input), &args); err != nil { if err := json.Unmarshal([]byte(input), &args); err != nil {
t.Error(err) t.Error(err)
} }
@ -434,16 +434,16 @@ func TestFilterOptions(t *testing.T) {
// } // }
} }
func TestFilterOptionsWords(t *testing.T) { func TestBlockFilterArgsWords(t *testing.T) {
input := `[{ input := `[{
"fromBlock": "latest", "fromBlock": "latest",
"toBlock": "pending" "toBlock": "pending"
}]` }]`
expected := new(FilterOptions) expected := new(BlockFilterArgs)
expected.Earliest = 0 expected.Earliest = 0
expected.Latest = -1 expected.Latest = -1
args := new(FilterOptions) args := new(BlockFilterArgs)
if err := json.Unmarshal([]byte(input), &args); err != nil { if err := json.Unmarshal([]byte(input), &args); err != nil {
t.Error(err) t.Error(err)
} }
@ -457,13 +457,13 @@ func TestFilterOptionsWords(t *testing.T) {
} }
} }
func TestFilterOptionsNums(t *testing.T) { func TestBlockFilterArgsNums(t *testing.T) {
input := `[{ input := `[{
"fromBlock": 2, "fromBlock": 2,
"toBlock": 3 "toBlock": 3
}]` }]`
args := new(FilterOptions) args := new(BlockFilterArgs)
err := json.Unmarshal([]byte(input), &args) err := json.Unmarshal([]byte(input), &args)
switch err.(type) { switch err.(type) {
case *DecodeParamError: case *DecodeParamError:
@ -474,10 +474,10 @@ func TestFilterOptionsNums(t *testing.T) {
} }
func TestFilterOptionsEmptyArgs(t *testing.T) { func TestBlockFilterArgsEmptyArgs(t *testing.T) {
input := `[]` input := `[]`
args := new(FilterOptions) args := new(BlockFilterArgs)
err := json.Unmarshal([]byte(input), &args) err := json.Unmarshal([]byte(input), &args)
if err == nil { if err == nil {
t.Error("Expected error but didn't get one") t.Error("Expected error but didn't get one")

16
rpc/http.go
View File

@ -10,7 +10,7 @@ import (
"github.com/ethereum/go-ethereum/xeth" "github.com/ethereum/go-ethereum/xeth"
) )
var rpchttplogger = logger.NewLogger("RPC-HTTP") var rpclogger = logger.NewLogger("RPC")
const ( const (
jsonrpcver = "2.0" jsonrpcver = "2.0"
@ -28,7 +28,7 @@ func JSONRPC(pipe *xeth.XEth, dataDir string) http.Handler {
// Limit request size to resist DoS // Limit request size to resist DoS
if req.ContentLength > maxSizeReqLength { if req.ContentLength > maxSizeReqLength {
jsonerr := &RpcErrorObject{-32700, "Request too large"} jsonerr := &RpcErrorObject{-32700, "Request too large"}
Send(w, &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: nil, Error: jsonerr}) send(w, &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: nil, Error: jsonerr})
return return
} }
@ -37,14 +37,14 @@ func JSONRPC(pipe *xeth.XEth, dataDir string) http.Handler {
body, err := ioutil.ReadAll(req.Body) body, err := ioutil.ReadAll(req.Body)
if err != nil { if err != nil {
jsonerr := &RpcErrorObject{-32700, "Could not read request body"} jsonerr := &RpcErrorObject{-32700, "Could not read request body"}
Send(w, &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: nil, Error: jsonerr}) send(w, &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: nil, Error: jsonerr})
} }
// Try to parse the request as a single // Try to parse the request as a single
var reqSingle RpcRequest var reqSingle RpcRequest
if err := json.Unmarshal(body, &reqSingle); err == nil { if err := json.Unmarshal(body, &reqSingle); err == nil {
response := RpcResponse(api, &reqSingle) response := RpcResponse(api, &reqSingle)
Send(w, &response) send(w, &response)
return return
} }
@ -57,13 +57,13 @@ func JSONRPC(pipe *xeth.XEth, dataDir string) http.Handler {
response := RpcResponse(api, &request) response := RpcResponse(api, &request)
resBatch[i] = response resBatch[i] = response
} }
Send(w, resBatch) send(w, resBatch)
return return
} }
// Not a batch or single request, error // Not a batch or single request, error
jsonerr := &RpcErrorObject{-32600, "Could not decode request"} jsonerr := &RpcErrorObject{-32600, "Could not decode request"}
Send(w, &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: nil, Error: jsonerr}) send(w, &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: nil, Error: jsonerr})
}) })
} }
@ -84,11 +84,11 @@ func RpcResponse(api *EthereumApi, request *RpcRequest) *interface{} {
response = &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: request.Id, Error: jsonerr} response = &RpcErrorResponse{Jsonrpc: jsonrpcver, Id: request.Id, Error: jsonerr}
} }
rpchttplogger.DebugDetailf("Generated response: %T %s", response, response) rpclogger.DebugDetailf("Generated response: %T %s", response, response)
return &response return &response
} }
func Send(writer io.Writer, v interface{}) (n int, err error) { func send(writer io.Writer, v interface{}) (n int, err error) {
var payload []byte var payload []byte
payload, err = json.MarshalIndent(v, "", "\t") payload, err = json.MarshalIndent(v, "", "\t")
if err != nil { if err != nil {

41
rpc/messages_test.go Normal file
View File

@ -0,0 +1,41 @@
package rpc
import (
"testing"
)
func TestInsufficientParamsError(t *testing.T) {
err := NewInsufficientParamsError(0, 1)
expected := "insufficient params, want 1 have 0"
if err.Error() != expected {
t.Error(err.Error())
}
}
func TestNotImplementedError(t *testing.T) {
err := NewNotImplementedError("foo")
expected := "foo method not implemented"
if err.Error() != expected {
t.Error(err.Error())
}
}
func TestDecodeParamError(t *testing.T) {
err := NewDecodeParamError("foo")
expected := "could not decode, foo"
if err.Error() != expected {
t.Error(err.Error())
}
}
func TestValidationError(t *testing.T) {
err := NewValidationError("foo", "should be `bar`")
expected := "foo not valid, should be `bar`"
if err.Error() != expected {
t.Error(err.Error())
}
}

26
rpc/responses.go
View File

@ -7,6 +7,7 @@ import (
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/state"
) )
type BlockRes struct { type BlockRes struct {
@ -215,3 +216,28 @@ type FilterWhisperRes struct {
Payload string `json:"payload"` Payload string `json:"payload"`
WorkProved string `json:"workProved"` WorkProved string `json:"workProved"`
} }
type LogRes struct {
Address string `json:"address"`
Topic []string `json:"topic"`
Data string `json:"data"`
Number uint64 `json:"number"`
}
func NewLogsRes(logs state.Logs) (ls []LogRes) {
ls = make([]LogRes, len(logs))
for i, log := range logs {
var l LogRes
l.Topic = make([]string, len(log.Topics()))
l.Address = log.Address().Hex()
l.Data = common.ToHex(log.Data())
l.Number = log.Number()
for j, topic := range log.Topics() {
l.Topic[j] = topic.Hex()
}
ls[i] = l
}
return
}

86
rpc/util.go
View File

@ -1,86 +0,0 @@
/*
This file is part of go-ethereum
go-ethereum is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
go-ethereum is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with go-ethereum. If not, see <http://www.gnu.org/licenses/>.
*/
package rpc
import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/state"
"github.com/ethereum/go-ethereum/xeth"
)
var rpclogger = logger.NewLogger("RPC")
type Log struct {
Address string `json:"address"`
Topic []string `json:"topic"`
Data string `json:"data"`
Number uint64 `json:"number"`
}
func toLogs(logs state.Logs) (ls []Log) {
ls = make([]Log, len(logs))
for i, log := range logs {
var l Log
l.Topic = make([]string, len(log.Topics()))
l.Address = log.Address().Hex()
l.Data = common.ToHex(log.Data())
l.Number = log.Number()
for j, topic := range log.Topics() {
l.Topic[j] = topic.Hex()
}
ls[i] = l
}
return
}
type whisperFilter struct {
messages []xeth.WhisperMessage
timeout time.Time
id int
}
func (w *whisperFilter) add(msgs ...xeth.WhisperMessage) {
w.messages = append(w.messages, msgs...)
}
func (w *whisperFilter) get() []xeth.WhisperMessage {
w.timeout = time.Now()
tmp := w.messages
w.messages = nil
return tmp
}
type logFilter struct {
logs state.Logs
timeout time.Time
id int
}
func (l *logFilter) add(logs ...state.Log) {
l.logs = append(l.logs, logs...)
}
func (l *logFilter) get() state.Logs {
l.timeout = time.Now()
tmp := l.logs
l.logs = nil
return tmp
}

326
xeth/xeth.go
View File

@ -6,6 +6,8 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"math/big" "math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/accounts"
"github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common"
@ -13,13 +15,19 @@ import (
"github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/event" "github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/event/filter"
"github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/state" "github.com/ethereum/go-ethereum/state"
"github.com/ethereum/go-ethereum/whisper" "github.com/ethereum/go-ethereum/whisper"
) )
var pipelogger = logger.NewLogger("XETH") var (
pipelogger = logger.NewLogger("XETH")
filterTickerTime = 5 * time.Minute
defaultGasPrice = big.NewInt(10000000000000) //150000000000
defaultGas = big.NewInt(90000) //500000
)
// to resolve the import cycle // to resolve the import cycle
type Backend interface { type Backend interface {
@ -62,6 +70,13 @@ type Frontend interface {
ConfirmTransaction(tx *types.Transaction) bool ConfirmTransaction(tx *types.Transaction) bool
} }
// dummyFrontend is a non-interactive frontend that allows all
// transactions but cannot not unlock any keys.
type dummyFrontend struct{}
func (dummyFrontend) UnlockAccount([]byte) bool { return false }
func (dummyFrontend) ConfirmTransaction(*types.Transaction) bool { return true }
type XEth struct { type XEth struct {
eth Backend eth Backend
blockProcessor *core.BlockProcessor blockProcessor *core.BlockProcessor
@ -71,15 +86,20 @@ type XEth struct {
whisper *Whisper whisper *Whisper
frontend Frontend frontend Frontend
quit chan struct{}
filterManager *filter.FilterManager
logMut sync.RWMutex
logs map[int]*logFilter
messagesMut sync.RWMutex
messages map[int]*whisperFilter
// regmut sync.Mutex
// register map[string][]*interface{} // TODO improve return type
} }
// dummyFrontend is a non-interactive frontend that allows all
// transactions but cannot not unlock any keys.
type dummyFrontend struct{}
func (dummyFrontend) UnlockAccount([]byte) bool { return false }
func (dummyFrontend) ConfirmTransaction(*types.Transaction) bool { return true }
// New creates an XEth that uses the given frontend. // New creates an XEth that uses the given frontend.
// If a nil Frontend is provided, a default frontend which // If a nil Frontend is provided, a default frontend which
// confirms all transactions will be used. // confirms all transactions will be used.
@ -90,15 +110,76 @@ func New(eth Backend, frontend Frontend) *XEth {
chainManager: eth.ChainManager(), chainManager: eth.ChainManager(),
accountManager: eth.AccountManager(), accountManager: eth.AccountManager(),
whisper: NewWhisper(eth.Whisper()), whisper: NewWhisper(eth.Whisper()),
quit: make(chan struct{}),
filterManager: filter.NewFilterManager(eth.EventMux()),
frontend: frontend, frontend: frontend,
logs: make(map[int]*logFilter),
messages: make(map[int]*whisperFilter),
} }
if frontend == nil { if frontend == nil {
xeth.frontend = dummyFrontend{} xeth.frontend = dummyFrontend{}
} }
xeth.state = NewState(xeth, xeth.chainManager.TransState()) xeth.state = NewState(xeth, xeth.chainManager.TransState())
go xeth.start()
go xeth.filterManager.Start()
return xeth return xeth
} }
func (self *XEth) start() {
timer := time.NewTicker(2 * time.Second)
done:
for {
select {
case <-timer.C:
self.logMut.Lock()
self.messagesMut.Lock()
for id, filter := range self.logs {
if time.Since(filter.timeout) > filterTickerTime {
self.filterManager.UninstallFilter(id)
delete(self.logs, id)
}
}
for id, filter := range self.messages {
if time.Since(filter.timeout) > filterTickerTime {
self.Whisper().Unwatch(id)
delete(self.messages, id)
}
}
self.messagesMut.Unlock()
self.logMut.Unlock()
case <-self.quit:
break done
}
}
}
func (self *XEth) stop() {
close(self.quit)
}
func (self *XEth) DefaultGas() *big.Int { return defaultGas }
func (self *XEth) DefaultGasPrice() *big.Int { return defaultGasPrice }
func (self *XEth) AtStateNum(num int64) *XEth {
chain := self.Backend().ChainManager()
var block *types.Block
if num < 0 {
num = chain.CurrentBlock().Number().Int64() + num + 1
}
block = chain.GetBlockByNumber(uint64(num))
var st *state.StateDB
if block != nil {
st = state.New(block.Root(), self.Backend().StateDb())
} else {
st = chain.State()
}
return self.WithState(st)
}
func (self *XEth) Backend() Backend { return self.eth } func (self *XEth) Backend() Backend { return self.eth }
func (self *XEth) WithState(statedb *state.StateDB) *XEth { func (self *XEth) WithState(statedb *state.StateDB) *XEth {
xeth := &XEth{ xeth := &XEth{
@ -241,6 +322,157 @@ func (self *XEth) SecretToAddress(key string) string {
return common.ToHex(pair.Address()) return common.ToHex(pair.Address())
} }
func (self *XEth) RegisterFilter(args *core.FilterOptions) int {
var id int
filter := core.NewFilter(self.Backend())
filter.SetOptions(args)
filter.LogsCallback = func(logs state.Logs) {
self.logMut.Lock()
defer self.logMut.Unlock()
self.logs[id].add(logs...)
}
id = self.filterManager.InstallFilter(filter)
self.logs[id] = &logFilter{timeout: time.Now()}
return id
}
func (self *XEth) UninstallFilter(id int) bool {
if _, ok := self.logs[id]; ok {
delete(self.logs, id)
self.filterManager.UninstallFilter(id)
return true
}
return false
}
func (self *XEth) NewFilterString(word string) int {
var id int
filter := core.NewFilter(self.Backend())
switch word {
case "pending":
filter.PendingCallback = func(tx *types.Transaction) {
self.logMut.Lock()
defer self.logMut.Unlock()
self.logs[id].add(&state.StateLog{})
}
case "latest":
filter.BlockCallback = func(block *types.Block, logs state.Logs) {
self.logMut.Lock()
defer self.logMut.Unlock()
for _, log := range logs {
self.logs[id].add(log)
}
self.logs[id].add(&state.StateLog{})
}
}
id = self.filterManager.InstallFilter(filter)
self.logs[id] = &logFilter{timeout: time.Now()}
return id
}
func (self *XEth) FilterChanged(id int) state.Logs {
self.logMut.Lock()
defer self.logMut.Unlock()
if self.logs[id] != nil {
return self.logs[id].get()
}
return nil
}
func (self *XEth) Logs(id int) state.Logs {
self.logMut.Lock()
defer self.logMut.Unlock()
filter := self.filterManager.GetFilter(id)
if filter != nil {
return filter.Find()
}
return nil
}
func (self *XEth) AllLogs(args *core.FilterOptions) state.Logs {
filter := core.NewFilter(self.Backend())
filter.SetOptions(args)
return filter.Find()
}
func (p *XEth) NewWhisperFilter(opts *Options) int {
var id int
opts.Fn = func(msg WhisperMessage) {
p.messagesMut.Lock()
defer p.messagesMut.Unlock()
p.messages[id].add(msg) // = append(p.messages[id], msg)
}
id = p.Whisper().Watch(opts)
p.messages[id] = &whisperFilter{timeout: time.Now()}
return id
}
func (p *XEth) UninstallWhisperFilter(id int) bool {
if _, ok := p.messages[id]; ok {
delete(p.messages, id)
return true
}
return false
}
func (self *XEth) MessagesChanged(id int) []WhisperMessage {
self.messagesMut.Lock()
defer self.messagesMut.Unlock()
if self.messages[id] != nil {
return self.messages[id].get()
}
return nil
}
// func (self *XEth) Register(args string) bool {
// self.regmut.Lock()
// defer self.regmut.Unlock()
// if _, ok := self.register[args]; ok {
// self.register[args] = nil // register with empty
// }
// return true
// }
// func (self *XEth) Unregister(args string) bool {
// self.regmut.Lock()
// defer self.regmut.Unlock()
// if _, ok := self.register[args]; ok {
// delete(self.register, args)
// return true
// }
// return false
// }
// // TODO improve return type
// func (self *XEth) PullWatchTx(args string) []*interface{} {
// self.regmut.Lock()
// defer self.regmut.Unlock()
// txs := self.register[args]
// self.register[args] = nil
// return txs
// }
type KeyVal struct { type KeyVal struct {
Key string `json:"key"` Key string `json:"key"`
Value string `json:"value"` Value string `json:"value"`
@ -298,11 +530,6 @@ func (self *XEth) PushTx(encodedTx string) (string, error) {
return tx.Hash().Hex(), nil return tx.Hash().Hex(), nil
} }
var (
defaultGasPrice = big.NewInt(10000000000000)
defaultGas = big.NewInt(90000)
)
func (self *XEth) Call(fromStr, toStr, valueStr, gasStr, gasPriceStr, dataStr string) (string, error) { func (self *XEth) Call(fromStr, toStr, valueStr, gasStr, gasPriceStr, dataStr string) (string, error) {
statedb := self.State().State() //self.chainManager.TransState() statedb := self.State().State() //self.chainManager.TransState()
msg := callmsg{ msg := callmsg{
@ -333,12 +560,44 @@ func (self *XEth) Transact(fromStr, toStr, valueStr, gasStr, gasPriceStr, codeSt
from = common.HexToAddress(fromStr) from = common.HexToAddress(fromStr)
to = common.HexToAddress(toStr) to = common.HexToAddress(toStr)
value = common.NewValue(valueStr) value = common.NewValue(valueStr)
gas = common.NewValue(gasStr) gas = common.Big(gasStr)
price = common.NewValue(gasPriceStr) price = common.Big(gasPriceStr)
data []byte data []byte
contractCreation bool contractCreation bool
) )
// TODO if no_private_key then
//if _, exists := p.register[args.From]; exists {
// p.register[args.From] = append(p.register[args.From], args)
//} else {
/*
account := accounts.Get(common.FromHex(args.From))
if account != nil {
if account.Unlocked() {
if !unlockAccount(account) {
return
}
}
result, _ := account.Transact(common.FromHex(args.To), common.FromHex(args.Value), common.FromHex(args.Gas), common.FromHex(args.GasPrice), common.FromHex(args.Data))
if len(result) > 0 {
*reply = common.ToHex(result)
}
} else if _, exists := p.register[args.From]; exists {
p.register[ags.From] = append(p.register[args.From], args)
}
*/
// TODO: align default values to have the same type, e.g. not depend on
// common.Value conversions later on
if gas.Cmp(big.NewInt(0)) == 0 {
gas = defaultGas
}
if price.Cmp(big.NewInt(0)) == 0 {
price = defaultGasPrice
}
data = common.FromHex(codeStr) data = common.FromHex(codeStr)
if len(toStr) == 0 { if len(toStr) == 0 {
contractCreation = true contractCreation = true
@ -346,9 +605,9 @@ func (self *XEth) Transact(fromStr, toStr, valueStr, gasStr, gasPriceStr, codeSt
var tx *types.Transaction var tx *types.Transaction
if contractCreation { if contractCreation {
tx = types.NewContractCreationTx(value.BigInt(), gas.BigInt(), price.BigInt(), data) tx = types.NewContractCreationTx(value.BigInt(), gas, price, data)
} else { } else {
tx = types.NewTransactionMessage(to, value.BigInt(), gas.BigInt(), price.BigInt(), data) tx = types.NewTransactionMessage(to, value.BigInt(), gas, price, data)
} }
state := self.chainManager.TxState() state := self.chainManager.TxState()
@ -407,3 +666,36 @@ func (m callmsg) GasPrice() *big.Int { return m.gasPrice }
func (m callmsg) Gas() *big.Int { return m.gas } func (m callmsg) Gas() *big.Int { return m.gas }
func (m callmsg) Value() *big.Int { return m.value } func (m callmsg) Value() *big.Int { return m.value }
func (m callmsg) Data() []byte { return m.data } func (m callmsg) Data() []byte { return m.data }
type whisperFilter struct {
messages []WhisperMessage
timeout time.Time
id int
}
func (w *whisperFilter) add(msgs ...WhisperMessage) {
w.messages = append(w.messages, msgs...)
}
func (w *whisperFilter) get() []WhisperMessage {
w.timeout = time.Now()
tmp := w.messages
w.messages = nil
return tmp
}
type logFilter struct {
logs state.Logs
timeout time.Time
id int
}
func (l *logFilter) add(logs ...state.Log) {
l.logs = append(l.logs, logs...)
}
func (l *logFilter) get() state.Logs {
l.timeout = time.Now()
tmp := l.logs
l.logs = nil
return tmp
}