cerc-io/plugeth
18
0
Fork 1
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

eth, eth/downloader, miner: use download events to check miner start

Browse Source
This commit is contained in:
obscuren 2015-05-15 00:43:00 +02:00
parent 0f76a1c6df
commit b71091e337
6 changed files with 83 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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 {

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{})
}
}()

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()
}