lotus/storage/sectors.go

319 lines
7.3 KiB
Go
Raw Normal View History

2019-10-31 18:46:53 +00:00
package storage
import (
"context"
2019-12-04 00:44:29 +00:00
"fmt"
2019-11-07 18:22:59 +00:00
"io"
"math"
2019-11-01 22:05:05 +00:00
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
2019-11-01 22:05:05 +00:00
xerrors "golang.org/x/xerrors"
2019-11-07 18:22:59 +00:00
"github.com/filecoin-project/lotus/api"
2019-12-02 12:51:16 +00:00
"github.com/filecoin-project/lotus/lib/padreader"
2019-10-31 18:46:53 +00:00
)
const NonceIncrement = math.MaxUint64
2019-11-01 13:58:48 +00:00
type sectorUpdate struct {
newState api.SectorState
id uint64
err error
nonce uint64
2019-11-01 13:58:48 +00:00
mut func(*SectorInfo)
}
2019-10-31 18:46:53 +00:00
func (u *sectorUpdate) fatal(err error) *sectorUpdate {
u.newState = api.FailedUnrecoverable
u.err = err
return u
}
func (u *sectorUpdate) error(err error) *sectorUpdate {
u.err = err
return u
}
func (u *sectorUpdate) state(m func(*SectorInfo)) *sectorUpdate {
u.mut = m
return u
}
func (u *sectorUpdate) to(newState api.SectorState) *sectorUpdate {
2019-12-09 13:32:58 +00:00
u.newState = newState
return u
}
func (u *sectorUpdate) setNonce(nc uint64) *sectorUpdate {
u.nonce = nc
return u
}
func (m *Miner) UpdateSectorState(ctx context.Context, sector uint64, snonce uint64, state api.SectorState) error {
select {
case m.sectorUpdated <- sectorUpdate{
newState: state,
nonce: snonce,
id: sector,
}:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
2019-11-17 12:12:05 +00:00
func (m *Miner) sectorStateLoop(ctx context.Context) error {
trackedSectors, err := m.ListSectors()
2019-11-17 12:12:05 +00:00
if err != nil {
log.Errorf("loading sector list: %+v", err)
2019-11-17 12:12:05 +00:00
}
go func() {
for _, si := range trackedSectors {
2019-11-17 12:12:05 +00:00
select {
case m.sectorUpdated <- sectorUpdate{
newState: si.State,
nonce: si.Nonce,
2019-11-17 12:12:05 +00:00
id: si.SectorID,
err: nil,
mut: nil,
}:
case <-ctx.Done():
log.Warn("didn't restart processing for all sectors: ", ctx.Err())
return
}
}
}()
2019-10-31 18:46:53 +00:00
{
// verify on-chain state
trackedByID := map[uint64]*SectorInfo{}
for _, si := range trackedSectors {
2019-12-02 12:51:16 +00:00
i := si
trackedByID[si.SectorID] = &i
}
curTs, err := m.api.ChainHead(ctx)
if err != nil {
return xerrors.Errorf("getting chain head: %w", err)
}
ps, err := m.api.StateMinerProvingSet(ctx, m.maddr, curTs)
if err != nil {
return xerrors.Errorf("getting miner proving set: %w", err)
}
for _, ocs := range ps {
if _, ok := trackedByID[ocs.SectorID]; ok {
continue // TODO: check state
}
// TODO: attempt recovery
log.Warnf("untracked sector %d found on chain", ocs.SectorID)
}
}
2019-11-01 13:58:48 +00:00
go func() {
defer log.Warn("quitting deal provider loop")
defer close(m.stopped)
for {
select {
case sector := <-m.sectorIncoming:
m.onSectorIncoming(sector)
case update := <-m.sectorUpdated:
m.onSectorUpdated(ctx, update)
case <-m.stop:
return
}
}
}()
2019-11-17 12:12:05 +00:00
return nil
2019-11-01 13:58:48 +00:00
}
2019-10-31 18:46:53 +00:00
2019-11-01 13:58:48 +00:00
func (m *Miner) onSectorIncoming(sector *SectorInfo) {
2019-11-07 14:45:53 +00:00
has, err := m.sectors.Has(sector.SectorID)
if err != nil {
return
}
if has {
2019-11-07 18:22:59 +00:00
log.Warnf("SealPiece called more than once for sector %d", sector.SectorID)
2019-11-07 14:45:53 +00:00
return
}
2019-11-01 13:58:48 +00:00
if err := m.sectors.Begin(sector.SectorID, sector); err != nil {
log.Errorf("sector tracking failed: %s", err)
2019-11-01 13:58:48 +00:00
return
}
2019-10-31 18:46:53 +00:00
2019-11-01 13:58:48 +00:00
go func() {
2019-11-01 22:05:05 +00:00
select {
case m.sectorUpdated <- sectorUpdate{
2019-11-06 23:09:48 +00:00
newState: api.Packing,
2019-11-01 13:58:48 +00:00
id: sector.SectorID,
2019-11-01 22:05:05 +00:00
}:
case <-m.stop:
log.Warn("failed to send incoming sector update, miner shutting down")
2019-11-01 13:58:48 +00:00
}
}()
2019-10-31 18:46:53 +00:00
}
2019-11-01 13:58:48 +00:00
func (m *Miner) onSectorUpdated(ctx context.Context, update sectorUpdate) {
log.Infof("Sector %d updated state to %s", update.id, api.SectorStates[update.newState])
2019-11-01 13:58:48 +00:00
var sector SectorInfo
err := m.sectors.Mutate(update.id, func(s *SectorInfo) error {
if update.nonce < s.Nonce {
return xerrors.Errorf("update nonce too low, ignoring (%d < %d)", update.nonce, s.Nonce)
}
if update.nonce != NonceIncrement {
s.Nonce = update.nonce
} else {
s.Nonce++ // forced update
}
2019-11-01 13:58:48 +00:00
s.State = update.newState
2019-12-04 00:44:29 +00:00
if update.err != nil {
if s.LastErr != "" {
s.LastErr += "---------\n\n"
}
s.LastErr += fmt.Sprintf("entering state %s: %+v", api.SectorStates[update.newState], update.err)
2019-12-04 00:44:29 +00:00
}
2019-11-01 22:44:55 +00:00
if update.mut != nil {
update.mut(s)
}
2019-11-01 13:58:48 +00:00
sector = *s
return nil
})
if update.err != nil {
log.Errorf("sector %d failed: %+v", update.id, update.err)
2019-10-31 18:46:53 +00:00
}
2019-11-01 13:58:48 +00:00
if err != nil {
log.Errorf("sector %d update error: %+v", update.id, err)
return
}
/*
* Empty
| |
| v
*<- Packing <- incoming
| |
| v
*<- Unsealed <--> SealFailed
| |
| v
* PreCommitting <--> PreCommitFailed
| | ^
| v |
*<- PreCommitted ------/
| |||
| vvv v--> SealCommitFailed
*<- Committing
| | ^--> CommitFailed
| v ^
*<- CommitWait ---/
| |
| v
*<- Proving
|
v
FailedUnrecoverable
UndefinedSectorState <- ¯\_()_/¯
| ^
*---------------------/
*/
2019-11-01 13:58:48 +00:00
switch update.newState {
// Happy path
2019-11-06 23:09:48 +00:00
case api.Packing:
m.handleSectorUpdate(ctx, sector, m.handlePacking)
2019-11-01 13:58:48 +00:00
case api.Unsealed:
m.handleSectorUpdate(ctx, sector, m.handleUnsealed)
2019-11-01 13:58:48 +00:00
case api.PreCommitting:
m.handleSectorUpdate(ctx, sector, m.handlePreCommitting)
2019-11-01 13:58:48 +00:00
case api.PreCommitted:
m.handleSectorUpdate(ctx, sector, m.handlePreCommitted)
2019-11-01 13:58:48 +00:00
case api.Committing:
m.handleSectorUpdate(ctx, sector, m.handleCommitting)
case api.CommitWait:
m.handleSectorUpdate(ctx, sector, m.handleCommitWait)
case api.Proving:
// TODO: track sector health / expiration
log.Infof("Proving sector %d", update.id)
// Handled failure modes
case api.SealFailed:
2019-12-09 15:42:47 +00:00
log.Warnf("sector %d entered unimplemented state 'SealFailed'", update.id)
case api.PreCommitFailed:
2019-12-09 15:42:47 +00:00
log.Warnf("sector %d entered unimplemented state 'PreCommitFailed'", update.id)
case api.SealCommitFailed:
2019-12-09 15:42:47 +00:00
log.Warnf("sector %d entered unimplemented state 'SealCommitFailed'", update.id)
case api.CommitFailed:
2019-12-09 15:42:47 +00:00
log.Warnf("sector %d entered unimplemented state 'CommitFailed'", update.id)
// Faults
case api.Faulty:
m.handleSectorUpdate(ctx, sector, m.handleFaulty)
case api.FaultReported:
m.handleSectorUpdate(ctx, sector, m.handleFaultReported)
// Fatal errors
case api.UndefinedSectorState:
log.Error("sector update with undefined state!")
case api.FailedUnrecoverable:
log.Errorf("sector %d failed unrecoverably", update.id)
default:
log.Errorf("unexpected sector update state: %d", update.newState)
2019-10-31 18:46:53 +00:00
}
}
2019-12-01 17:58:31 +00:00
func (m *Miner) AllocatePiece(size uint64) (sectorID uint64, offset uint64, err error) {
if padreader.PaddedSize(size) != size {
return 0, 0, xerrors.Errorf("cannot allocate unpadded piece")
}
2019-11-01 22:44:55 +00:00
2019-11-07 18:22:59 +00:00
sid, err := m.sb.AcquireSectorId() // TODO: Put more than one thing in a sector
if err != nil {
2019-12-01 17:58:31 +00:00
return 0, 0, xerrors.Errorf("acquiring sector ID: %w", err)
2019-11-07 18:22:59 +00:00
}
2019-12-01 17:58:31 +00:00
// offset hard-coded to 0 since we only put one thing in a sector for now
return sid, 0, nil
}
func (m *Miner) SealPiece(ctx context.Context, size uint64, r io.Reader, sectorID uint64, dealID uint64) error {
log.Infof("Seal piece for deal %d", dealID)
ppi, err := m.sb.AddPiece(size, sectorID, r, []uint64{})
2019-11-07 18:22:59 +00:00
if err != nil {
2019-12-01 17:58:31 +00:00
return xerrors.Errorf("adding piece to sector: %w", err)
2019-11-07 18:22:59 +00:00
}
2019-12-01 17:58:31 +00:00
return m.newSector(ctx, sectorID, dealID, ppi)
2019-11-07 18:22:59 +00:00
}
2019-12-01 17:58:31 +00:00
func (m *Miner) newSector(ctx context.Context, sid uint64, dealID uint64, ppi sectorbuilder.PublicPieceInfo) error {
2019-11-01 22:05:05 +00:00
si := &SectorInfo{
2019-11-01 13:58:48 +00:00
SectorID: sid,
2019-11-07 18:22:59 +00:00
Pieces: []Piece{
{
DealID: dealID,
2019-11-07 18:43:15 +00:00
Size: ppi.Size,
CommP: ppi.CommP[:],
2019-11-07 18:22:59 +00:00
},
},
2019-11-01 22:05:05 +00:00
}
select {
case m.sectorIncoming <- si:
2019-11-01 13:58:48 +00:00
return nil
case <-ctx.Done():
2019-11-01 22:05:05 +00:00
return xerrors.Errorf("failed to submit sector for sealing, queue full: %w", ctx.Err())
2019-11-01 13:58:48 +00:00
}
2019-10-31 18:46:53 +00:00
}