lotus/storage/miner.go

136 lines
3.9 KiB
Go
Raw Normal View History

2019-07-29 18:57:23 +00:00
package storage
import (
"context"
"sync"
2019-07-29 18:57:23 +00:00
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
2019-11-08 20:11:56 +00:00
"github.com/ipfs/go-datastore/namespace"
2019-07-29 18:57:23 +00:00
logging "github.com/ipfs/go-log"
"github.com/libp2p/go-libp2p-core/host"
2019-07-29 18:57:23 +00:00
"github.com/pkg/errors"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/address"
"github.com/filecoin-project/lotus/chain/events"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
2019-11-08 20:11:56 +00:00
"github.com/filecoin-project/lotus/lib/sectorbuilder"
"github.com/filecoin-project/lotus/lib/statestore"
2019-07-29 18:57:23 +00:00
)
var log = logging.Logger("storageminer")
const PoStConfidence = 3
2019-07-29 18:57:23 +00:00
type Miner struct {
api storageMinerApi
events *events.Events
2019-11-01 13:58:48 +00:00
h host.Host
2019-07-29 18:57:23 +00:00
2019-11-01 13:58:48 +00:00
maddr address.Address
2019-07-29 18:57:23 +00:00
worker address.Address
2019-11-01 13:58:48 +00:00
// PoSt
postLk sync.Mutex
schedPost uint64
2019-07-29 18:57:23 +00:00
2019-11-01 13:58:48 +00:00
// Sealing
2019-11-07 18:43:15 +00:00
sb *sectorbuilder.SectorBuilder
2019-11-01 13:58:48 +00:00
sectors *statestore.StateStore
2019-11-07 18:43:15 +00:00
tktFn TicketFn
2019-09-19 16:17:49 +00:00
2019-11-01 13:58:48 +00:00
sectorIncoming chan *SectorInfo
sectorUpdated chan sectorUpdate
stop chan struct{}
stopped chan struct{}
2019-07-29 18:57:23 +00:00
}
type storageMinerApi interface {
// I think I want this... but this is tricky
//ReadState(ctx context.Context, addr address.Address) (????, error)
// Call a read only method on actors (no interaction with the chain required)
StateCall(ctx context.Context, msg *types.Message, ts *types.TipSet) (*types.MessageReceipt, error)
StateMinerWorker(context.Context, address.Address, *types.TipSet) (address.Address, error)
StateMinerProvingPeriodEnd(context.Context, address.Address, *types.TipSet) (uint64, error)
2019-11-08 18:15:13 +00:00
StateMinerSectors(context.Context, address.Address, *types.TipSet) ([]*api.ChainSectorInfo, error)
StateMinerProvingSet(context.Context, address.Address, *types.TipSet) ([]*api.ChainSectorInfo, error)
StateMinerSectorSize(context.Context, address.Address, *types.TipSet) (uint64, error)
StateWaitMsg(context.Context, cid.Cid) (*api.MsgWait, error)
2019-07-29 18:57:23 +00:00
2019-09-19 16:17:49 +00:00
MpoolPushMessage(context.Context, *types.Message) (*types.SignedMessage, error)
2019-07-29 18:57:23 +00:00
ChainHead(context.Context) (*types.TipSet, error)
2019-09-18 11:01:52 +00:00
ChainNotify(context.Context) (<-chan []*store.HeadChange, error)
ChainGetRandomness(context.Context, types.TipSetKey, []*types.Ticket, int) ([]byte, error)
ChainGetTipSetByHeight(context.Context, uint64, *types.TipSet) (*types.TipSet, error)
ChainGetBlockMessages(context.Context, cid.Cid) (*api.BlockMessages, error)
2019-08-08 04:24:49 +00:00
WalletSign(context.Context, address.Address, []byte) (*types.Signature, error)
2019-08-08 04:24:49 +00:00
WalletBalance(context.Context, address.Address) (types.BigInt, error)
2019-08-08 17:29:23 +00:00
WalletHas(context.Context, address.Address) (bool, error)
2019-07-29 18:57:23 +00:00
}
2019-11-07 18:22:59 +00:00
func NewMiner(api storageMinerApi, addr address.Address, h host.Host, ds datastore.Batching, sb *sectorbuilder.SectorBuilder, tktFn TicketFn) (*Miner, error) {
2019-07-29 18:57:23 +00:00
return &Miner{
api: api,
2019-07-29 18:57:23 +00:00
maddr: addr,
h: h,
2019-11-07 18:43:15 +00:00
sb: sb,
2019-11-07 18:22:59 +00:00
tktFn: tktFn,
2019-11-01 13:58:48 +00:00
sectors: statestore.New(namespace.Wrap(ds, datastore.NewKey("/sectors"))),
2019-11-01 22:44:55 +00:00
sectorIncoming: make(chan *SectorInfo),
sectorUpdated: make(chan sectorUpdate),
stop: make(chan struct{}),
stopped: make(chan struct{}),
2019-07-29 18:57:23 +00:00
}, nil
}
func (m *Miner) Run(ctx context.Context) error {
if err := m.runPreflightChecks(ctx); err != nil {
return errors.Wrap(err, "miner preflight checks failed")
}
m.events = events.NewEvents(ctx, m.api)
2019-09-26 20:57:20 +00:00
go m.beginPosting(ctx)
2019-11-01 13:58:48 +00:00
go m.sectorStateLoop(ctx)
2019-07-29 18:57:23 +00:00
return nil
}
2019-11-01 13:58:48 +00:00
func (m *Miner) Stop(ctx context.Context) error {
close(m.stop)
select {
case <-m.stopped:
return nil
case <-ctx.Done():
return ctx.Err()
}
2019-07-29 18:57:23 +00:00
}
func (m *Miner) runPreflightChecks(ctx context.Context) error {
worker, err := m.api.StateMinerWorker(ctx, m.maddr, nil)
2019-07-29 18:57:23 +00:00
if err != nil {
return err
}
m.worker = worker
2019-08-08 17:29:23 +00:00
has, err := m.api.WalletHas(ctx, worker)
if err != nil {
return errors.Wrap(err, "failed to check wallet for worker key")
}
if !has {
return errors.New("key for worker not found in local wallet")
2019-07-29 18:57:23 +00:00
}
log.Infof("starting up miner %s, worker addr %s", m.maddr, m.worker)
2019-07-29 18:57:23 +00:00
return nil
}