lotus/extern/storage-sealing/precommit_batch.go

489 lines
12 KiB
Go
Raw Normal View History

2021-05-18 14:54:55 +00:00
package sealing
import (
"bytes"
"context"
"sort"
"sync"
"time"
2021-10-01 14:23:21 +00:00
"github.com/filecoin-project/go-state-types/network"
2021-05-18 14:54:55 +00:00
"github.com/ipfs/go-cid"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
miner0 "github.com/filecoin-project/specs-actors/actors/builtin/miner"
miner5 "github.com/filecoin-project/specs-actors/v5/actors/builtin/miner"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
2021-05-18 14:54:55 +00:00
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"github.com/filecoin-project/lotus/chain/actors/policy"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/extern/storage-sealing/sealiface"
2021-06-08 13:43:43 +00:00
"github.com/filecoin-project/lotus/node/config"
2021-05-18 14:54:55 +00:00
)
2021-06-09 15:18:09 +00:00
//go:generate go run github.com/golang/mock/mockgen -destination=mocks/mock_precommit_batcher.go -package=mocks . PreCommitBatcherApi
2021-05-18 14:54:55 +00:00
type PreCommitBatcherApi interface {
SendMsg(ctx context.Context, from, to address.Address, method abi.MethodNum, value, maxFee abi.TokenAmount, params []byte) (cid.Cid, error)
StateMinerInfo(context.Context, address.Address, TipSetToken) (miner.MinerInfo, error)
StateMinerAvailableBalance(context.Context, address.Address, TipSetToken) (big.Int, error)
2021-05-18 14:54:55 +00:00
ChainHead(ctx context.Context) (TipSetToken, abi.ChainEpoch, error)
ChainBaseFee(context.Context, TipSetToken) (abi.TokenAmount, error)
2021-10-01 14:23:21 +00:00
StateNetworkVersion(ctx context.Context, tok TipSetToken) (network.Version, error)
2021-05-18 14:54:55 +00:00
}
2021-05-18 15:21:10 +00:00
type preCommitEntry struct {
deposit abi.TokenAmount
pci *miner0.SectorPreCommitInfo
}
2021-05-18 14:54:55 +00:00
type PreCommitBatcher struct {
api PreCommitBatcherApi
maddr address.Address
mctx context.Context
addrSel AddrSel
2021-06-08 13:43:43 +00:00
feeCfg config.MinerFeeConfig
2021-05-18 14:54:55 +00:00
getConfig GetSealingConfigFunc
cutoffs map[abi.SectorNumber]time.Time
todo map[abi.SectorNumber]*preCommitEntry
waiting map[abi.SectorNumber][]chan sealiface.PreCommitBatchRes
2021-05-18 14:54:55 +00:00
notify, stop, stopped chan struct{}
force chan chan []sealiface.PreCommitBatchRes
2021-05-18 14:54:55 +00:00
lk sync.Mutex
}
2021-06-08 13:43:43 +00:00
func NewPreCommitBatcher(mctx context.Context, maddr address.Address, api PreCommitBatcherApi, addrSel AddrSel, feeCfg config.MinerFeeConfig, getConfig GetSealingConfigFunc) *PreCommitBatcher {
2021-05-18 14:54:55 +00:00
b := &PreCommitBatcher{
api: api,
maddr: maddr,
mctx: mctx,
addrSel: addrSel,
feeCfg: feeCfg,
getConfig: getConfig,
cutoffs: map[abi.SectorNumber]time.Time{},
todo: map[abi.SectorNumber]*preCommitEntry{},
waiting: map[abi.SectorNumber][]chan sealiface.PreCommitBatchRes{},
2021-05-18 14:54:55 +00:00
notify: make(chan struct{}, 1),
force: make(chan chan []sealiface.PreCommitBatchRes),
2021-05-18 14:54:55 +00:00
stop: make(chan struct{}),
stopped: make(chan struct{}),
}
go b.run()
return b
}
func (b *PreCommitBatcher) run() {
var forceRes chan []sealiface.PreCommitBatchRes
var lastRes []sealiface.PreCommitBatchRes
2021-05-18 14:54:55 +00:00
cfg, err := b.getConfig()
if err != nil {
panic(err)
}
timer := time.NewTimer(b.batchWait(cfg.PreCommitBatchWait, cfg.PreCommitBatchSlack))
2021-05-18 14:54:55 +00:00
for {
if forceRes != nil {
forceRes <- lastRes
2021-05-18 14:54:55 +00:00
forceRes = nil
}
lastRes = nil
2021-05-18 14:54:55 +00:00
2021-06-23 16:30:32 +00:00
var sendAboveMax bool
2021-05-18 14:54:55 +00:00
select {
case <-b.stop:
close(b.stopped)
return
case <-b.notify:
sendAboveMax = true
2021-06-30 08:56:40 +00:00
case <-timer.C:
2021-06-23 16:30:32 +00:00
// do nothing
2021-05-18 14:54:55 +00:00
case fr := <-b.force: // user triggered
forceRes = fr
}
var err error
2021-06-23 16:30:32 +00:00
lastRes, err = b.maybeStartBatch(sendAboveMax)
2021-05-18 14:54:55 +00:00
if err != nil {
2021-05-18 15:37:52 +00:00
log.Warnw("PreCommitBatcher processBatch error", "error", err)
2021-05-18 14:54:55 +00:00
}
2021-06-30 08:56:40 +00:00
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
timer.Reset(b.batchWait(cfg.PreCommitBatchWait, cfg.PreCommitBatchSlack))
2021-05-18 14:54:55 +00:00
}
}
2021-06-30 08:56:40 +00:00
func (b *PreCommitBatcher) batchWait(maxWait, slack time.Duration) time.Duration {
2021-05-18 14:54:55 +00:00
now := time.Now()
b.lk.Lock()
defer b.lk.Unlock()
if len(b.todo) == 0 {
2021-06-30 08:56:40 +00:00
return maxWait
}
var cutoff time.Time
2021-05-18 14:54:55 +00:00
for sn := range b.todo {
sectorCutoff := b.cutoffs[sn]
if cutoff.IsZero() || (!sectorCutoff.IsZero() && sectorCutoff.Before(cutoff)) {
cutoff = sectorCutoff
2021-05-18 14:54:55 +00:00
}
}
for sn := range b.waiting {
sectorCutoff := b.cutoffs[sn]
if cutoff.IsZero() || (!sectorCutoff.IsZero() && sectorCutoff.Before(cutoff)) {
cutoff = sectorCutoff
2021-05-18 14:54:55 +00:00
}
}
if cutoff.IsZero() {
2021-06-30 08:56:40 +00:00
return maxWait
2021-05-18 14:54:55 +00:00
}
cutoff = cutoff.Add(-slack)
if cutoff.Before(now) {
2021-06-30 08:56:40 +00:00
return time.Nanosecond // can't return 0
2021-05-18 14:54:55 +00:00
}
wait := cutoff.Sub(now)
2021-05-18 14:54:55 +00:00
if wait > maxWait {
wait = maxWait
}
2021-06-30 08:56:40 +00:00
return wait
2021-05-18 14:54:55 +00:00
}
2021-06-23 16:30:32 +00:00
func (b *PreCommitBatcher) maybeStartBatch(notif bool) ([]sealiface.PreCommitBatchRes, error) {
2021-05-18 14:54:55 +00:00
b.lk.Lock()
defer b.lk.Unlock()
total := len(b.todo)
if total == 0 {
return nil, nil // nothing to do
}
cfg, err := b.getConfig()
if err != nil {
return nil, xerrors.Errorf("getting config: %w", err)
}
if notif && total < cfg.MaxPreCommitBatch {
return nil, nil
}
2021-10-01 14:23:21 +00:00
tok, _, err := b.api.ChainHead(b.mctx)
if err != nil {
return nil, err
}
2021-10-01 14:23:21 +00:00
bf, err := b.api.ChainBaseFee(b.mctx, tok)
if err != nil {
return nil, xerrors.Errorf("couldn't get base fee: %w", err)
}
2021-10-01 14:23:21 +00:00
// TODO: Drop this once nv14 has come and gone
nv, err := b.api.StateNetworkVersion(b.mctx, tok)
if err != nil {
return nil, xerrors.Errorf("couldn't get network version: %w", err)
}
individual := false
if !cfg.BatchPreCommitAboveBaseFee.Equals(big.Zero()) && bf.LessThan(cfg.BatchPreCommitAboveBaseFee) && nv >= network.Version14 {
individual = true
}
// todo support multiple batches
var res []sealiface.PreCommitBatchRes
if !individual {
2021-10-01 14:23:21 +00:00
res, err = b.processBatch(cfg, tok, bf, nv)
} else {
res, err = b.processIndividually(cfg)
}
if err != nil && len(res) == 0 {
return nil, err
}
for _, r := range res {
if err != nil {
r.Error = err.Error()
}
for _, sn := range r.Sectors {
for _, ch := range b.waiting[sn] {
ch <- r // buffered
}
delete(b.waiting, sn)
delete(b.todo, sn)
delete(b.cutoffs, sn)
}
}
return res, nil
}
func (b *PreCommitBatcher) processIndividually(cfg sealiface.Config) ([]sealiface.PreCommitBatchRes, error) {
mi, err := b.api.StateMinerInfo(b.mctx, b.maddr, nil)
if err != nil {
return nil, xerrors.Errorf("couldn't get miner info: %w", err)
}
avail := types.TotalFilecoinInt
if cfg.CollateralFromMinerBalance && !cfg.DisableCollateralFallback {
avail, err = b.api.StateMinerAvailableBalance(b.mctx, b.maddr, nil)
if err != nil {
return nil, xerrors.Errorf("getting available miner balance: %w", err)
}
avail = big.Sub(avail, cfg.AvailableBalanceBuffer)
if avail.LessThan(big.Zero()) {
avail = big.Zero()
}
}
var res []sealiface.PreCommitBatchRes
for sn, info := range b.todo {
r := sealiface.PreCommitBatchRes{
Sectors: []abi.SectorNumber{sn},
}
mcid, err := b.processSingle(cfg, mi, &avail, info)
if err != nil {
r.Error = err.Error()
} else {
r.Msg = &mcid
}
res = append(res, r)
}
return res, nil
}
func (b *PreCommitBatcher) processSingle(cfg sealiface.Config, mi miner.MinerInfo, avail *abi.TokenAmount, params *preCommitEntry) (cid.Cid, error) {
enc := new(bytes.Buffer)
if err := params.pci.MarshalCBOR(enc); err != nil {
2021-09-30 16:46:25 +00:00
return cid.Undef, xerrors.Errorf("marshaling precommit params: %w", err)
}
deposit := params.deposit
if cfg.CollateralFromMinerBalance {
c := big.Sub(deposit, *avail)
*avail = big.Sub(*avail, deposit)
deposit = c
if deposit.LessThan(big.Zero()) {
deposit = big.Zero()
}
if (*avail).LessThan(big.Zero()) {
*avail = big.Zero()
}
}
goodFunds := big.Add(deposit, big.Int(b.feeCfg.MaxPreCommitGasFee))
from, _, err := b.addrSel(b.mctx, mi, api.PreCommitAddr, goodFunds, deposit)
if err != nil {
2021-09-30 16:46:25 +00:00
return cid.Undef, xerrors.Errorf("no good address to send precommit message from: %w", err)
}
mcid, err := b.api.SendMsg(b.mctx, from, b.maddr, miner.Methods.PreCommitSector, deposit, big.Int(b.feeCfg.MaxPreCommitGasFee), enc.Bytes())
if err != nil {
return cid.Undef, xerrors.Errorf("pushing message to mpool: %w", err)
}
return mcid, nil
}
2021-10-01 14:23:21 +00:00
func (b *PreCommitBatcher) processBatch(cfg sealiface.Config, tok TipSetToken, bf abi.TokenAmount, nv network.Version) ([]sealiface.PreCommitBatchRes, error) {
params := miner5.PreCommitSectorBatchParams{}
2021-05-18 15:21:10 +00:00
deposit := big.Zero()
var res sealiface.PreCommitBatchRes
2021-05-18 15:21:10 +00:00
2021-05-18 14:54:55 +00:00
for _, p := range b.todo {
2021-05-19 18:34:50 +00:00
if len(params.Sectors) >= cfg.MaxPreCommitBatch {
log.Infow("precommit batch full")
break
}
res.Sectors = append(res.Sectors, p.pci.SectorNumber)
2021-05-27 19:54:31 +00:00
params.Sectors = append(params.Sectors, *p.pci)
2021-05-18 15:21:10 +00:00
deposit = big.Add(deposit, p.deposit)
2021-05-18 14:54:55 +00:00
}
2021-05-18 14:54:55 +00:00
enc := new(bytes.Buffer)
if err := params.MarshalCBOR(enc); err != nil {
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("couldn't serialize PreCommitSectorBatchParams: %w", err)
2021-05-18 14:54:55 +00:00
}
mi, err := b.api.StateMinerInfo(b.mctx, b.maddr, nil)
if err != nil {
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("couldn't get miner info: %w", err)
2021-05-18 14:54:55 +00:00
}
2021-06-08 13:43:43 +00:00
maxFee := b.feeCfg.MaxPreCommitBatchGasFee.FeeForSectors(len(params.Sectors))
2021-10-01 14:23:21 +00:00
aggFeeRaw, err := policy.AggregatePreCommitNetworkFee(nv, len(params.Sectors), bf)
if err != nil {
log.Errorf("getting aggregate precommit network fee: %s", err)
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("getting aggregate precommit network fee: %s", err)
}
aggFee := big.Div(big.Mul(aggFeeRaw, aggFeeNum), aggFeeDen)
needFunds := big.Add(deposit, aggFee)
needFunds, err = collateralSendAmount(b.mctx, b.api, b.maddr, cfg, needFunds)
if err != nil {
return []sealiface.PreCommitBatchRes{res}, err
}
goodFunds := big.Add(maxFee, needFunds)
2021-05-18 15:21:10 +00:00
from, _, err := b.addrSel(b.mctx, mi, api.PreCommitAddr, goodFunds, deposit)
2021-05-18 14:54:55 +00:00
if err != nil {
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("no good address found: %w", err)
2021-05-18 14:54:55 +00:00
}
2021-10-01 14:23:21 +00:00
mcid, err := b.api.SendMsg(b.mctx, from, b.maddr, miner.Methods.PreCommitSectorBatch, needFunds, maxFee, enc.Bytes())
2021-05-18 14:54:55 +00:00
if err != nil {
return []sealiface.PreCommitBatchRes{res}, xerrors.Errorf("sending message failed: %w", err)
2021-05-18 14:54:55 +00:00
}
res.Msg = &mcid
2021-05-18 14:54:55 +00:00
2021-06-09 15:18:09 +00:00
log.Infow("Sent PreCommitSectorBatch message", "cid", mcid, "from", from, "sectors", len(b.todo))
2021-05-18 14:54:55 +00:00
return []sealiface.PreCommitBatchRes{res}, nil
2021-05-18 14:54:55 +00:00
}
// register PreCommit, wait for batch message, return message CID
func (b *PreCommitBatcher) AddPreCommit(ctx context.Context, s SectorInfo, deposit abi.TokenAmount, in *miner0.SectorPreCommitInfo) (res sealiface.PreCommitBatchRes, err error) {
2021-05-18 14:54:55 +00:00
_, curEpoch, err := b.api.ChainHead(b.mctx)
if err != nil {
log.Errorf("getting chain head: %s", err)
return sealiface.PreCommitBatchRes{}, err
2021-05-18 14:54:55 +00:00
}
sn := s.SectorNumber
b.lk.Lock()
b.cutoffs[sn] = getPreCommitCutoff(curEpoch, s)
2021-05-18 15:21:10 +00:00
b.todo[sn] = &preCommitEntry{
deposit: deposit,
pci: in,
}
2021-05-18 14:54:55 +00:00
sent := make(chan sealiface.PreCommitBatchRes, 1)
2021-05-18 14:54:55 +00:00
b.waiting[sn] = append(b.waiting[sn], 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, nil
case <-ctx.Done():
return sealiface.PreCommitBatchRes{}, ctx.Err()
2021-05-18 14:54:55 +00:00
}
}
func (b *PreCommitBatcher) Flush(ctx context.Context) ([]sealiface.PreCommitBatchRes, error) {
resCh := make(chan []sealiface.PreCommitBatchRes, 1)
2021-05-18 14:54:55 +00:00
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 *PreCommitBatcher) 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 _, s := range b.todo {
res = append(res, abi.SectorID{
Miner: abi.ActorID(mid),
2021-05-18 15:21:10 +00:00
Number: s.pci.SectorNumber,
2021-05-18 14:54:55 +00:00
})
}
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 *PreCommitBatcher) Stop(ctx context.Context) error {
close(b.stop)
select {
case <-b.stopped:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
2021-06-09 16:26:20 +00:00
// TODO: If this returned epochs, it would make testing much easier
func getPreCommitCutoff(curEpoch abi.ChainEpoch, si SectorInfo) time.Time {
cutoffEpoch := si.TicketEpoch + policy.MaxPreCommitRandomnessLookback
for _, p := range si.Pieces {
if p.DealInfo == nil {
continue
}
startEpoch := p.DealInfo.DealSchedule.StartEpoch
if startEpoch < cutoffEpoch {
cutoffEpoch = startEpoch
}
}
if cutoffEpoch <= curEpoch {
return time.Now()
}
return time.Now().Add(time.Duration(cutoffEpoch-curEpoch) * time.Duration(build.BlockDelaySecs) * time.Second)
}