376 lines
10 KiB
Go
376 lines
10 KiB
Go
package sealing
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/ipfs/go-cid"
|
|
"golang.org/x/xerrors"
|
|
|
|
"github.com/filecoin-project/go-address"
|
|
"github.com/filecoin-project/go-bitfield"
|
|
"github.com/filecoin-project/go-state-types/abi"
|
|
"github.com/filecoin-project/go-state-types/big"
|
|
"github.com/filecoin-project/go-state-types/builtin"
|
|
"github.com/filecoin-project/go-state-types/builtin/v8/miner"
|
|
"github.com/filecoin-project/go-state-types/dline"
|
|
"github.com/filecoin-project/lotus/api"
|
|
lminer "github.com/filecoin-project/lotus/chain/actors/builtin/miner"
|
|
"github.com/filecoin-project/lotus/chain/types"
|
|
"github.com/filecoin-project/lotus/node/config"
|
|
"github.com/filecoin-project/lotus/node/modules/dtypes"
|
|
)
|
|
|
|
type TerminateBatcherApi interface {
|
|
StateSectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tsk types.TipSetKey) (*lminer.SectorLocation, error)
|
|
MpoolPushMessage(context.Context, *types.Message, *api.MessageSendSpec) (*types.SignedMessage, error)
|
|
StateMinerInfo(context.Context, address.Address, types.TipSetKey) (api.MinerInfo, error)
|
|
StateMinerProvingDeadline(context.Context, address.Address, types.TipSetKey) (*dline.Info, error)
|
|
StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tsk types.TipSetKey) ([]api.Partition, error)
|
|
|
|
// Address selector
|
|
WalletBalance(context.Context, address.Address) (types.BigInt, error)
|
|
WalletHas(context.Context, address.Address) (bool, error)
|
|
StateAccountKey(context.Context, address.Address, types.TipSetKey) (address.Address, error)
|
|
StateLookupID(context.Context, address.Address, types.TipSetKey) (address.Address, error)
|
|
}
|
|
|
|
type TerminateBatcher struct {
|
|
api TerminateBatcherApi
|
|
maddr address.Address
|
|
mctx context.Context
|
|
addrSel AddressSelector
|
|
feeCfg config.MinerFeeConfig
|
|
getConfig dtypes.GetSealingConfigFunc
|
|
|
|
todo map[lminer.SectorLocation]*bitfield.BitField // MinerSectorLocation -> BitField
|
|
|
|
waiting map[abi.SectorNumber][]chan cid.Cid
|
|
|
|
notify, stop, stopped chan struct{}
|
|
force chan chan *cid.Cid
|
|
lk sync.Mutex
|
|
}
|
|
|
|
func NewTerminationBatcher(mctx context.Context, maddr address.Address, api TerminateBatcherApi, addrSel AddressSelector, feeCfg config.MinerFeeConfig, getConfig dtypes.GetSealingConfigFunc) *TerminateBatcher {
|
|
b := &TerminateBatcher{
|
|
api: api,
|
|
maddr: maddr,
|
|
mctx: mctx,
|
|
addrSel: addrSel,
|
|
feeCfg: feeCfg,
|
|
getConfig: getConfig,
|
|
|
|
todo: map[lminer.SectorLocation]*bitfield.BitField{},
|
|
waiting: map[abi.SectorNumber][]chan cid.Cid{},
|
|
|
|
notify: make(chan struct{}, 1),
|
|
force: make(chan chan *cid.Cid),
|
|
stop: make(chan struct{}),
|
|
stopped: make(chan struct{}),
|
|
}
|
|
|
|
go b.run()
|
|
|
|
return b
|
|
}
|
|
|
|
func (b *TerminateBatcher) run() {
|
|
var forceRes chan *cid.Cid
|
|
var lastMsg *cid.Cid
|
|
|
|
for {
|
|
if forceRes != nil {
|
|
forceRes <- lastMsg
|
|
forceRes = nil
|
|
}
|
|
lastMsg = nil
|
|
|
|
cfg, err := b.getConfig()
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher getconfig error", "error", err)
|
|
}
|
|
|
|
var sendAboveMax, sendAboveMin bool
|
|
select {
|
|
case <-b.stop:
|
|
close(b.stopped)
|
|
return
|
|
case <-b.notify:
|
|
sendAboveMax = true
|
|
case <-time.After(cfg.TerminateBatchWait):
|
|
sendAboveMin = true
|
|
case fr := <-b.force: // user triggered
|
|
forceRes = fr
|
|
}
|
|
|
|
lastMsg, err = b.processBatch(sendAboveMax, sendAboveMin)
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher processBatch error", "error", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (b *TerminateBatcher) processBatch(notif, after bool) (*cid.Cid, error) {
|
|
dl, err := b.api.StateMinerProvingDeadline(b.mctx, b.maddr, types.EmptyTSK)
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("getting proving deadline info failed: %w", err)
|
|
}
|
|
|
|
cfg, err := b.getConfig()
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("getting sealing config: %W", err)
|
|
}
|
|
|
|
b.lk.Lock()
|
|
defer b.lk.Unlock()
|
|
params := miner.TerminateSectorsParams{}
|
|
|
|
var total uint64
|
|
for loc, sectors := range b.todo {
|
|
n, err := sectors.Count()
|
|
if err != nil {
|
|
log.Errorw("TerminateBatcher: failed to count sectors to terminate", "deadline", loc.Deadline, "partition", loc.Partition, "error", err)
|
|
continue
|
|
}
|
|
|
|
// don't send terminations for currently challenged sectors
|
|
if loc.Deadline == (dl.Index+1)%miner.WPoStPeriodDeadlines || // not in next (in case the terminate message takes a while to get on chain)
|
|
loc.Deadline == dl.Index || // not in current
|
|
(loc.Deadline+1)%miner.WPoStPeriodDeadlines == dl.Index { // not in previous
|
|
continue
|
|
}
|
|
|
|
if n < 1 {
|
|
log.Warnw("TerminateBatcher: zero sectors in bucket", "deadline", loc.Deadline, "partition", loc.Partition)
|
|
continue
|
|
}
|
|
|
|
toTerminate, err := sectors.Copy()
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher: copy sectors bitfield", "deadline", loc.Deadline, "partition", loc.Partition, "error", err)
|
|
continue
|
|
}
|
|
|
|
ps, err := b.api.StateMinerPartitions(b.mctx, b.maddr, loc.Deadline, types.EmptyTSK)
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher: getting miner partitions", "deadline", loc.Deadline, "partition", loc.Partition, "error", err)
|
|
continue
|
|
}
|
|
|
|
toTerminate, err = bitfield.IntersectBitField(ps[loc.Partition].LiveSectors, toTerminate)
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher: intersecting liveSectors and toTerminate bitfields", "deadline", loc.Deadline, "partition", loc.Partition, "error", err)
|
|
continue
|
|
}
|
|
|
|
if total+n > uint64(miner.AddressedSectorsMax) {
|
|
n = uint64(miner.AddressedSectorsMax) - total
|
|
|
|
toTerminate, err = toTerminate.Slice(0, n)
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher: slice toTerminate bitfield", "deadline", loc.Deadline, "partition", loc.Partition, "error", err)
|
|
continue
|
|
}
|
|
|
|
s, err := bitfield.SubtractBitField(*sectors, toTerminate)
|
|
if err != nil {
|
|
log.Warnw("TerminateBatcher: sectors-toTerminate", "deadline", loc.Deadline, "partition", loc.Partition, "error", err)
|
|
continue
|
|
}
|
|
*sectors = s
|
|
}
|
|
|
|
total += n
|
|
|
|
params.Terminations = append(params.Terminations, miner.TerminationDeclaration{
|
|
Deadline: loc.Deadline,
|
|
Partition: loc.Partition,
|
|
Sectors: toTerminate,
|
|
})
|
|
|
|
if total >= uint64(miner.AddressedSectorsMax) || total >= cfg.TerminateBatchMax {
|
|
break
|
|
}
|
|
|
|
if len(params.Terminations) >= miner.DeclarationsMax {
|
|
break
|
|
}
|
|
}
|
|
|
|
if len(params.Terminations) == 0 {
|
|
return nil, nil // nothing to do
|
|
}
|
|
|
|
if notif && total < cfg.TerminateBatchMax {
|
|
return nil, nil
|
|
}
|
|
|
|
if after && total < cfg.TerminateBatchMin {
|
|
return nil, nil
|
|
}
|
|
|
|
enc := new(bytes.Buffer)
|
|
if err := params.MarshalCBOR(enc); err != nil {
|
|
return nil, xerrors.Errorf("couldn't serialize TerminateSectors params: %w", err)
|
|
}
|
|
|
|
mi, err := b.api.StateMinerInfo(b.mctx, b.maddr, types.EmptyTSK)
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("couldn't get miner info: %w", err)
|
|
}
|
|
|
|
from, _, err := b.addrSel.AddressFor(b.mctx, b.api, mi, api.TerminateSectorsAddr, big.Int(b.feeCfg.MaxTerminateGasFee), big.Int(b.feeCfg.MaxTerminateGasFee))
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("no good address found: %w", err)
|
|
}
|
|
|
|
mcid, err := sendMsg(b.mctx, b.api, from, b.maddr, builtin.MethodsMiner.TerminateSectors, big.Zero(), big.Int(b.feeCfg.MaxTerminateGasFee), enc.Bytes())
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("sending message failed: %w", err)
|
|
}
|
|
log.Infow("Sent TerminateSectors message", "cid", mcid, "from", from, "terminations", len(params.Terminations))
|
|
|
|
for _, t := range params.Terminations {
|
|
delete(b.todo, lminer.SectorLocation{
|
|
Deadline: t.Deadline,
|
|
Partition: t.Partition,
|
|
})
|
|
|
|
err := t.Sectors.ForEach(func(sn uint64) error {
|
|
for _, ch := range b.waiting[abi.SectorNumber(sn)] {
|
|
ch <- mcid // buffered
|
|
}
|
|
delete(b.waiting, abi.SectorNumber(sn))
|
|
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, xerrors.Errorf("sectors foreach: %w", err)
|
|
}
|
|
}
|
|
|
|
return &mcid, nil
|
|
}
|
|
|
|
// register termination, wait for batch message, return message CID
|
|
// can return cid.Undef,true if the sector is already terminated on-chain
|
|
func (b *TerminateBatcher) AddTermination(ctx context.Context, s abi.SectorID) (mcid cid.Cid, terminated bool, err error) {
|
|
maddr, err := address.NewIDAddress(uint64(s.Miner))
|
|
if err != nil {
|
|
return cid.Undef, false, err
|
|
}
|
|
|
|
loc, err := b.api.StateSectorPartition(ctx, maddr, s.Number, types.EmptyTSK)
|
|
if err != nil {
|
|
return cid.Undef, false, xerrors.Errorf("getting sector location: %w", err)
|
|
}
|
|
if loc == nil {
|
|
return cid.Undef, false, xerrors.New("sector location not found")
|
|
}
|
|
|
|
{
|
|
// check if maybe already terminated
|
|
parts, err := b.api.StateMinerPartitions(ctx, maddr, loc.Deadline, types.EmptyTSK)
|
|
if err != nil {
|
|
return cid.Cid{}, false, xerrors.Errorf("getting partitions: %w", err)
|
|
}
|
|
live, err := parts[loc.Partition].LiveSectors.IsSet(uint64(s.Number))
|
|
if err != nil {
|
|
return cid.Cid{}, false, xerrors.Errorf("checking if sector is in live set: %w", err)
|
|
}
|
|
if !live {
|
|
// already terminated
|
|
return cid.Undef, true, nil
|
|
}
|
|
}
|
|
|
|
b.lk.Lock()
|
|
bf, ok := b.todo[*loc]
|
|
if !ok {
|
|
n := bitfield.New()
|
|
bf = &n
|
|
b.todo[*loc] = bf
|
|
}
|
|
bf.Set(uint64(s.Number))
|
|
|
|
sent := make(chan cid.Cid, 1)
|
|
b.waiting[s.Number] = append(b.waiting[s.Number], sent)
|
|
|
|
select {
|
|
case b.notify <- struct{}{}:
|
|
default: // already have a pending notification, don't need more
|
|
}
|
|
b.lk.Unlock()
|
|
|
|
select {
|
|
case c := <-sent:
|
|
return c, false, nil
|
|
case <-ctx.Done():
|
|
return cid.Undef, false, ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (b *TerminateBatcher) Flush(ctx context.Context) (*cid.Cid, error) {
|
|
resCh := make(chan *cid.Cid, 1)
|
|
select {
|
|
case b.force <- resCh:
|
|
select {
|
|
case res := <-resCh:
|
|
return res, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (b *TerminateBatcher) Pending(ctx context.Context) ([]abi.SectorID, error) {
|
|
b.lk.Lock()
|
|
defer b.lk.Unlock()
|
|
|
|
mid, err := address.IDFromAddress(b.maddr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
res := make([]abi.SectorID, 0)
|
|
for _, bf := range b.todo {
|
|
err := bf.ForEach(func(id uint64) error {
|
|
res = append(res, abi.SectorID{
|
|
Miner: abi.ActorID(mid),
|
|
Number: abi.SectorNumber(id),
|
|
})
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
sort.Slice(res, func(i, j int) bool {
|
|
if res[i].Miner != res[j].Miner {
|
|
return res[i].Miner < res[j].Miner
|
|
}
|
|
|
|
return res[i].Number < res[j].Number
|
|
})
|
|
|
|
return res, nil
|
|
}
|
|
|
|
func (b *TerminateBatcher) Stop(ctx context.Context) error {
|
|
close(b.stop)
|
|
|
|
select {
|
|
case <-b.stopped:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|