simplify BlockMiner.

This commit is contained in:
Raúl Kripalani 2021-05-19 10:51:32 +01:00
parent 0cfef0fdbb
commit 2a71c47397
4 changed files with 46 additions and 37 deletions

View File

@ -2,7 +2,6 @@ package kit
import ( import (
"context" "context"
"fmt"
"sync/atomic" "sync/atomic"
"testing" "testing"
"time" "time"
@ -11,42 +10,40 @@ import (
"github.com/filecoin-project/lotus/miner" "github.com/filecoin-project/lotus/miner"
) )
// BlockMiner is a utility that makes a test Miner Mine blocks on a timer. // BlockMiner is a utility that makes a test miner Mine blocks on a timer.
type BlockMiner struct { type BlockMiner struct {
ctx context.Context
t *testing.T t *testing.T
miner TestStorageNode miner TestStorageNode
blocktime time.Duration
done chan struct{}
Mine int64 nextNulls int64
Nulls int64 stopCh chan chan struct{}
} }
func NewBlockMiner(ctx context.Context, t *testing.T, miner TestStorageNode, blocktime time.Duration) *BlockMiner { func NewBlockMiner(t *testing.T, miner TestStorageNode) *BlockMiner {
return &BlockMiner{ return &BlockMiner{
ctx: ctx,
t: t, t: t,
miner: miner, miner: miner,
blocktime: blocktime, stopCh: make(chan chan struct{}),
Mine: int64(1),
done: make(chan struct{}),
} }
} }
func (bm *BlockMiner) MineBlocks() { func (bm *BlockMiner) MineBlocks(ctx context.Context, blocktime time.Duration) {
time.Sleep(time.Second) time.Sleep(time.Second)
go func() { go func() {
defer close(bm.done) for {
for atomic.LoadInt64(&bm.Mine) == 1 {
select { select {
case <-bm.ctx.Done(): case <-time.After(blocktime):
case <-ctx.Done():
return
case ch := <-bm.stopCh:
close(ch)
close(bm.stopCh)
return return
case <-time.After(bm.blocktime):
} }
nulls := atomic.SwapInt64(&bm.Nulls, 0) nulls := atomic.SwapInt64(&bm.nextNulls, 0)
if err := bm.miner.MineOne(bm.ctx, miner.MineReq{ if err := bm.miner.MineOne(ctx, miner.MineReq{
InjectNulls: abi.ChainEpoch(nulls), InjectNulls: abi.ChainEpoch(nulls),
Done: func(bool, abi.ChainEpoch, error) {}, Done: func(bool, abi.ChainEpoch, error) {},
}); err != nil { }); err != nil {
@ -56,8 +53,20 @@ func (bm *BlockMiner) MineBlocks() {
}() }()
} }
func (bm *BlockMiner) Stop() { // InjectNulls injects the specified amount of null rounds in the next
atomic.AddInt64(&bm.Mine, -1) // mining rounds.
fmt.Println("shutting down mining") func (bm *BlockMiner) InjectNulls(rounds abi.ChainEpoch) {
<-bm.done atomic.AddInt64(&bm.nextNulls, int64(rounds))
}
// Stop stops the block miner.
func (bm *BlockMiner) Stop() {
bm.t.Log("shutting down mining")
if _, ok := <-bm.stopCh; ok {
// already stopped
return
}
ch := make(chan struct{})
bm.stopCh <- ch
<-ch
} }

View File

@ -279,8 +279,8 @@ func ConnectAndStartMining(t *testing.T, blocktime time.Duration, client *impl.F
} }
time.Sleep(time.Second) time.Sleep(time.Second)
blockMiner := NewBlockMiner(ctx, t, miner, blocktime) blockMiner := NewBlockMiner(t, miner)
blockMiner.MineBlocks() blockMiner.MineBlocks(ctx, blocktime)
return &DealsScaffold{ return &DealsScaffold{
Ctx: ctx, Ctx: ctx,

View File

@ -28,8 +28,8 @@ func StartOneNodeOneMiner(ctx context.Context, t *testing.T, blocktime time.Dura
} }
// Start mining blocks // Start mining blocks
bm := NewBlockMiner(ctx, t, miner, blocktime) bm := NewBlockMiner(t, miner)
bm.MineBlocks() bm.MineBlocks(ctx, blocktime)
t.Cleanup(bm.Stop) t.Cleanup(bm.Stop)
// Get the full node's wallet address // Get the full node's wallet address
@ -64,8 +64,8 @@ func StartTwoNodesOneMiner(ctx context.Context, t *testing.T, blocktime time.Dur
} }
// Start mining blocks // Start mining blocks
bm := NewBlockMiner(ctx, t, miner, blocktime) bm := NewBlockMiner(t, miner)
bm.MineBlocks() bm.MineBlocks(ctx, blocktime)
t.Cleanup(bm.Stop) t.Cleanup(bm.Stop)
// Send some funds to register the second node // Send some funds to register the second node

View File

@ -3,7 +3,6 @@ package itests
import ( import (
"context" "context"
"fmt" "fmt"
"sync/atomic"
"testing" "testing"
"time" "time"
@ -52,8 +51,9 @@ func TestPaymentChannelsAPI(t *testing.T) {
} }
// start mining blocks // start mining blocks
bm := kit.NewBlockMiner(ctx, t, miner, 5*time.Millisecond) bm := kit.NewBlockMiner(t, miner)
bm.MineBlocks() bm.MineBlocks(ctx, 5*time.Millisecond)
t.Cleanup(bm.Stop)
// send some funds to register the receiver // send some funds to register the receiver
receiverAddr, err := paymentReceiver.WalletNew(ctx, types.KTSecp256k1) receiverAddr, err := paymentReceiver.WalletNew(ctx, types.KTSecp256k1)
@ -278,7 +278,7 @@ func waitForBlocks(ctx context.Context, t *testing.T, bm *kit.BlockMiner, paymen
} }
// Add a batch of null blocks // Add a batch of null blocks
atomic.StoreInt64(&bm.Nulls, int64(size-1)) bm.InjectNulls(abi.ChainEpoch(size - 1))
// Add a real block // Add a real block
m, err := paymentReceiver.MpoolPushMessage(ctx, &types.Message{ m, err := paymentReceiver.MpoolPushMessage(ctx, &types.Message{