interesting
/
go-ethereum
mirror of https://github.com/ethereum/go-ethereum.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #987 from obscuren/develop 

Miner updates and downloader events
This commit is contained in:
Jeffrey Wilcke 2015-05-15 04:12:46 -07:00
parent 158efbaa45 cfb2b51b74
commit c1f0d40e34
12 changed files with 110 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
core/block_processor.go
View File

@ -347,17 +347,17 @@ func (sm *BlockProcessor) VerifyUncles(statedb *state.StateDB, block, parent *ty
for i, uncle := range block.Uncles() {
if uncles.Has(uncle.Hash()) {
// Error not unique
return UncleError("Uncle not unique")
return UncleError("uncle[%d] not unique", i)
}
uncles.Add(uncle.Hash())
if ancestors.Has(uncle.Hash()) {
return UncleError("Uncle is ancestor")
return UncleError("uncle[%d] is ancestor", i)
}
if !ancestors.Has(uncle.ParentHash) {
return UncleError(fmt.Sprintf("Uncle's parent unknown (%x)", uncle.ParentHash[0:4]))
return UncleError("uncle[%d]'s parent unknown (%x)", i, uncle.ParentHash[0:4])
}
if err := sm.ValidateHeader(uncle, ancestorHeaders[uncle.ParentHash]); err != nil {

2
core/chain_makers.go
View File

@ -98,7 +98,7 @@ func makeChain(bman *BlockProcessor, parent *types.Block, max int, db common.Dat
fmt.Println("process with parent failed", err)
panic(err)
}
block.Td = CalculateTD(block, parent)
block.Td = CalcTD(block, parent)
blocks[i] = block
parent = block
}

28
core/chain_manager.go
View File

@ -48,7 +48,7 @@ func CalcDifficulty(block, parent *types.Header) *big.Int {
return diff
}
func CalculateTD(block, parent *types.Block) *big.Int {
func CalcTD(block, parent *types.Block) *big.Int {
if parent == nil {
return block.Difficulty()
}
@ -59,14 +59,20 @@ func CalculateTD(block, parent *types.Block) *big.Int {
}
func CalcGasLimit(parent *types.Block) *big.Int {
// ((1024-1) * parent.gasLimit + (gasUsed * 6 / 5)) / 1024
previous := new(big.Int).Mul(big.NewInt(1024-1), parent.GasLimit())
current := new(big.Rat).Mul(new(big.Rat).SetInt(parent.GasUsed()), big.NewRat(6, 5))
curInt := new(big.Int).Div(current.Num(), current.Denom())
decay := new(big.Int).Div(parent.GasLimit(), params.GasLimitBoundDivisor)
contrib := new(big.Int).Mul(parent.GasUsed(), big.NewInt(3))
contrib = contrib.Div(contrib, big.NewInt(2))
contrib = contrib.Div(contrib, params.GasLimitBoundDivisor)
result := new(big.Int).Add(previous, curInt)
result.Div(result, big.NewInt(1024))
return common.BigMax(params.GenesisGasLimit, result)
gl := new(big.Int).Sub(parent.GasLimit(), decay)
gl = gl.Add(gl, contrib)
gl = common.BigMax(gl, params.MinGasLimit)
if gl.Cmp(params.GenesisGasLimit) < 0 {
gl2 := new(big.Int).Add(parent.GasLimit(), decay)
return common.BigMin(params.GenesisGasLimit, gl2)
}
return gl
}
type ChainManager struct {
@ -525,7 +531,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
}
// Setting block.Td regardless of error (known for example) prevents errors down the line
// in the protocol handler
block.Td = new(big.Int).Set(CalculateTD(block, self.GetBlock(block.ParentHash())))
block.Td = new(big.Int).Set(CalcTD(block, self.GetBlock(block.ParentHash())))
// Call in to the block processor and check for errors. It's likely that if one block fails
// all others will fail too (unless a known block is returned).
@ -593,7 +599,7 @@ func (self *ChainManager) InsertChain(chain types.Blocks) (int, error) {
self.setTransState(state.New(block.Root(), self.stateDb))
self.txState.SetState(state.New(block.Root(), self.stateDb))
queue[i] = ChainEvent{block, logs}
queue[i] = ChainEvent{block, block.Hash(), logs}
queueEvent.canonicalCount++
if glog.V(logger.Debug) {
@ -683,7 +689,7 @@ out:
case ChainEvent:
// We need some control over the mining operation. Acquiring locks and waiting for the miner to create new block takes too long
// and in most cases isn't even necessary.
if i+1 == ev.canonicalCount {
if self.lastBlockHash == event.Hash {
self.currentGasLimit = CalcGasLimit(event.Block)
self.eventMux.Post(ChainHeadEvent{event.Block})
}

2
core/chain_manager_test.go
View File

@ -81,7 +81,7 @@ func testChain(chainB types.Blocks, bman *BlockProcessor) (*big.Int, error) {
return nil, err
}
parent := bman.bc.GetBlock(block.ParentHash())
block.Td = CalculateTD(block, parent)
block.Td = CalcTD(block, parent)
td = block.Td
bman.bc.mu.Lock()

4
core/error.go
View File

@ -42,8 +42,8 @@ func (err *UncleErr) Error() string {
return err.Message
}
func UncleError(str string) error {
return &UncleErr{Message: str}
func UncleError(format string, v ...interface{}) error {
return &UncleErr{Message: fmt.Sprintf(format, v...)}
}
func IsUncleErr(err error) bool {

2
core/events.go
View File

@ -3,6 +3,7 @@ package core
import (
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
)
@ -27,6 +28,7 @@ type ChainSplitEvent struct {
type ChainEvent struct {
Block *types.Block
Hash common.Hash
Logs state.Logs
}

3
core/manager.go
View File

@ -3,10 +3,12 @@ package core
import (
"github.com/ethereum/go-ethereum/accounts"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/p2p"
)
// TODO move this to types?
type Backend interface {
AccountManager() *accounts.Manager
BlockProcessor() *BlockProcessor
@ -18,4 +20,5 @@ type Backend interface {
BlockDb() common.Database
StateDb() common.Database
EventMux() *event.TypeMux
Downloader() *downloader.Downloader
}

4
eth/backend.go
View File

@ -265,12 +265,12 @@ func New(config *Config) (*Ethereum, error) {
}
eth.chainManager = core.NewChainManager(blockDb, stateDb, eth.EventMux())
eth.downloader = downloader.New(eth.chainManager.HasBlock, eth.chainManager.GetBlock)
eth.downloader = downloader.New(eth.EventMux(), eth.chainManager.HasBlock, eth.chainManager.GetBlock)
eth.pow = ethash.New()
eth.txPool = core.NewTxPool(eth.EventMux(), eth.chainManager.State, eth.chainManager.GasLimit)
eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.txPool, eth.chainManager, eth.EventMux())
eth.chainManager.SetProcessor(eth.blockProcessor)
eth.miner = miner.New(eth, eth.pow)
eth.miner = miner.New(eth, eth.EventMux(), eth.pow)
eth.miner.SetGasPrice(config.GasPrice)
eth.protocolManager = NewProtocolManager(config.ProtocolVersion, config.NetworkId, eth.eventMux, eth.txPool, eth.chainManager, eth.downloader)

17
eth/downloader/downloader.go
View File

@ -9,6 +9,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
)
@ -55,6 +56,8 @@ type hashPack struct {
}
type Downloader struct {
mux *event.TypeMux
mu sync.RWMutex
queue *queue
peers *peerSet
@ -76,8 +79,9 @@ type Downloader struct {
cancelLock sync.RWMutex // Lock to protect the cancel channel in delivers
}
func New(hasBlock hashCheckFn, getBlock getBlockFn) *Downloader {
func New(mux *event.TypeMux, hasBlock hashCheckFn, getBlock getBlockFn) *Downloader {
downloader := &Downloader{
mux: mux,
queue: newQueue(),
peers: newPeerSet(),
hasBlock: hasBlock,
@ -93,6 +97,11 @@ func (d *Downloader) Stats() (current int, max int) {
return d.queue.Size()
}
// Synchronising returns the state of the downloader
func (d *Downloader) Synchronising() bool {
return atomic.LoadInt32(&d.synchronising) > 0
}
// RegisterPeer injects a new download peer into the set of block source to be
// used for fetching hashes and blocks from.
func (d *Downloader) RegisterPeer(id string, head common.Hash, getHashes hashFetcherFn, getBlocks blockFetcherFn) error {
@ -129,6 +138,9 @@ func (d *Downloader) Synchronise(id string, hash common.Hash) error {
if atomic.CompareAndSwapInt32(&d.notified, 0, 1) {
glog.V(logger.Info).Infoln("Block synchronisation started")
}
d.mux.Post(StartEvent{})
// Create cancel channel for aborting mid-flight
d.cancelLock.Lock()
d.cancelCh = make(chan struct{})
@ -166,6 +178,9 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash) (err error) {
// reset on error
if err != nil {
d.queue.Reset()
d.mux.Post(FailedEvent{err})
} else {
d.mux.Post(DoneEvent{})
}
}()

4
eth/downloader/downloader_test.go
View File

@ -8,6 +8,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event"
)
var (
@ -67,7 +68,8 @@ func newTester(t *testing.T, hashes []common.Hash, blocks map[common.Hash]*types
done: make(chan bool),
}
downloader := New(tester.hasBlock, tester.getBlock)
var mux event.TypeMux
downloader := New(&mux, tester.hasBlock, tester.getBlock)
tester.downloader = downloader
return tester

5
eth/downloader/events.go Normal file
View File

@ -0,0 +1,5 @@
package downloader
type DoneEvent struct{}
type StartEvent struct{}
type FailedEvent struct{ Err error }

62
miner/miner.go
View File

@ -2,33 +2,63 @@ package miner
import (
"math/big"
"sync/atomic"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/event"
"github.com/ethereum/go-ethereum/logger"
"github.com/ethereum/go-ethereum/logger/glog"
"github.com/ethereum/go-ethereum/pow"
)
type Miner struct {
mux *event.TypeMux
worker *worker
MinAcceptedGasPrice *big.Int
threads int
mining bool
coinbase common.Address
mining int32
eth core.Backend
pow pow.PoW
canStart int32 // can start indicates whether we can start the mining operation
shouldStart int32 // should start indicates whether we should start after sync
}
func New(eth core.Backend, pow pow.PoW) *Miner {
return &Miner{eth: eth, pow: pow, worker: newWorker(common.Address{}, eth)}
func New(eth core.Backend, mux *event.TypeMux, pow pow.PoW) *Miner {
miner := &Miner{eth: eth, mux: mux, pow: pow, worker: newWorker(common.Address{}, eth), canStart: 1}
go miner.update()
return miner
}
func (self *Miner) Mining() bool {
return self.mining
func (self *Miner) update() {
events := self.mux.Subscribe(downloader.StartEvent{}, downloader.DoneEvent{}, downloader.FailedEvent{})
for ev := range events.Chan() {
switch ev.(type) {
case downloader.StartEvent:
atomic.StoreInt32(&self.canStart, 0)
if self.Mining() {
self.Stop()
glog.V(logger.Info).Infoln("Mining operation aborted due to sync operation")
}
case downloader.DoneEvent, downloader.FailedEvent:
shouldStart := atomic.LoadInt32(&self.shouldStart) == 1
atomic.StoreInt32(&self.canStart, 1)
atomic.StoreInt32(&self.shouldStart, 0)
if shouldStart {
self.Start(self.coinbase, self.threads)
}
}
}
}
func (m *Miner) SetGasPrice(price *big.Int) {
@ -41,34 +71,46 @@ func (m *Miner) SetGasPrice(price *big.Int) {
}
func (self *Miner) Start(coinbase common.Address, threads int) {
atomic.StoreInt32(&self.shouldStart, 1)
self.threads = threads
self.worker.coinbase = coinbase
self.mining = true
if atomic.LoadInt32(&self.canStart) == 0 {
glog.V(logger.Info).Infoln("Can not start mining operation due to network sync (starts when finished)")
return
}
atomic.StoreInt32(&self.mining, 1)
for i := 0; i < threads; i++ {
self.worker.register(NewCpuAgent(i, self.pow))
}
self.threads = threads
glog.V(logger.Info).Infof("Starting mining operation (CPU=%d TOT=%d)\n", threads, len(self.worker.agents))
self.worker.coinbase = coinbase
self.worker.start()
self.worker.commitNewWork()
}
func (self *Miner) Stop() {
self.worker.stop()
self.mining = false
atomic.StoreInt32(&self.mining, 0)
atomic.StoreInt32(&self.shouldStart, 0)
}
func (self *Miner) Register(agent Agent) {
if self.mining {
if atomic.LoadInt32(&self.mining) == 0 {
agent.Start()
}
self.worker.register(agent)
}
func (self *Miner) Mining() bool {
return atomic.LoadInt32(&self.mining) > 0
}
func (self *Miner) HashRate() int64 {
return self.pow.GetHashrate()
}