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

cmd, eth, miner: make recommit configurable

Browse source
This commit is contained in:
rjl493456442 2018-08-20 11:00:33 +08:00
parent 7d38d53ae4
commit 72f710ab87
11 changed files with 313 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/geth/main.go
View file

@ -106,6 +106,7 @@ var (
utils.MinerLegacyEtherbaseFlag,
utils.MinerExtraDataFlag,
utils.MinerLegacyExtraDataFlag,
utils.MinerRecommitIntervalFlag,
utils.NATFlag,
utils.NoDiscoverFlag,
utils.DiscoveryV5Flag,

1
cmd/geth/usage.go
View file

@ -190,6 +190,7 @@ var AppHelpFlagGroups = []flagGroup{
utils.MinerGasTargetFlag,
utils.MinerEtherbaseFlag,
utils.MinerExtraDataFlag,
utils.MinerRecommitIntervalFlag,
},
},
{

8
cmd/utils/flags.go
View file

@ -360,6 +360,11 @@ var (
Name: "extradata",
Usage: "Block extra data set by the miner (default = client version, deprecated, use --miner.extradata)",
}
MinerRecommitIntervalFlag = cli.IntFlag{
Name: "miner.recommit",
Usage: "Sealing work recommit interval(ms), will be dynamically adjusted by the system if it is too short",
Value: 3000,
}
// Account settings
UnlockedAccountFlag = cli.StringFlag{
Name: "unlock",
@ -1135,6 +1140,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
if ctx.GlobalIsSet(MinerGasPriceFlag.Name) {
cfg.GasPrice = GlobalBig(ctx, MinerGasPriceFlag.Name)
}
if ctx.GlobalIsSet(MinerRecommitIntervalFlag.Name) {
cfg.RecommitInterval = time.Duration(ctx.GlobalInt(MinerRecommitIntervalFlag.Name)) * time.Millisecond
}
if ctx.GlobalIsSet(VMEnableDebugFlag.Name) {
// TODO(fjl): force-enable this in --dev mode
cfg.EnablePreimageRecording = ctx.GlobalBool(VMEnableDebugFlag.Name)

6
eth/api.go
View file

@ -25,6 +25,7 @@ import (
"math/big"
"os"
"strings"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
@ -160,6 +161,11 @@ func (api *PrivateMinerAPI) SetEtherbase(etherbase common.Address) bool {
return true
}
// SetRecommitInterval updates the interval for miner sealing work recommitting.
func (api *PrivateMinerAPI) SetRecommitInterval(interval int) {
api.e.Miner().SetRecommitInterval(time.Duration(interval) * time.Millisecond)
}
// GetHashrate returns the current hashrate of the miner.
func (api *PrivateMinerAPI) GetHashrate() uint64 {
return api.e.miner.HashRate()

2
eth/backend.go
View file

@ -167,7 +167,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
return nil, err
}
eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine)
eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine, config.RecommitInterval)
eth.miner.SetExtra(makeExtraData(config.ExtraData))
eth.APIBackend = &EthAPIBackend{eth, nil}

1
eth/config.go
View file

@ -100,6 +100,7 @@ type Config struct {
MinerNotify []string `toml:",omitempty"`
ExtraData []byte `toml:",omitempty"`
GasPrice *big.Int
RecommitInterval time.Duration
// Ethash options
Ethash ethash.Config

33
eth/gen_config.go
View file

@ -4,6 +4,7 @@ package eth
import (
"math/big"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
@ -15,20 +16,26 @@ import (
var _ = (*configMarshaling)(nil)
// MarshalTOML marshals as TOML.
func (c Config) MarshalTOML() (interface{}, error) {
type Config struct {
Genesis *core.Genesis `toml:",omitempty"`
NetworkId uint64
SyncMode downloader.SyncMode
NoPruning bool
LightServ int `toml:",omitempty"`
LightPeers int `toml:",omitempty"`
SkipBcVersionCheck bool `toml:"-"`
DatabaseHandles int `toml:"-"`
DatabaseCache int
TrieCache int
TrieTimeout time.Duration
Etherbase common.Address `toml:",omitempty"`
MinerThreads int `toml:",omitempty"`
MinerNotify []string `toml:",omitempty"`
ExtraData hexutil.Bytes `toml:",omitempty"`
GasPrice *big.Int
RecommitInterval time.Duration
Ethash ethash.Config
TxPool core.TxPoolConfig
GPO gasprice.Config
@ -39,15 +46,20 @@ func (c Config) MarshalTOML() (interface{}, error) {
enc.Genesis = c.Genesis
enc.NetworkId = c.NetworkId
enc.SyncMode = c.SyncMode
enc.NoPruning = c.NoPruning
enc.LightServ = c.LightServ
enc.LightPeers = c.LightPeers
enc.SkipBcVersionCheck = c.SkipBcVersionCheck
enc.DatabaseHandles = c.DatabaseHandles
enc.DatabaseCache = c.DatabaseCache
enc.TrieCache = c.TrieCache
enc.TrieTimeout = c.TrieTimeout
enc.Etherbase = c.Etherbase
enc.MinerThreads = c.MinerThreads
enc.MinerNotify = c.MinerNotify
enc.ExtraData = c.ExtraData
enc.GasPrice = c.GasPrice
enc.RecommitInterval = c.RecommitInterval
enc.Ethash = c.Ethash
enc.TxPool = c.TxPool
enc.GPO = c.GPO
@ -56,20 +68,26 @@ func (c Config) MarshalTOML() (interface{}, error) {
return &enc, nil
}
// UnmarshalTOML unmarshals from TOML.
func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
type Config struct {
Genesis *core.Genesis `toml:",omitempty"`
NetworkId *uint64
SyncMode *downloader.SyncMode
NoPruning *bool
LightServ *int `toml:",omitempty"`
LightPeers *int `toml:",omitempty"`
SkipBcVersionCheck *bool `toml:"-"`
DatabaseHandles *int `toml:"-"`
DatabaseCache *int
TrieCache *int
TrieTimeout *time.Duration
Etherbase *common.Address `toml:",omitempty"`
MinerThreads *int `toml:",omitempty"`
MinerNotify []string `toml:",omitempty"`
ExtraData *hexutil.Bytes `toml:",omitempty"`
GasPrice *big.Int
RecommitInterval *time.Duration
Ethash *ethash.Config
TxPool *core.TxPoolConfig
GPO *gasprice.Config
@ -89,6 +107,9 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.SyncMode != nil {
c.SyncMode = *dec.SyncMode
}
if dec.NoPruning != nil {
c.NoPruning = *dec.NoPruning
}
if dec.LightServ != nil {
c.LightServ = *dec.LightServ
}
@ -104,18 +125,30 @@ func (c *Config) UnmarshalTOML(unmarshal func(interface{}) error) error {
if dec.DatabaseCache != nil {
c.DatabaseCache = *dec.DatabaseCache
}
if dec.TrieCache != nil {
c.TrieCache = *dec.TrieCache
}
if dec.TrieTimeout != nil {
c.TrieTimeout = *dec.TrieTimeout
}
if dec.Etherbase != nil {
c.Etherbase = *dec.Etherbase
}
if dec.MinerThreads != nil {
c.MinerThreads = *dec.MinerThreads
}
if dec.MinerNotify != nil {
c.MinerNotify = dec.MinerNotify
}
if dec.ExtraData != nil {
c.ExtraData = *dec.ExtraData
}
if dec.GasPrice != nil {
c.GasPrice = dec.GasPrice
}
if dec.RecommitInterval != nil {
c.RecommitInterval = *dec.RecommitInterval
}
if dec.Ethash != nil {
c.Ethash = *dec.Ethash
}

5
internal/web3ext/web3ext.go
View file

@ -519,6 +519,11 @@ web3._extend({
params: 1,
inputFormatter: [web3._extend.utils.fromDecimal]
}),
new web3._extend.Method({
name: 'setRecommitInterval',
call: 'miner_setRecommitInterval',
params: 1,
}),
new web3._extend.Method({
name: 'getHashrate',
call: 'miner_getHashrate'

9
miner/miner.go
View file

@ -20,6 +20,7 @@ package miner
import (
"fmt"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/consensus"
@ -51,13 +52,13 @@ type Miner struct {
shouldStart int32 // should start indicates whether we should start after sync
}
func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine consensus.Engine) *Miner {
func New(eth Backend, config *params.ChainConfig, mux *event.TypeMux, engine consensus.Engine, recommitInterval time.Duration) *Miner {
miner := &Miner{
eth: eth,
mux: mux,
engine: engine,
exitCh: make(chan struct{}),
worker: newWorker(config, engine, eth, mux),
worker: newWorker(config, engine, eth, mux, recommitInterval),
canStart: 1,
}
go miner.update()
@ -144,6 +145,10 @@ func (self *Miner) SetExtra(extra []byte) error {
return nil
}
func (self *Miner) SetRecommitInterval(interval time.Duration) {
self.worker.setRecommitInterval(interval)
}
// Pending returns the currently pending block and associated state.
func (self *Miner) Pending() (*types.Block, *state.StateDB) {
return self.worker.pending()

130
miner/worker.go
View file

@ -51,12 +51,27 @@ const (
// chainSideChanSize is the size of channel listening to ChainSideEvent.
chainSideChanSize = 10
// resubmitAdjustChanSize is the size of resubmitting interval adjustment channel.
resubmitAdjustChanSize = 10
// miningLogAtDepth is the number of confirmations before logging successful mining.
miningLogAtDepth = 5
// blockRecommitInterval is the time interval to recreate the mining block with
// minRecommitInterval is the minimal time interval to recreate the mining block with
// any newly arrived transactions.
blockRecommitInterval = 3 * time.Second
minRecommitInterval = 1 * time.Second
// maxRecommitInterval is the maximum time interval to recreate the mining block with
// any newly arrived transactions.
maxRecommitInterval = 15 * time.Second
// intervalAdjustRatio is the impact a single interval adjustment has on sealing work
// resubmitting interval.
intervalAdjustRatio = 0.1
// intervalAdjustBias is applied during the new resubmit interval calculation in favor of
// increasing upper limit or decreasing lower limit so that the limit can be reachable.
intervalAdjustBias = 200 * 1000.0 * 1000.0
)
// environment is the worker's current environment and holds all of the current state information.
@ -89,11 +104,18 @@ const (
commitInterruptResubmit
)
// newWorkReq represents a request for new sealing work submitting with relative interrupt notifier.
type newWorkReq struct {
interrupt *int32
noempty bool
}
// intervalAdjust represents a resubmitting interval adjustment.
type intervalAdjust struct {
ratio float64
inc bool
}
// worker is the main object which takes care of submitting new work to consensus engine
// and gathering the sealing result.
type worker struct {
@ -117,6 +139,8 @@ type worker struct {
resultCh chan *task
startCh chan struct{}
exitCh chan struct{}
resubmitIntervalCh chan time.Duration
resubmitAdjustCh chan *intervalAdjust
current *environment // An environment for current running cycle.
possibleUncles map[common.Hash]*types.Block // A set of side blocks as the possible uncle blocks.
@ -134,12 +158,13 @@ type worker struct {
running int32 // The indicator whether the consensus engine is running or not.
// Test hooks
newTaskHook func(*task) // Method to call upon receiving a new sealing task
newTaskHook func(*task) // Method to call upon receiving a new sealing task.
skipSealHook func(*task) bool // Method to decide whether skipping the sealing.
fullTaskHook func() // Method to call before pushing the full sealing task
fullTaskHook func() // Method to call before pushing the full sealing task.
resubmitHook func(time.Duration, time.Duration) // Method to call upon updating resubmitting interval.
}
func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux) *worker {
func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend, mux *event.TypeMux, recommitInterval time.Duration) *worker {
worker := &worker{
config: config,
engine: engine,
@ -156,6 +181,8 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
resultCh: make(chan *task, resultQueueSize),
exitCh: make(chan struct{}),
startCh: make(chan struct{}, 1),
resubmitIntervalCh: make(chan time.Duration),
resubmitAdjustCh: make(chan *intervalAdjust, resubmitAdjustChanSize),
}
// Subscribe NewTxsEvent for tx pool
worker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh)
@ -163,8 +190,14 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, eth Backend,
worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
worker.chainSideSub = eth.BlockChain().SubscribeChainSideEvent(worker.chainSideCh)
// Recap recommit interval if the user-specified one is too short.
if recommitInterval < minRecommitInterval {
log.Info("Recap miner recommit interval", "from", recommitInterval, "to", minRecommitInterval)
recommitInterval = minRecommitInterval
}
go worker.mainLoop()
go worker.newWorkLoop()
go worker.newWorkLoop(recommitInterval)
go worker.resultLoop()
go worker.taskLoop()
@ -188,6 +221,11 @@ func (w *worker) setExtra(extra []byte) {
w.extra = extra
}
// setRecommitInterval updates the interval for miner sealing work recommitting.
func (w *worker) setRecommitInterval(interval time.Duration) {
w.resubmitIntervalCh <- interval
}
// pending returns the pending state and corresponding block.
func (w *worker) pending() (*types.Block, *state.StateDB) {
// return a snapshot to avoid contention on currentMu mutex
@ -238,8 +276,11 @@ func (w *worker) close() {
}
// newWorkLoop is a standalone goroutine to submit new mining work upon received events.
func (w *worker) newWorkLoop() {
var interrupt *int32
func (w *worker) newWorkLoop(recommitInterval time.Duration) {
var (
interrupt *int32
minInterval = recommitInterval // minimal resubmit interval specified by user.
)
timer := time.NewTimer(0)
<-timer.C // discard the initial tick
@ -251,7 +292,32 @@ func (w *worker) newWorkLoop() {
}
interrupt = new(int32)
w.newWorkCh <- &newWorkReq{interrupt: interrupt, noempty: noempty}
timer.Reset(blockRecommitInterval)
timer.Reset(recommitInterval)
}
// recalcuInterval recalculates the resubmitting interval upon feedback.
recalcuInterval := func(target float64, inc bool) {
var (
old = float64(recommitInterval.Nanoseconds())
new float64
)
if inc {
new = old*(1-intervalAdjustRatio) + intervalAdjustRatio*(target+intervalAdjustBias)
// Recap if interval is larger than the maximum time interval
if new > float64(maxRecommitInterval.Nanoseconds()) {
new = float64(maxRecommitInterval.Nanoseconds())
}
} else {
// Short circuit if the interval not larger than the minimal interval specified by user.
if recommitInterval <= minInterval {
return
}
new = old*(1-intervalAdjustRatio) + intervalAdjustRatio*(target-intervalAdjustBias)
// Recap if interval is less than the user specified minimum
if new < float64(minInterval.Nanoseconds()) {
new = float64(minInterval.Nanoseconds())
}
}
recommitInterval = time.Duration(int64(new))
}
for {
@ -269,6 +335,35 @@ func (w *worker) newWorkLoop() {
recommit(true, commitInterruptResubmit)
}
case interval := <-w.resubmitIntervalCh:
// Adjust resubmit interval explicitly by user.
if interval < minRecommitInterval {
log.Info("Recap miner recommit interval", "from", interval, "to", minRecommitInterval)
interval = minRecommitInterval
}
log.Info("Miner recommit interval update", "from", minInterval, "to", interval)
minInterval, recommitInterval = interval, interval
if w.resubmitHook != nil {
w.resubmitHook(minInterval, recommitInterval)
}
case adjust := <-w.resubmitAdjustCh:
// Adjust resubmit interval by feedback.
if adjust.inc {
before := recommitInterval
recalcuInterval(float64(recommitInterval.Nanoseconds())/adjust.ratio, true)
log.Trace("Increase miner recommit interval", "from", before, "to", recommitInterval)
} else {
before := recommitInterval
recalcuInterval(float64(minInterval.Nanoseconds()), false)
log.Trace("Decrease miner recommit interval", "from", before, "to", recommitInterval)
}
if w.resubmitHook != nil {
w.resubmitHook(minInterval, recommitInterval)
}
case <-w.exitCh:
return
}
@ -568,8 +663,18 @@ func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coin
// (3) worker recreate the mining block with any newly arrived transactions, the interrupt signal is 2.
// For the first two cases, the semi-finished work will be discarded.
// For the third case, the semi-finished work will be submitted to the consensus engine.
// TODO(rjl493456442) give feedback to newWorkLoop to adjust resubmit interval if it is too short.
if interrupt != nil && atomic.LoadInt32(interrupt) != commitInterruptNone {
// Notify resubmit loop to increase resubmitting interval due to too frequent commits.
if atomic.LoadInt32(interrupt) == commitInterruptResubmit {
ratio := float64(w.current.header.GasLimit-w.current.gasPool.Gas()) / float64(w.current.header.GasLimit)
if ratio < 0.1 {
ratio = 0.1
}
w.resubmitAdjustCh <- &intervalAdjust{
ratio: ratio,
inc: true,
}
}
return atomic.LoadInt32(interrupt) == commitInterruptNewHead
}
// If we don't have enough gas for any further transactions then we're done
@ -644,6 +749,11 @@ func (w *worker) commitTransactions(txs *types.TransactionsByPriceAndNonce, coin
}
go w.mux.Post(core.PendingLogsEvent{Logs: cpy})
}
// Notify resubmit loop to decrease resubmitting interval if current interval is larger
// than the user-specified one.
if interrupt != nil {
w.resubmitAdjustCh <- &intervalAdjust{inc: false}
}
return false
}

106
miner/worker_test.go
View file

@ -119,7 +119,7 @@ func (b *testWorkerBackend) PostChainEvents(events []interface{}) {
func newTestWorker(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) (*worker, *testWorkerBackend) {
backend := newTestWorkerBackend(t, chainConfig, engine)
backend.txPool.AddLocals(pendingTxs)
w := newWorker(chainConfig, engine, backend, new(event.TypeMux))
w := newWorker(chainConfig, engine, backend, new(event.TypeMux), time.Second)
w.setEtherbase(testBankAddress)
return w, backend
}
@ -327,7 +327,7 @@ func testRegenerateMiningBlock(t *testing.T, chainConfig *params.ChainConfig, en
}
}
b.txPool.AddLocals(newTxs)
time.Sleep(3 * time.Second)
time.Sleep(time.Second)
select {
case <-taskCh:
@ -335,3 +335,105 @@ func testRegenerateMiningBlock(t *testing.T, chainConfig *params.ChainConfig, en
t.Error("new task timeout")
}
}
func TestAdjustIntervalEthash(t *testing.T) {
testAdjustInterval(t, ethashChainConfig, ethash.NewFaker())
}
func TestAdjustIntervalClique(t *testing.T) {
testAdjustInterval(t, cliqueChainConfig, clique.New(cliqueChainConfig.Clique, ethdb.NewMemDatabase()))
}
func testAdjustInterval(t *testing.T, chainConfig *params.ChainConfig, engine consensus.Engine) {
defer engine.Close()
w, _ := newTestWorker(t, chainConfig, engine)
defer w.close()
w.skipSealHook = func(task *task) bool {
return true
}
w.fullTaskHook = func() {
time.Sleep(100 * time.Millisecond)
}
var (
progress = make(chan struct{}, 10)
result = make([]float64, 0, 10)
index = 0
start = false
)
w.resubmitHook = func(minInterval time.Duration, recommitInterval time.Duration) {
// Short circuit if interval checking hasn't started.
if !start {
return
}
var wantMinInterval, wantRecommitInterval time.Duration
switch index {
case 0:
wantMinInterval, wantRecommitInterval = 3*time.Second, 3*time.Second
case 1:
origin := float64(3 * time.Second.Nanoseconds())
estimate := origin*(1-intervalAdjustRatio) + intervalAdjustRatio*(origin/0.8+intervalAdjustBias)
wantMinInterval, wantRecommitInterval = 3*time.Second, time.Duration(int(estimate))*time.Nanosecond
case 2:
estimate := result[index-1]
min := float64(3 * time.Second.Nanoseconds())
estimate = estimate*(1-intervalAdjustRatio) + intervalAdjustRatio*(min-intervalAdjustBias)
wantMinInterval, wantRecommitInterval = 3*time.Second, time.Duration(int(estimate))*time.Nanosecond
case 3:
wantMinInterval, wantRecommitInterval = time.Second, time.Second
}
// Check interval
if minInterval != wantMinInterval {
t.Errorf("resubmit min interval mismatch want %s has %s", wantMinInterval, minInterval)
}
if recommitInterval != wantRecommitInterval {
t.Errorf("resubmit interval mismatch want %s has %s", wantRecommitInterval, recommitInterval)
}
result = append(result, float64(recommitInterval.Nanoseconds()))
index += 1
progress <- struct{}{}
}
// Ensure worker has finished initialization
for {
b := w.pendingBlock()
if b != nil && b.NumberU64() == 1 {
break
}
}
w.start()
time.Sleep(time.Second)
start = true
w.setRecommitInterval(3 * time.Second)
select {
case <-progress:
case <-time.NewTimer(time.Second).C:
t.Error("interval reset timeout")
}
w.resubmitAdjustCh <- &intervalAdjust{inc: true, ratio: 0.8}
select {
case <-progress:
case <-time.NewTimer(time.Second).C:
t.Error("interval reset timeout")
}
w.resubmitAdjustCh <- &intervalAdjust{inc: false}
select {
case <-progress:
case <-time.NewTimer(time.Second).C:
t.Error("interval reset timeout")
}
w.setRecommitInterval(500 * time.Millisecond)
select {
case <-progress:
case <-time.NewTimer(time.Second).C:
t.Error("interval reset timeout")
}
}