Merge pull request #722 from filecoin-project/feat/ec/8

Mining refactor and timing aware scheduler.

api/impl/mining.go

@@ -21,16 +21,16 @@ func (api *nodeMining) Once(ctx context.Context) (*types.Block, error) {
 	nd := api.api.node
 	ts := nd.ChainMgr.GetHeaviestTipSet()
 
-	blockGenerator := mining.NewBlockGenerator(nd.MsgPool, func(ctx context.Context, ts core.TipSet) (state.Tree, error) {
-		return nd.ChainMgr.State(ctx, ts.ToSlice())
-	}, nd.ChainMgr.Weight, core.ApplyMessages, nd.ChainMgr.PwrTableView, nd.Blockstore, nd.CborStore)
-
 	miningAddr, err := nd.MiningAddress()
 	if err != nil {
 		return nil, err
 	}
 
-	res := mining.MineOnce(ctx, mining.NewWorker(blockGenerator, miningAddr), ts)
+	worker := mining.NewDefaultWorker(nd.MsgPool, func(ctx context.Context, ts core.TipSet) (state.Tree, error) {
+		return nd.ChainMgr.State(ctx, ts.ToSlice())
+	}, nd.ChainMgr.Weight, core.ApplyMessages, nd.ChainMgr.PwrTableView, nd.Blockstore, nd.CborStore, miningAddr)
+
+	res := mining.MineOnce(ctx, mining.NewScheduler(worker), ts)
 	if res.Err != nil {
 		return nil, res.Err
 	}

mining/block_generator.go → mining/block_generate.go

@@ -1,78 +1,37 @@
 package mining
 
+// Block generation is part of the logic of the DefaultWorker.
+// 'generate' is that function that actually creates a new block from a base
+// TipSet using the DefaultWorker's many utilities.
+
 import (
 	"context"
 
-	"gx/ipfs/QmSkuaNgyGmV8c1L3cZNWcUxRJV6J3nsD96JVQPcWcwtyW/go-hamt-ipld"
 	errors "gx/ipfs/QmVmDhyTTUcQXFD1rRQ64fGLMSAoaQvNH3hwuaCFAPq2hy/errors"
-	"gx/ipfs/QmcD7SqfyQyA91TZUQ7VPRYbGarxmY7EsQewVYMuN5LNSv/go-ipfs-blockstore"
-	logging "gx/ipfs/QmcVVHfdyv15GVPk7NrxdWjh2hLVccXnoD8j2tyQShiXJb/go-log"
 
 	"github.com/filecoin-project/go-filecoin/address"
 	"github.com/filecoin-project/go-filecoin/core"
-	"github.com/filecoin-project/go-filecoin/state"
 	"github.com/filecoin-project/go-filecoin/types"
 	"github.com/filecoin-project/go-filecoin/vm"
 )
 
-var log = logging.Logger("mining")
-
-// GetStateTree is a function that gets the aggregate state tree of a TipSet. It's
-// its own function to facilitate testing.
-type GetStateTree func(context.Context, core.TipSet) (state.Tree, error)
-
-// GetWeight is a function that calculates the weight of a TipSet.  Weight is
-// expressed as two uint64s comprising a rational number.
-type GetWeight func(context.Context, core.TipSet) (uint64, uint64, error)
-
-// BlockGenerator is the primary interface for blockGenerator.
-type BlockGenerator interface {
-	Generate(context.Context, core.TipSet, types.Signature, uint64, types.Address) (*types.Block, error)
-}
-
-// NewBlockGenerator returns a new BlockGenerator.
-func NewBlockGenerator(messagePool *core.MessagePool, getStateTree GetStateTree, getWeight GetWeight, applyMessages miningApplier, powerTable core.PowerTableView, bs blockstore.Blockstore, cstore *hamt.CborIpldStore) BlockGenerator {
-	return &blockGenerator{
-		messagePool:   messagePool,
-		getStateTree:  getStateTree,
-		getWeight:     getWeight,
-		applyMessages: applyMessages,
-		powerTable:    powerTable,
-		blockstore:    bs,
-		cstore:        cstore,
-	}
-}
-
-type miningApplier func(ctx context.Context, messages []*types.SignedMessage, st state.Tree, vms vm.StorageMap, bh *types.BlockHeight) (core.ApplyMessagesResponse, error)
-
-// blockGenerator generates new blocks for inclusion in the chain.
-type blockGenerator struct {
-	messagePool   *core.MessagePool
-	getStateTree  GetStateTree
-	getWeight     GetWeight
-	applyMessages miningApplier
-	powerTable    core.PowerTableView
-	blockstore    blockstore.Blockstore
-	cstore        *hamt.CborIpldStore
-}
-
 // Generate returns a new block created from the messages in the pool.
-func (b blockGenerator) Generate(ctx context.Context, baseTipSet core.TipSet, ticket types.Signature, nullBlockCount uint64, miningAddress types.Address) (*types.Block, error) {
-	stateTree, err := b.getStateTree(ctx, baseTipSet)
+func (w *DefaultWorker) Generate(ctx context.Context, baseTipSet core.TipSet, ticket types.Signature, nullBlockCount uint64) (*types.Block, error) {
+	stateTree, err := w.getStateTree(ctx, baseTipSet)
 	if err != nil {
 		return nil, errors.Wrap(err, "get state tree")
 	}
 
-	if !b.powerTable.HasPower(ctx, stateTree, b.blockstore, miningAddress) {
-		return nil, errors.Errorf("bad miner address, miner must store files before mining: %s", miningAddress)
+	if !w.powerTable.HasPower(ctx, stateTree, w.blockstore, w.minerAddr) {
+		return nil, errors.Errorf("bad miner address, miner must store files before mining: %s", w.minerAddr)
 	}
 
-	wNum, wDenom, err := b.getWeight(ctx, baseTipSet)
+	wNum, wDenom, err := w.getWeight(ctx, baseTipSet)
 	if err != nil {
 		return nil, errors.Wrap(err, "get weight")
 	}
 
-	nonce, err := core.NextNonce(ctx, stateTree, b.messagePool, address.NetworkAddress)
+	nonce, err := core.NextNonce(ctx, stateTree, w.messagePool, address.NetworkAddress)
 	if err != nil {
 		return nil, errors.Wrap(err, "next nonce")
 	}
@@ -83,19 +42,20 @@ func (b blockGenerator) Generate(ctx context.Context, baseTipSet core.TipSet, ti
 	}
 
 	blockHeight := baseHeight + nullBlockCount + 1
-	rewardMsg := types.NewMessage(address.NetworkAddress, miningAddress, nonce, types.NewAttoFILFromFIL(1000), "", nil)
+	rewardMsg := types.NewMessage(address.NetworkAddress, w.minerAddr, nonce, types.NewAttoFILFromFIL(1000), "", nil)
 	srewardMsg := &types.SignedMessage{
 		Message:   *rewardMsg,
 		Signature: nil,
 	}
 
-	pending := b.messagePool.Pending()
+	pending := w.messagePool.Pending()
 	messages := make([]*types.SignedMessage, len(pending)+1)
 	messages[0] = srewardMsg // Reward message must come first since this is a part of the consensus rules.
-	copy(messages[1:], core.OrderMessagesByNonce(b.messagePool.Pending()))
+	copy(messages[1:], core.OrderMessagesByNonce(w.messagePool.Pending()))
+
+	vms := vm.NewStorageMap(w.blockstore)
+	res, err := w.applyMessages(ctx, messages, stateTree, vms, types.NewBlockHeight(blockHeight))
 
-	vms := vm.NewStorageMap(b.blockstore)
-	res, err := b.applyMessages(ctx, messages, stateTree, vms, types.NewBlockHeight(blockHeight))
 	if err != nil {
 		return nil, errors.Wrap(err, "generate apply messages")
 	}
@@ -111,7 +71,7 @@ func (b blockGenerator) Generate(ctx context.Context, baseTipSet core.TipSet, ti
 	}
 
 	next := &types.Block{
-		Miner:             miningAddress,
+		Miner:             w.minerAddr,
 		Height:            types.Uint64(blockHeight),
 		Messages:          res.SuccessfulMessages,
 		MessageReceipts:   receipts,
@@ -136,7 +96,7 @@ func (b blockGenerator) Generate(ctx context.Context, baseTipSet core.TipSet, ti
 	for _, msg := range res.SuccessfulMessages {
 		mc, err := msg.Cid()
 		if err == nil {
-			b.messagePool.Remove(mc)
+			w.messagePool.Remove(mc)
 		}
 	}
 
@@ -148,7 +108,7 @@ func (b blockGenerator) Generate(ctx context.Context, baseTipSet core.TipSet, ti
 		log.Infof("permanent ApplyMessage failure, [%S]", mc.String())
 		// Intentionally not handling error case, since it just means we won't be able to remove from pool.
 		if err == nil {
-			b.messagePool.Remove(mc)
+			w.messagePool.Remove(mc)
 		}
 	}