lotus/itests/kit/blockminer.go
2022-09-27 15:34:01 +00:00

374 lines
9.8 KiB
Go

package kit
import (
"bytes"
"context"
"fmt"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/filecoin-project/go-bitfield"
"github.com/filecoin-project/go-jsonrpc"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/builtin"
minertypes "github.com/filecoin-project/go-state-types/builtin/v8/miner"
"github.com/filecoin-project/go-state-types/dline"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/miner"
)
// BlockMiner is a utility that makes a test miner Mine blocks on a timer.
type BlockMiner struct {
t *testing.T
miner *TestMiner
nextNulls int64
pause chan struct{}
unpause chan struct{}
wg sync.WaitGroup
cancel context.CancelFunc
}
func NewBlockMiner(t *testing.T, miner *TestMiner) *BlockMiner {
return &BlockMiner{
t: t,
miner: miner,
cancel: func() {},
unpause: make(chan struct{}),
pause: make(chan struct{}),
}
}
type partitionTracker struct {
partitions []api.Partition
posted bitfield.BitField
}
func newPartitionTracker(ctx context.Context, dlIdx uint64, bm *BlockMiner) *partitionTracker {
dlines, err := bm.miner.FullNode.StateMinerDeadlines(ctx, bm.miner.ActorAddr, types.EmptyTSK)
require.NoError(bm.t, err)
dl := dlines[dlIdx]
parts, err := bm.miner.FullNode.StateMinerPartitions(ctx, bm.miner.ActorAddr, dlIdx, types.EmptyTSK)
require.NoError(bm.t, err)
return &partitionTracker{
partitions: parts,
posted: dl.PostSubmissions,
}
}
func (p *partitionTracker) count(t *testing.T) uint64 {
pCnt, err := p.posted.Count()
require.NoError(t, err)
return pCnt
}
func (p *partitionTracker) done(t *testing.T) bool {
return uint64(len(p.partitions)) == p.count(t)
}
func (p *partitionTracker) recordIfPost(t *testing.T, bm *BlockMiner, msg *types.Message) (ret bool) {
defer func() {
ret = p.done(t)
}()
if !(msg.To == bm.miner.ActorAddr) {
return
}
if msg.Method != builtin.MethodsMiner.SubmitWindowedPoSt {
return
}
params := minertypes.SubmitWindowedPoStParams{}
require.NoError(t, params.UnmarshalCBOR(bytes.NewReader(msg.Params)))
for _, part := range params.Partitions {
p.posted.Set(part.Index)
}
return
}
func (bm *BlockMiner) forcePoSt(ctx context.Context, ts *types.TipSet, dlinfo *dline.Info) {
tracker := newPartitionTracker(ctx, dlinfo.Index, bm)
if !tracker.done(bm.t) { // need to wait for post
bm.t.Logf("expect %d partitions proved but only see %d", len(tracker.partitions), tracker.count(bm.t))
poolEvts, err := bm.miner.FullNode.MpoolSub(ctx) //subscribe before checking pending so we don't miss any events
require.NoError(bm.t, err)
// First check pending messages we'll mine this epoch
msgs, err := bm.miner.FullNode.MpoolPending(ctx, types.EmptyTSK)
require.NoError(bm.t, err)
for _, msg := range msgs {
if tracker.recordIfPost(bm.t, bm, &msg.Message) {
fmt.Printf("found post in mempool pending\n")
}
}
// Account for included but not yet executed messages
for _, bc := range ts.Cids() {
msgs, err := bm.miner.FullNode.ChainGetBlockMessages(ctx, bc)
require.NoError(bm.t, err)
for _, msg := range msgs.BlsMessages {
if tracker.recordIfPost(bm.t, bm, msg) {
fmt.Printf("found post in message of prev tipset\n")
}
}
for _, msg := range msgs.SecpkMessages {
if tracker.recordIfPost(bm.t, bm, &msg.Message) {
fmt.Printf("found post in message of prev tipset\n")
}
}
}
// post not yet in mpool, wait for it
if !tracker.done(bm.t) {
bm.t.Logf("post missing from mpool, block mining suspended until it arrives")
POOL:
for {
bm.t.Logf("mpool event wait loop at block height %d, ts: %s", ts.Height(), ts.Key())
select {
case <-ctx.Done():
return
case evt := <-poolEvts:
bm.t.Logf("pool event: %d", evt.Type)
if evt.Type == api.MpoolAdd {
bm.t.Logf("incoming message %v", evt.Message)
if tracker.recordIfPost(bm.t, bm, &evt.Message.Message) {
fmt.Printf("found post in mempool evt\n")
break POOL
}
}
}
}
bm.t.Logf("done waiting on mpool")
}
}
}
// Like MineBlocks but refuses to mine until the window post scheduler has wdpost messages in the mempool
// and everything shuts down if a post fails. It also enforces that every block mined succeeds
func (bm *BlockMiner) MineBlocksMustPost(ctx context.Context, blocktime time.Duration) {
time.Sleep(time.Second)
// wrap context in a cancellable context.
ctx, bm.cancel = context.WithCancel(ctx)
bm.wg.Add(1)
go func() {
defer bm.wg.Done()
ts, err := bm.miner.FullNode.ChainHead(ctx)
require.NoError(bm.t, err)
wait := make(chan bool)
chg, err := bm.miner.FullNode.ChainNotify(ctx)
require.NoError(bm.t, err)
// read current out
curr := <-chg
require.Equal(bm.t, ts.Height(), curr[0].Val.Height(), "failed sanity check: are multiple miners mining with must post?")
for {
select {
case <-time.After(blocktime):
case <-ctx.Done():
return
}
nulls := atomic.SwapInt64(&bm.nextNulls, 0)
// Wake up and figure out if we are at the end of an active deadline
ts, err := bm.miner.FullNode.ChainHead(ctx)
require.NoError(bm.t, err)
dlinfo, err := bm.miner.FullNode.StateMinerProvingDeadline(ctx, bm.miner.ActorAddr, ts.Key())
require.NoError(bm.t, err)
if ts.Height()+1+abi.ChainEpoch(nulls) >= dlinfo.Last() { // Next block brings us past the last epoch in dline, we need to wait for miner to post
bm.forcePoSt(ctx, ts, dlinfo)
}
var target abi.ChainEpoch
reportSuccessFn := func(success bool, epoch abi.ChainEpoch, err error) {
// if api shuts down before mining, we may get an error which we should probably just ignore
// (fixing it will require rewriting most of the mining loop)
if err != nil && !strings.Contains(err.Error(), "websocket connection closed") && !api.ErrorIsIn(err, []error{&jsonrpc.RPCConnectionError{}}) {
require.NoError(bm.t, err)
}
target = epoch
wait <- success
}
var success bool
for i := int64(0); !success; i++ {
err = bm.miner.MineOne(ctx, miner.MineReq{
InjectNulls: abi.ChainEpoch(nulls + i),
Done: reportSuccessFn,
})
select {
case success = <-wait:
case <-ctx.Done():
return
}
if !success {
// if we are mining a new null block and it brings us past deadline boundary we need to wait for miner to post
if ts.Height()+1+abi.ChainEpoch(nulls+i) >= dlinfo.Last() {
bm.forcePoSt(ctx, ts, dlinfo)
}
}
}
// Wait until it shows up on the given full nodes ChainHead
// TODO this replicates a flaky condition from MineUntil,
// it would be better to use api to wait for sync,
// but currently this is a bit difficult
// and flaky failure is easy to debug and retry
nloops := 200
for i := 0; i < nloops; i++ {
ts, err := bm.miner.FullNode.ChainHead(ctx)
require.NoError(bm.t, err)
if ts.Height() == target {
break
}
require.NotEqual(bm.t, i, nloops-1, "block never managed to sync to node")
time.Sleep(time.Millisecond * 10)
}
switch {
case err == nil: // wrap around
case ctx.Err() != nil: // context fired.
return
default: // log error
bm.t.Error(err)
}
}
}()
}
func (bm *BlockMiner) MineBlocks(ctx context.Context, blocktime time.Duration) {
time.Sleep(time.Second)
// wrap context in a cancellable context.
ctx, bm.cancel = context.WithCancel(ctx)
bm.wg.Add(1)
go func() {
defer bm.wg.Done()
for {
select {
case <-bm.pause:
select {
case <-bm.unpause:
case <-ctx.Done():
return
}
case <-ctx.Done():
return
default:
}
select {
case <-time.After(blocktime):
case <-ctx.Done():
return
}
nulls := atomic.SwapInt64(&bm.nextNulls, 0)
err := bm.miner.MineOne(ctx, miner.MineReq{
InjectNulls: abi.ChainEpoch(nulls),
Done: func(bool, abi.ChainEpoch, error) {},
})
switch {
case err == nil: // wrap around
case ctx.Err() != nil: // context fired.
return
default: // log error
bm.t.Error(err)
}
}
}()
}
// InjectNulls injects the specified amount of null rounds in the next
// mining rounds.
func (bm *BlockMiner) InjectNulls(rounds abi.ChainEpoch) {
atomic.AddInt64(&bm.nextNulls, int64(rounds))
}
// Pause compels the miner to wait for a signal to restart
func (bm *BlockMiner) Pause() {
bm.pause <- struct{}{}
}
// Restart continues mining after a pause. This will hang if called before pause
func (bm *BlockMiner) Restart() {
bm.unpause <- struct{}{}
}
func (bm *BlockMiner) MineUntilBlock(ctx context.Context, fn *TestFullNode, cb func(abi.ChainEpoch)) {
for i := 0; i < 1000; i++ {
var (
success bool
err error
epoch abi.ChainEpoch
wait = make(chan struct{})
)
doneFn := func(win bool, ep abi.ChainEpoch, e error) {
success = win
err = e
epoch = ep
wait <- struct{}{}
}
mineErr := bm.miner.MineOne(ctx, miner.MineReq{Done: doneFn})
require.NoError(bm.t, mineErr)
<-wait
require.NoError(bm.t, err)
if success {
// Wait until it shows up on the given full nodes ChainHead
nloops := 200
for i := 0; i < nloops; i++ {
ts, err := fn.ChainHead(ctx)
require.NoError(bm.t, err)
if ts.Height() == epoch {
break
}
require.NotEqual(bm.t, i, nloops-1, "block never managed to sync to node")
time.Sleep(time.Millisecond * 10)
}
if cb != nil {
cb(epoch)
}
return
}
bm.t.Log("did not Mine block, trying again", i)
}
bm.t.Fatal("failed to Mine 1000 times in a row...")
}
// Stop stops the block miner.
func (bm *BlockMiner) Stop() {
bm.t.Log("shutting down mining")
bm.cancel()
bm.wg.Wait()
if bm.unpause != nil {
close(bm.unpause)
bm.unpause = nil
}
if bm.pause != nil {
close(bm.pause)
bm.pause = nil
}
}