cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge branch 'next' into refactor/lib/blockstore

Browse Source
This commit is contained in:
Łukasz Magiera 2021-03-01 19:04:40 +01:00 committed by GitHub
commit 8bd5173a54
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
34 changed files with 4435 additions and 611 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
api/api_full.go
View File

@ -32,6 +32,8 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
//go:generate go run github.com/golang/mock/mockgen -destination=mocks/mock_full.go -package=mocks . FullNode
// ChainIO abstracts operations for accessing raw IPLD objects.
type ChainIO interface {
ChainReadObj(context.Context, cid.Cid) ([]byte, error)

2
api/api_storage.go
View File

@ -36,7 +36,7 @@ type StorageMiner interface {
MiningBase(context.Context) (*types.TipSet, error)
// Temp api for testing
PledgeSector(context.Context) error
PledgeSector(context.Context) (abi.SectorID, error)
// Get the status of a given sector by ID
SectorsStatus(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (SectorInfo, error)

4
api/apistruct/struct.go
View File

@ -304,7 +304,7 @@ type StorageMinerStruct struct {
MarketPendingDeals func(ctx context.Context) (api.PendingDealInfo, error) `perm:"write"`
MarketPublishPendingDeals func(ctx context.Context) error `perm:"admin"`
PledgeSector func(context.Context) error `perm:"write"`
PledgeSector func(context.Context) (abi.SectorID, error) `perm:"write"`
SectorsStatus func(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (api.SectorInfo, error) `perm:"read"`
SectorsList func(context.Context) ([]abi.SectorNumber, error) `perm:"read"`
@ -1274,7 +1274,7 @@ func (c *StorageMinerStruct) ActorAddressConfig(ctx context.Context) (api.Addres
return c.Internal.ActorAddressConfig(ctx)
}
func (c *StorageMinerStruct) PledgeSector(ctx context.Context) error {
func (c *StorageMinerStruct) PledgeSector(ctx context.Context) (abi.SectorID, error) {
return c.Internal.PledgeSector(ctx)
}

2972
api/mocks/mock_full.go Normal file
View File

File diff suppressed because it is too large Load Diff

19
api/test/tape.go
View File

@ -6,7 +6,6 @@ import (
"testing"
"time"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/network"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
@ -75,23 +74,9 @@ func testTapeFix(t *testing.T, b APIBuilder, blocktime time.Duration, after bool
<-done
}()
err = miner.PledgeSector(ctx)
sid, err := miner.PledgeSector(ctx)
require.NoError(t, err)
// Wait till done.
var sectorNo abi.SectorNumber
for {
s, err := miner.SectorsList(ctx) // Note - the test builder doesn't import genesis sectors into FSM
require.NoError(t, err)
fmt.Printf("Sectors: %d\n", len(s))
if len(s) == 1 {
sectorNo = s[0]
break
}
build.Clock.Sleep(100 * time.Millisecond)
}
fmt.Printf("All sectors is fsm\n")
// If before, we expect the precommit to fail
@ -103,7 +88,7 @@ func testTapeFix(t *testing.T, b APIBuilder, blocktime time.Duration, after bool
}
for {
st, err := miner.SectorsStatus(ctx, sectorNo, false)
st, err := miner.SectorsStatus(ctx, sid.Number, false)
require.NoError(t, err)
if st.State == successState {
break

2
api/test/window_post.go
View File

@ -162,7 +162,7 @@ func pledgeSectors(t *testing.T, ctx context.Context, miner TestStorageNode, n,
log.Errorf("WAIT")
}
log.Errorf("PLEDGING %d", i)
err := miner.PledgeSector(ctx)
_, err := miner.PledgeSector(ctx)
require.NoError(t, err)
}

1
build/tools.go
View File

@ -3,6 +3,7 @@
package build
import (
_ "github.com/golang/mock/mockgen"
_ "github.com/GeertJohan/go.rice/rice"
_ "github.com/whyrusleeping/bencher"
)

5
cli/chain.go
View File

@ -642,7 +642,10 @@ var chainListCmd = &cli.Command{
gasUsed += r.GasUsed
}
fmt.Printf("\ttipset: \t%d msgs, %d / %d (%0.2f%%)\n", len(msgs), gasUsed, limitSum, 100*float64(gasUsed)/float64(limitSum))
gasEfficiency := 100 * float64(gasUsed) / float64(limitSum)
gasCapacity := 100 * float64(limitSum) / float64(build.BlockGasLimit)
fmt.Printf("\ttipset: \t%d msgs, %d (%0.2f%%) / %d (%0.2f%%)\n", len(msgs), gasUsed, gasEfficiency, limitSum, gasCapacity)
}
fmt.Println()
}

13
cli/cmd.go
View File

@ -207,6 +207,19 @@ func GetFullNodeAPI(ctx *cli.Context) (api.FullNode, jsonrpc.ClientCloser, error
return client.NewFullNodeRPC(ctx.Context, addr, headers)
}
func GetFullNodeServices(ctx *cli.Context) (ServicesAPI, error) {
if tn, ok := ctx.App.Metadata["test-services"]; ok {
return tn.(ServicesAPI), nil
}
api, c, err := GetFullNodeAPI(ctx)
if err != nil {
return nil, err
}
return &ServicesImpl{api: api, closer: c}, nil
}
type GetStorageMinerOptions struct {
PreferHttp bool
}

144
cli/send.go
View File

@ -1,22 +1,17 @@
package cli
import (
"bytes"
"context"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"reflect"
"github.com/urfave/cli/v2"
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors/builtin"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/types"
)
@ -72,15 +67,16 @@ var sendCmd = &cli.Command{
return ShowHelp(cctx, fmt.Errorf("'send' expects two arguments, target and amount"))
}
api, closer, err := GetFullNodeAPI(cctx)
srv, err := GetFullNodeServices(cctx)
if err != nil {
return err
}
defer closer()
defer srv.Close() //nolint:errcheck
ctx := ReqContext(cctx)
var params SendParams
toAddr, err := address.NewFromString(cctx.Args().Get(0))
params.To, err = address.NewFromString(cctx.Args().Get(0))
if err != nil {
return ShowHelp(cctx, fmt.Errorf("failed to parse target address: %w", err))
}
@ -89,123 +85,75 @@ var sendCmd = &cli.Command{
if err != nil {
return ShowHelp(cctx, fmt.Errorf("failed to parse amount: %w", err))
}
params.Val = abi.TokenAmount(val)
var fromAddr address.Address
if from := cctx.String("from"); from == "" {
defaddr, err := api.WalletDefaultAddress(ctx)
if err != nil {
return err
}
fromAddr = defaddr
} else {
if from := cctx.String("from"); from != "" {
addr, err := address.NewFromString(from)
if err != nil {
return err
}
fromAddr = addr
params.From = addr
}
gp, err := types.BigFromString(cctx.String("gas-premium"))
if err != nil {
return err
}
gfc, err := types.BigFromString(cctx.String("gas-feecap"))
if err != nil {
return err
if cctx.IsSet("gas-premium") {
gp, err := types.BigFromString(cctx.String("gas-premium"))
if err != nil {
return err
}
params.GasPremium = &gp
}
method := abi.MethodNum(cctx.Uint64("method"))
if cctx.IsSet("gas-feecap") {
gfc, err := types.BigFromString(cctx.String("gas-feecap"))
if err != nil {
return err
}
params.GasFeeCap = &gfc
}
if cctx.IsSet("gas-limit") {
limit := cctx.Int64("gas-limit")
params.GasLimit = &limit
}
params.Method = abi.MethodNum(cctx.Uint64("method"))
var params []byte
if cctx.IsSet("params-json") {
decparams, err := decodeTypedParams(ctx, api, toAddr, method, cctx.String("params-json"))
decparams, err := srv.DecodeTypedParamsFromJSON(ctx, params.To, params.Method, cctx.String("params-json"))
if err != nil {
return fmt.Errorf("failed to decode json params: %w", err)
}
params = decparams
params.Params = decparams
}
if cctx.IsSet("params-hex") {
if params != nil {
if params.Params != nil {
return fmt.Errorf("can only specify one of 'params-json' and 'params-hex'")
}
decparams, err := hex.DecodeString(cctx.String("params-hex"))
if err != nil {
return fmt.Errorf("failed to decode hex params: %w", err)
}
params = decparams
params.Params = decparams
}
msg := &types.Message{
From: fromAddr,
To: toAddr,
Value: types.BigInt(val),
GasPremium: gp,
GasFeeCap: gfc,
GasLimit: cctx.Int64("gas-limit"),
Method: method,
Params: params,
}
if !cctx.Bool("force") {
// Funds insufficient check
fromBalance, err := api.WalletBalance(ctx, msg.From)
if err != nil {
return err
}
totalCost := types.BigAdd(types.BigMul(msg.GasFeeCap, types.NewInt(uint64(msg.GasLimit))), msg.Value)
if fromBalance.LessThan(totalCost) {
fmt.Printf("WARNING: From balance %s less than total cost %s\n", types.FIL(fromBalance), types.FIL(totalCost))
return fmt.Errorf("--force must be specified for this action to have an effect; you have been warned")
}
}
params.Force = cctx.Bool("force")
if cctx.IsSet("nonce") {
msg.Nonce = cctx.Uint64("nonce")
sm, err := api.WalletSignMessage(ctx, fromAddr, msg)
if err != nil {
return err
}
_, err = api.MpoolPush(ctx, sm)
if err != nil {
return err
}
fmt.Println(sm.Cid())
} else {
sm, err := api.MpoolPushMessage(ctx, msg, nil)
if err != nil {
return err
}
fmt.Println(sm.Cid())
n := cctx.Uint64("nonce")
params.Nonce = &n
}
msgCid, err := srv.Send(ctx, params)
if err != nil {
if errors.Is(err, ErrSendBalanceTooLow) {
return fmt.Errorf("--force must be specified for this action to have an effect; you have been warned: %w", err)
}
return xerrors.Errorf("executing send: %w", err)
}
fmt.Fprintf(cctx.App.Writer, "%s\n", msgCid)
return nil
},
}
func decodeTypedParams(ctx context.Context, fapi api.FullNode, to address.Address, method abi.MethodNum, paramstr string) ([]byte, error) {
act, err := fapi.StateGetActor(ctx, to, types.EmptyTSK)
if err != nil {
return nil, err
}
methodMeta, found := stmgr.MethodsMap[act.Code][method]
if !found {
return nil, fmt.Errorf("method %d not found on actor %s", method, act.Code)
}
p := reflect.New(methodMeta.Params.Elem()).Interface().(cbg.CBORMarshaler)
if err := json.Unmarshal([]byte(paramstr), p); err != nil {
return nil, fmt.Errorf("unmarshaling input into params type: %w", err)
}
buf := new(bytes.Buffer)
if err := p.MarshalCBOR(buf); err != nil {
return nil, err
}
return buf.Bytes(), nil
}

128
cli/send_test.go Normal file
View File

@ -0,0 +1,128 @@
package cli
import (
"bytes"
"errors"
"testing"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
types "github.com/filecoin-project/lotus/chain/types"
gomock "github.com/golang/mock/gomock"
cid "github.com/ipfs/go-cid"
"github.com/stretchr/testify/assert"
ucli "github.com/urfave/cli/v2"
)
var arbtCid = (&types.Message{
From: mustAddr(address.NewIDAddress(2)),
To: mustAddr(address.NewIDAddress(1)),
Value: types.NewInt(1000),
}).Cid()
func mustAddr(a address.Address, err error) address.Address {
if err != nil {
panic(err)
}
return a
}
func newMockApp(t *testing.T, cmd *ucli.Command) (*ucli.App, *MockServicesAPI, *bytes.Buffer, func()) {
app := ucli.NewApp()
app.Commands = ucli.Commands{cmd}
app.Setup()
mockCtrl := gomock.NewController(t)
mockSrvcs := NewMockServicesAPI(mockCtrl)
app.Metadata["test-services"] = mockSrvcs
buf := &bytes.Buffer{}
app.Writer = buf
return app, mockSrvcs, buf, mockCtrl.Finish
}
func TestSendCLI(t *testing.T) {
oneFil := abi.TokenAmount(types.MustParseFIL("1"))
t.Run("simple", func(t *testing.T) {
app, mockSrvcs, buf, done := newMockApp(t, sendCmd)
defer done()
gomock.InOrder(
mockSrvcs.EXPECT().Send(gomock.Any(), SendParams{
To: mustAddr(address.NewIDAddress(1)),
Val: oneFil,
}).Return(arbtCid, nil),
mockSrvcs.EXPECT().Close(),
)
err := app.Run([]string{"lotus", "send", "t01", "1"})
assert.NoError(t, err)
assert.EqualValues(t, arbtCid.String()+"\n", buf.String())
})
t.Run("ErrSendBalanceTooLow", func(t *testing.T) {
app, mockSrvcs, _, done := newMockApp(t, sendCmd)
defer done()
gomock.InOrder(
mockSrvcs.EXPECT().Send(gomock.Any(), SendParams{
To: mustAddr(address.NewIDAddress(1)),
Val: oneFil,
}).Return(cid.Undef, ErrSendBalanceTooLow),
mockSrvcs.EXPECT().Close(),
)
err := app.Run([]string{"lotus", "send", "t01", "1"})
assert.ErrorIs(t, err, ErrSendBalanceTooLow)
})
t.Run("generic-err-is-forwarded", func(t *testing.T) {
app, mockSrvcs, _, done := newMockApp(t, sendCmd)
defer done()
errMark := errors.New("something")
gomock.InOrder(
mockSrvcs.EXPECT().Send(gomock.Any(), SendParams{
To: mustAddr(address.NewIDAddress(1)),
Val: oneFil,
}).Return(cid.Undef, errMark),
mockSrvcs.EXPECT().Close(),
)
err := app.Run([]string{"lotus", "send", "t01", "1"})
assert.ErrorIs(t, err, errMark)
})
t.Run("from-specific", func(t *testing.T) {
app, mockSrvcs, buf, done := newMockApp(t, sendCmd)
defer done()
gomock.InOrder(
mockSrvcs.EXPECT().Send(gomock.Any(), SendParams{
To: mustAddr(address.NewIDAddress(1)),
From: mustAddr(address.NewIDAddress(2)),
Val: oneFil,
}).Return(arbtCid, nil),
mockSrvcs.EXPECT().Close(),
)
err := app.Run([]string{"lotus", "send", "--from=t02", "t01", "1"})
assert.NoError(t, err)
assert.EqualValues(t, arbtCid.String()+"\n", buf.String())
})
t.Run("nonce-specific", func(t *testing.T) {
app, mockSrvcs, buf, done := newMockApp(t, sendCmd)
defer done()
zero := uint64(0)
gomock.InOrder(
mockSrvcs.EXPECT().Send(gomock.Any(), SendParams{
To: mustAddr(address.NewIDAddress(1)),
Nonce: &zero,
Val: oneFil,
}).Return(arbtCid, nil),
mockSrvcs.EXPECT().Close(),
)
err := app.Run([]string{"lotus", "send", "--nonce=0", "t01", "1"})
assert.NoError(t, err)
assert.EqualValues(t, arbtCid.String()+"\n", buf.String())
})
}

166
cli/services.go Normal file
View File

@ -0,0 +1,166 @@
package cli
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-jsonrpc"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/stmgr"
types "github.com/filecoin-project/lotus/chain/types"
cid "github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
"golang.org/x/xerrors"
)
//go:generate go run github.com/golang/mock/mockgen -destination=servicesmock_test.go -package=cli -self_package github.com/filecoin-project/lotus/cli . ServicesAPI
type ServicesAPI interface {
// Sends executes a send given SendParams
Send(ctx context.Context, params SendParams) (cid.Cid, error)
// DecodeTypedParamsFromJSON takes in information needed to identify a method and converts JSON
// parameters to bytes of their CBOR encoding
DecodeTypedParamsFromJSON(ctx context.Context, to address.Address, method abi.MethodNum, paramstr string) ([]byte, error)
// Close ends the session of services and disconnects from RPC, using Services after Close is called
// most likely will result in an error
// Should not be called concurrently
Close() error
}
type ServicesImpl struct {
api api.FullNode
closer jsonrpc.ClientCloser
}
func (s *ServicesImpl) Close() error {
if s.closer == nil {
return xerrors.Errorf("Services already closed")
}
s.closer()
s.closer = nil
return nil
}
func (s *ServicesImpl) DecodeTypedParamsFromJSON(ctx context.Context, to address.Address, method abi.MethodNum, paramstr string) ([]byte, error) {
act, err := s.api.StateGetActor(ctx, to, types.EmptyTSK)
if err != nil {
return nil, err
}
methodMeta, found := stmgr.MethodsMap[act.Code][method]
if !found {
return nil, fmt.Errorf("method %d not found on actor %s", method, act.Code)
}
p := reflect.New(methodMeta.Params.Elem()).Interface().(cbg.CBORMarshaler)
if err := json.Unmarshal([]byte(paramstr), p); err != nil {
return nil, fmt.Errorf("unmarshaling input into params type: %w", err)
}
buf := new(bytes.Buffer)
if err := p.MarshalCBOR(buf); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
type SendParams struct {
To address.Address
From address.Address
Val abi.TokenAmount
GasPremium *abi.TokenAmount
GasFeeCap *abi.TokenAmount
GasLimit *int64
Nonce *uint64
Method abi.MethodNum
Params []byte
Force bool
}
// This is specialised Send for Send command
// There might be room for generic Send that other commands can use to send their messages
// We will see
var ErrSendBalanceTooLow = errors.New("balance too low")
func (s *ServicesImpl) Send(ctx context.Context, params SendParams) (cid.Cid, error) {
if params.From == address.Undef {
defaddr, err := s.api.WalletDefaultAddress(ctx)
if err != nil {
return cid.Undef, err
}
params.From = defaddr
}
msg := &types.Message{
From: params.From,
To: params.To,
Value: params.Val,
Method: params.Method,
Params: params.Params,
}
if params.GasPremium != nil {
msg.GasPremium = *params.GasPremium
} else {
msg.GasPremium = types.NewInt(0)
}
if params.GasFeeCap != nil {
msg.GasFeeCap = *params.GasFeeCap
} else {
msg.GasFeeCap = types.NewInt(0)
}
if params.GasLimit != nil {
msg.GasLimit = *params.GasLimit
} else {
msg.GasLimit = 0
}
if !params.Force {
// Funds insufficient check
fromBalance, err := s.api.WalletBalance(ctx, msg.From)
if err != nil {
return cid.Undef, err
}
totalCost := types.BigAdd(types.BigMul(msg.GasFeeCap, types.NewInt(uint64(msg.GasLimit))), msg.Value)
if fromBalance.LessThan(totalCost) {
return cid.Undef, xerrors.Errorf("From balance %s less than total cost %s: %w", types.FIL(fromBalance), types.FIL(totalCost), ErrSendBalanceTooLow)
}
}
if params.Nonce != nil {
msg.Nonce = *params.Nonce
sm, err := s.api.WalletSignMessage(ctx, params.From, msg)
if err != nil {
return cid.Undef, err
}
_, err = s.api.MpoolPush(ctx, sm)
if err != nil {
return cid.Undef, err
}
return sm.Cid(), nil
}
sm, err := s.api.MpoolPushMessage(ctx, msg, nil)
if err != nil {
return cid.Undef, err
}
return sm.Cid(), nil
}

266
cli/services_send_test.go Normal file
View File

@ -0,0 +1,266 @@
package cli
import (
"context"
"fmt"
"testing"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/go-state-types/crypto"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/mocks"
types "github.com/filecoin-project/lotus/chain/types"
gomock "github.com/golang/mock/gomock"
cid "github.com/ipfs/go-cid"
"github.com/stretchr/testify/assert"
)
type markerKeyType struct{}
var markerKey = markerKeyType{}
type contextMatcher struct {
marker *int
}
// Matches returns whether x is a match.
func (cm contextMatcher) Matches(x interface{}) bool {
ctx, ok := x.(context.Context)
if !ok {
return false
}
maybeMarker, ok := ctx.Value(markerKey).(*int)
if !ok {
return false
}
return cm.marker == maybeMarker
}
func (cm contextMatcher) String() string {
return fmt.Sprintf("Context with Value(%v/%T, %p)", markerKey, markerKey, cm.marker)
}
func ContextWithMarker(ctx context.Context) (context.Context, gomock.Matcher) {
marker := new(int)
outCtx := context.WithValue(ctx, markerKey, marker)
return outCtx, contextMatcher{marker: marker}
}
func setupMockSrvcs(t *testing.T) (*ServicesImpl, *mocks.MockFullNode) {
mockCtrl := gomock.NewController(t)
mockApi := mocks.NewMockFullNode(mockCtrl)
srvcs := &ServicesImpl{
api: mockApi,
closer: mockCtrl.Finish,
}
return srvcs, mockApi
}
func fakeSign(msg *types.Message) *types.SignedMessage {
return &types.SignedMessage{
Message: *msg,
Signature: crypto.Signature{Type: crypto.SigTypeSecp256k1, Data: make([]byte, 32)},
}
}
func makeMessageSigner() (*cid.Cid, interface{}) {
smCid := cid.Undef
return &smCid,
func(_ context.Context, msg *types.Message, _ *api.MessageSendSpec) (*types.SignedMessage, error) {
sm := fakeSign(msg)
smCid = sm.Cid()
return sm, nil
}
}
type MessageMatcher SendParams
var _ gomock.Matcher = MessageMatcher{}
// Matches returns whether x is a match.
func (mm MessageMatcher) Matches(x interface{}) bool {
m, ok := x.(*types.Message)
if !ok {
return false
}
if mm.From != address.Undef && mm.From != m.From {
return false
}
if mm.To != address.Undef && mm.To != m.To {
return false
}
if types.BigCmp(mm.Val, m.Value) != 0 {
return false
}
if mm.Nonce != nil && *mm.Nonce != m.Nonce {
return false
}
if mm.GasPremium != nil && big.Cmp(*mm.GasPremium, m.GasPremium) != 0 {
return false
}
if mm.GasPremium == nil && m.GasPremium.Sign() != 0 {
return false
}
if mm.GasFeeCap != nil && big.Cmp(*mm.GasFeeCap, m.GasFeeCap) != 0 {
return false
}
if mm.GasFeeCap == nil && m.GasFeeCap.Sign() != 0 {
return false
}
if mm.GasLimit != nil && *mm.GasLimit != m.GasLimit {
return false
}
if mm.GasLimit == nil && m.GasLimit != 0 {
return false
}
// handle rest of options
return true
}
// String describes what the matcher matches.
func (mm MessageMatcher) String() string {
return fmt.Sprintf("%#v", SendParams(mm))
}
func TestSendService(t *testing.T) {
addrGen := address.NewForTestGetter()
a1 := addrGen()
a2 := addrGen()
const balance = 10000
params := SendParams{
From: a1,
To: a2,
Val: types.NewInt(balance - 100),
}
ctx, ctxM := ContextWithMarker(context.Background())
t.Run("happy", func(t *testing.T) {
params := params
srvcs, mockApi := setupMockSrvcs(t)
defer srvcs.Close() //nolint:errcheck
msgCid, sign := makeMessageSigner()
gomock.InOrder(
mockApi.EXPECT().WalletBalance(ctxM, params.From).Return(types.NewInt(balance), nil),
mockApi.EXPECT().MpoolPushMessage(ctxM, MessageMatcher(params), nil).DoAndReturn(sign),
)
c, err := srvcs.Send(ctx, params)
assert.NoError(t, err)
assert.Equal(t, *msgCid, c)
})
t.Run("balance-too-low", func(t *testing.T) {
params := params
srvcs, mockApi := setupMockSrvcs(t)
defer srvcs.Close() //nolint:errcheck
gomock.InOrder(
mockApi.EXPECT().WalletBalance(ctxM, a1).Return(types.NewInt(balance-200), nil),
// no MpoolPushMessage
)
c, err := srvcs.Send(ctx, params)
assert.Equal(t, c, cid.Undef)
assert.ErrorIs(t, err, ErrSendBalanceTooLow)
})
t.Run("force", func(t *testing.T) {
params := params
params.Force = true
srvcs, mockApi := setupMockSrvcs(t)
defer srvcs.Close() //nolint:errcheck
msgCid, sign := makeMessageSigner()
gomock.InOrder(
mockApi.EXPECT().WalletBalance(ctxM, a1).Return(types.NewInt(balance-200), nil).AnyTimes(),
mockApi.EXPECT().MpoolPushMessage(ctxM, MessageMatcher(params), nil).DoAndReturn(sign),
)
c, err := srvcs.Send(ctx, params)
assert.NoError(t, err)
assert.Equal(t, *msgCid, c)
})
t.Run("default-from", func(t *testing.T) {
params := params
params.From = address.Undef
mm := MessageMatcher(params)
mm.From = a1
srvcs, mockApi := setupMockSrvcs(t)
defer srvcs.Close() //nolint:errcheck
msgCid, sign := makeMessageSigner()
gomock.InOrder(
mockApi.EXPECT().WalletDefaultAddress(ctxM).Return(a1, nil),
mockApi.EXPECT().WalletBalance(ctxM, a1).Return(types.NewInt(balance), nil),
mockApi.EXPECT().MpoolPushMessage(ctxM, mm, nil).DoAndReturn(sign),
)
c, err := srvcs.Send(ctx, params)
assert.NoError(t, err)
assert.Equal(t, *msgCid, c)
})
t.Run("set-nonce", func(t *testing.T) {
params := params
n := uint64(5)
params.Nonce = &n
mm := MessageMatcher(params)
srvcs, mockApi := setupMockSrvcs(t)
defer srvcs.Close() //nolint:errcheck
_, _ = mm, mockApi
var sm *types.SignedMessage
gomock.InOrder(
mockApi.EXPECT().WalletBalance(ctxM, a1).Return(types.NewInt(balance), nil),
mockApi.EXPECT().WalletSignMessage(ctxM, a1, mm).DoAndReturn(
func(_ context.Context, _ address.Address, msg *types.Message) (*types.SignedMessage, error) {
sm = fakeSign(msg)
// now we expect MpoolPush with that SignedMessage
mockApi.EXPECT().MpoolPush(ctxM, sm).Return(sm.Cid(), nil)
return sm, nil
}),
)
c, err := srvcs.Send(ctx, params)
assert.NoError(t, err)
assert.Equal(t, sm.Cid(), c)
})
t.Run("gas-params", func(t *testing.T) {
params := params
limit := int64(1)
params.GasLimit = &limit
gfc := big.NewInt(100)
params.GasFeeCap = &gfc
gp := big.NewInt(10)
params.GasPremium = &gp
srvcs, mockApi := setupMockSrvcs(t)
defer srvcs.Close() //nolint:errcheck
msgCid, sign := makeMessageSigner()
gomock.InOrder(
mockApi.EXPECT().WalletBalance(ctxM, params.From).Return(types.NewInt(balance), nil),
mockApi.EXPECT().MpoolPushMessage(ctxM, MessageMatcher(params), nil).DoAndReturn(sign),
)
c, err := srvcs.Send(ctx, params)
assert.NoError(t, err)
assert.Equal(t, *msgCid, c)
})
}

81
cli/servicesmock_test.go Normal file
View File

@ -0,0 +1,81 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/filecoin-project/lotus/cli (interfaces: ServicesAPI)
// Package cli is a generated GoMock package.
package cli
import (
context "context"
go_address "github.com/filecoin-project/go-address"
abi "github.com/filecoin-project/go-state-types/abi"
gomock "github.com/golang/mock/gomock"
go_cid "github.com/ipfs/go-cid"
reflect "reflect"
)
// MockServicesAPI is a mock of ServicesAPI interface
type MockServicesAPI struct {
ctrl *gomock.Controller
recorder *MockServicesAPIMockRecorder
}
// MockServicesAPIMockRecorder is the mock recorder for MockServicesAPI
type MockServicesAPIMockRecorder struct {
mock *MockServicesAPI
}
// NewMockServicesAPI creates a new mock instance
func NewMockServicesAPI(ctrl *gomock.Controller) *MockServicesAPI {
mock := &MockServicesAPI{ctrl: ctrl}
mock.recorder = &MockServicesAPIMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use
func (m *MockServicesAPI) EXPECT() *MockServicesAPIMockRecorder {
return m.recorder
}
// Close mocks base method
func (m *MockServicesAPI) Close() error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Close")
ret0, _ := ret[0].(error)
return ret0
}
// Close indicates an expected call of Close
func (mr *MockServicesAPIMockRecorder) Close() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockServicesAPI)(nil).Close))
}
// DecodeTypedParamsFromJSON mocks base method
func (m *MockServicesAPI) DecodeTypedParamsFromJSON(arg0 context.Context, arg1 go_address.Address, arg2 abi.MethodNum, arg3 string) ([]byte, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DecodeTypedParamsFromJSON", arg0, arg1, arg2, arg3)
ret0, _ := ret[0].([]byte)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// DecodeTypedParamsFromJSON indicates an expected call of DecodeTypedParamsFromJSON
func (mr *MockServicesAPIMockRecorder) DecodeTypedParamsFromJSON(arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DecodeTypedParamsFromJSON", reflect.TypeOf((*MockServicesAPI)(nil).DecodeTypedParamsFromJSON), arg0, arg1, arg2, arg3)
}
// Send mocks base method
func (m *MockServicesAPI) Send(arg0 context.Context, arg1 SendParams) (go_cid.Cid, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Send", arg0, arg1)
ret0, _ := ret[0].(go_cid.Cid)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// Send indicates an expected call of Send
func (mr *MockServicesAPIMockRecorder) Send(arg0, arg1 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Send", reflect.TypeOf((*MockServicesAPI)(nil).Send), arg0, arg1)
}

2
cmd/lotus-storage-miner/info.go
View File

@ -282,6 +282,7 @@ var stateList = []stateMeta{
{col: color.FgBlue, state: sealing.Empty},
{col: color.FgBlue, state: sealing.WaitDeals},
{col: color.FgBlue, state: sealing.AddPiece},
{col: color.FgRed, state: sealing.UndefinedSectorState},
{col: color.FgYellow, state: sealing.Packing},
@ -304,6 +305,7 @@ var stateList = []stateMeta{
{col: color.FgCyan, state: sealing.Removed},
{col: color.FgRed, state: sealing.FailedUnrecoverable},
{col: color.FgRed, state: sealing.AddPieceFailed},
{col: color.FgRed, state: sealing.SealPreCommit1Failed},
{col: color.FgRed, state: sealing.SealPreCommit2Failed},
{col: color.FgRed, state: sealing.PreCommitFailed},

9
cmd/lotus-storage-miner/sectors.go
View File

@ -55,7 +55,14 @@ var sectorsPledgeCmd = &cli.Command{
defer closer()
ctx := lcli.ReqContext(cctx)
return nodeApi.PledgeSector(ctx)
id, err := nodeApi.PledgeSector(ctx)
if err != nil {
return err
}
fmt.Println("Created CC sector: ", id.Number)
return nil
},
}

2
cmd/lotus/daemon.go
View File

@ -400,7 +400,7 @@ func ImportChain(ctx context.Context, r repo.Repo, fname string, snapshot bool)
defer resp.Body.Close() //nolint:errcheck
if resp.StatusCode != http.StatusOK {
return xerrors.Errorf("non-200 response: %d", resp.StatusCode)
return xerrors.Errorf("fetching chain CAR failed with non-200 response: %d", resp.StatusCode)
}
rd = resp.Body

8
documentation/en/api-methods-miner.md
View File

@ -1143,7 +1143,13 @@ Perms: write
Inputs: `null`
Response: `{}`
Response:
```json
{
"Miner": 1000,
"Number": 9
}
```
## Return

50
extern/storage-sealing/cbor_gen.go vendored
View File

@ -519,7 +519,7 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
_, err := w.Write(cbg.CborNull)
return err
}
if _, err := w.Write([]byte{184, 25}); err != nil {
if _, err := w.Write([]byte{184, 26}); err != nil {
return err
}
@ -586,6 +586,28 @@ func (t *SectorInfo) MarshalCBOR(w io.Writer) error {
}
}
// t.CreationTime (int64) (int64)
if len("CreationTime") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"CreationTime\" was too long")
}
if err := cbg.WriteMajorTypeHeaderBuf(scratch, w, cbg.MajTextString, uint64(len("CreationTime"))); err != nil {
return err
}
if _, err := io.WriteString(w, string("CreationTime")); err != nil {
return err
}
if t.CreationTime >= 0 {
if err := cbg.WriteMajorTypeHeaderBuf(scratch, w, cbg.MajUnsignedInt, uint64(t.CreationTime)); err != nil {
return err
}
} else {
if err := cbg.WriteMajorTypeHeaderBuf(scratch, w, cbg.MajNegativeInt, uint64(-t.CreationTime-1)); err != nil {
return err
}
}
// t.Pieces ([]sealing.Piece) (slice)
if len("Pieces") > cbg.MaxLength {
return xerrors.Errorf("Value in field \"Pieces\" was too long")
@ -1151,6 +1173,32 @@ func (t *SectorInfo) UnmarshalCBOR(r io.Reader) error {
t.SectorType = abi.RegisteredSealProof(extraI)
}
// t.CreationTime (int64) (int64)
case "CreationTime":
{
maj, extra, err := cbg.CborReadHeaderBuf(br, scratch)
var extraI int64
if err != nil {
return err
}
switch maj {
case cbg.MajUnsignedInt:
extraI = int64(extra)
if extraI < 0 {
return fmt.Errorf("int64 positive overflow")
}
case cbg.MajNegativeInt:
extraI = int64(extra)
if extraI < 0 {
return fmt.Errorf("int64 negative oveflow")
}
extraI = -1 - extraI
default:
return fmt.Errorf("wrong type for int64 field: %d", maj)
}
t.CreationTime = int64(extraI)
}
// t.Pieces ([]sealing.Piece) (slice)
case "Pieces":

160
extern/storage-sealing/fsm.go vendored
View File

@ -37,14 +37,22 @@ var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *Secto
// Sealing
UndefinedSectorState: planOne(
on(SectorStart{}, Empty),
on(SectorStart{}, WaitDeals),
on(SectorStartCC{}, Packing),
),
Empty: planOne(on(SectorAddPiece{}, WaitDeals)),
WaitDeals: planOne(
on(SectorAddPiece{}, WaitDeals),
Empty: planOne( // deprecated
on(SectorAddPiece{}, AddPiece),
on(SectorStartPacking{}, Packing),
),
WaitDeals: planOne(
on(SectorAddPiece{}, AddPiece),
on(SectorStartPacking{}, Packing),
),
AddPiece: planOne(
on(SectorPieceAdded{}, WaitDeals),
apply(SectorStartPacking{}),
on(SectorAddPieceFailed{}, AddPieceFailed),
),
Packing: planOne(on(SectorPacked{}, GetTicket)),
GetTicket: planOne(
on(SectorTicket{}, PreCommit1),
@ -97,6 +105,7 @@ var fsmPlanners = map[SectorState]func(events []statemachine.Event, state *Secto
// Sealing errors
AddPieceFailed: planOne(),
SealPreCommit1Failed: planOne(
on(SectorRetrySealPreCommit1{}, PreCommit1),
),
@ -238,12 +247,11 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
/*
* Empty <- incoming deals
| |
| v
*<- WaitDeals <- incoming deals
| |
| v
UndefinedSectorState (start)
v |
*<- WaitDeals <-> AddPiece |
| | /--------------------/
| v v
*<- Packing <- incoming committed capacity
| |
| v
@ -282,10 +290,6 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
v
FailedUnrecoverable
UndefinedSectorState <- ¯\_(ツ)_/¯
| ^
*---------------------/
*/
m.stats.updateSector(m.minerSectorID(state.SectorNumber), state.State)
@ -295,7 +299,9 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
case Empty:
fallthrough
case WaitDeals:
log.Infof("Waiting for deals %d", state.SectorNumber)
return m.handleWaitDeals, processed, nil
case AddPiece:
return m.handleAddPiece, processed, nil
case Packing:
return m.handlePacking, processed, nil
case GetTicket:
@ -418,60 +424,10 @@ func (m *Sealing) restartSectors(ctx context.Context) error {
log.Errorf("loading sector list: %+v", err)
}
cfg, err := m.getConfig()
if err != nil {
return xerrors.Errorf("getting the sealing delay: %w", err)
}
spt, err := m.currentSealProof(ctx)
if err != nil {
return xerrors.Errorf("getting current seal proof: %w", err)
}
ssize, err := spt.SectorSize()
if err != nil {
return err
}
// m.unsealedInfoMap.lk.Lock() taken early in .New to prevent races
defer m.unsealedInfoMap.lk.Unlock()
for _, sector := range trackedSectors {
if err := m.sectors.Send(uint64(sector.SectorNumber), SectorRestart{}); err != nil {
log.Errorf("restarting sector %d: %+v", sector.SectorNumber, err)
}
if sector.State == WaitDeals {
// put the sector in the unsealedInfoMap
if _, ok := m.unsealedInfoMap.infos[sector.SectorNumber]; ok {
// something's funky here, but probably safe to move on
log.Warnf("sector %v was already in the unsealedInfoMap when restarting", sector.SectorNumber)
} else {
ui := UnsealedSectorInfo{
ssize: ssize,
}
for _, p := range sector.Pieces {
if p.DealInfo != nil {
ui.numDeals++
}
ui.stored += p.Piece.Size
ui.pieceSizes = append(ui.pieceSizes, p.Piece.Size.Unpadded())
}
m.unsealedInfoMap.infos[sector.SectorNumber] = ui
}
// start a fresh timer for the sector
if cfg.WaitDealsDelay > 0 {
timer := time.NewTimer(cfg.WaitDealsDelay)
go func() {
<-timer.C
if err := m.StartPacking(sector.SectorNumber); err != nil {
log.Errorf("starting sector %d: %+v", sector.SectorNumber, err)
}
}()
}
}
}
// TODO: Grab on-chain sector set and diff with trackedSectors
@ -494,56 +450,72 @@ func final(events []statemachine.Event, state *SectorInfo) (uint64, error) {
return 0, xerrors.Errorf("didn't expect any events in state %s, got %+v", state.State, events)
}
func on(mut mutator, next SectorState) func() (mutator, func(*SectorInfo) error) {
return func() (mutator, func(*SectorInfo) error) {
return mut, func(state *SectorInfo) error {
func on(mut mutator, next SectorState) func() (mutator, func(*SectorInfo) (bool, error)) {
return func() (mutator, func(*SectorInfo) (bool, error)) {
return mut, func(state *SectorInfo) (bool, error) {
state.State = next
return nil
return false, nil
}
}
}
func onReturning(mut mutator) func() (mutator, func(*SectorInfo) error) {
return func() (mutator, func(*SectorInfo) error) {
return mut, func(state *SectorInfo) error {
// like `on`, but doesn't change state
func apply(mut mutator) func() (mutator, func(*SectorInfo) (bool, error)) {
return func() (mutator, func(*SectorInfo) (bool, error)) {
return mut, func(state *SectorInfo) (bool, error) {
return true, nil
}
}
}
func onReturning(mut mutator) func() (mutator, func(*SectorInfo) (bool, error)) {
return func() (mutator, func(*SectorInfo) (bool, error)) {
return mut, func(state *SectorInfo) (bool, error) {
if state.Return == "" {
return xerrors.Errorf("return state not set")
return false, xerrors.Errorf("return state not set")
}
state.State = SectorState(state.Return)
state.Return = ""
return nil
return false, nil
}
}
}
func planOne(ts ...func() (mut mutator, next func(*SectorInfo) error)) func(events []statemachine.Event, state *SectorInfo) (uint64, error) {
func planOne(ts ...func() (mut mutator, next func(*SectorInfo) (more bool, err error))) func(events []statemachine.Event, state *SectorInfo) (uint64, error) {
return func(events []statemachine.Event, state *SectorInfo) (uint64, error) {
if gm, ok := events[0].User.(globalMutator); ok {
gm.applyGlobal(state)
return 1, nil
}
for i, event := range events {
if gm, ok := event.User.(globalMutator); ok {
gm.applyGlobal(state)
return uint64(i + 1), nil
}
for _, t := range ts {
mut, next := t()
for _, t := range ts {
mut, next := t()
if reflect.TypeOf(events[0].User) != reflect.TypeOf(mut) {
if reflect.TypeOf(event.User) != reflect.TypeOf(mut) {
continue
}
if err, iserr := event.User.(error); iserr {
log.Warnf("sector %d got error event %T: %+v", state.SectorNumber, event.User, err)
}
event.User.(mutator).apply(state)
more, err := next(state)
if err != nil || !more {
return uint64(i + 1), err
}
}
_, ok := event.User.(Ignorable)
if ok {
continue
}
if err, iserr := events[0].User.(error); iserr {
log.Warnf("sector %d got error event %T: %+v", state.SectorNumber, events[0].User, err)
}
events[0].User.(mutator).apply(state)
return 1, next(state)
return uint64(i + 1), xerrors.Errorf("planner for state %s received unexpected event %T (%+v)", state.State, event.User, event)
}
_, ok := events[0].User.(Ignorable)
if ok {
return 1, nil
}
return 0, xerrors.Errorf("planner for state %s received unexpected event %T (%+v)", state.State, events[0].User, events[0])
return uint64(len(events)), nil
}
}

28
extern/storage-sealing/fsm_events.go vendored
View File

@ -1,13 +1,16 @@
package sealing
import (
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
"time"
"github.com/ipfs/go-cid"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/lotus/chain/actors/builtin/miner"
)
type mutator interface {
@ -67,23 +70,34 @@ func (evt SectorStart) apply(state *SectorInfo) {
type SectorStartCC struct {
ID abi.SectorNumber
SectorType abi.RegisteredSealProof
Pieces []Piece
}
func (evt SectorStartCC) apply(state *SectorInfo) {
state.SectorNumber = evt.ID
state.Pieces = evt.Pieces
state.SectorType = evt.SectorType
}
type SectorAddPiece struct {
NewPiece Piece
}
type SectorAddPiece struct{}
func (evt SectorAddPiece) apply(state *SectorInfo) {
state.Pieces = append(state.Pieces, evt.NewPiece)
if state.CreationTime == 0 {
state.CreationTime = time.Now().Unix()
}
}
type SectorPieceAdded struct {
NewPieces []Piece
}
func (evt SectorPieceAdded) apply(state *SectorInfo) {
state.Pieces = append(state.Pieces, evt.NewPieces...)
}
type SectorAddPieceFailed struct{ error }
func (evt SectorAddPieceFailed) FormatError(xerrors.Printer) (next error) { return evt.error }
func (evt SectorAddPieceFailed) apply(si *SectorInfo) {}
type SectorStartPacking struct{}
func (evt SectorStartPacking) apply(*SectorInfo) {}

95
extern/storage-sealing/garbage.go vendored
View File

@ -5,91 +5,42 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/specs-storage/storage"
)
func (m *Sealing) pledgeSector(ctx context.Context, sectorID storage.SectorRef, existingPieceSizes []abi.UnpaddedPieceSize, sizes ...abi.UnpaddedPieceSize) ([]abi.PieceInfo, error) {
if len(sizes) == 0 {
return nil, nil
}
func (m *Sealing) PledgeSector(ctx context.Context) (storage.SectorRef, error) {
m.inputLk.Lock()
defer m.inputLk.Unlock()
log.Infof("Pledge %d, contains %+v", sectorID, existingPieceSizes)
out := make([]abi.PieceInfo, len(sizes))
for i, size := range sizes {
ppi, err := m.sealer.AddPiece(ctx, sectorID, existingPieceSizes, size, NewNullReader(size))
if err != nil {
return nil, xerrors.Errorf("add piece: %w", err)
}
existingPieceSizes = append(existingPieceSizes, size)
out[i] = ppi
}
return out, nil
}
func (m *Sealing) PledgeSector() error {
cfg, err := m.getConfig()
if err != nil {
return xerrors.Errorf("getting config: %w", err)
return storage.SectorRef{}, xerrors.Errorf("getting config: %w", err)
}
if cfg.MaxSealingSectors > 0 {
if m.stats.curSealing() >= cfg.MaxSealingSectors {
return xerrors.Errorf("too many sectors sealing (curSealing: %d, max: %d)", m.stats.curSealing(), cfg.MaxSealingSectors)
return storage.SectorRef{}, xerrors.Errorf("too many sectors sealing (curSealing: %d, max: %d)", m.stats.curSealing(), cfg.MaxSealingSectors)
}
}
go func() {
ctx := context.TODO() // we can't use the context from command which invokes
// this, as we run everything here async, and it's cancelled when the
// command exits
spt, err := m.currentSealProof(ctx)
if err != nil {
return storage.SectorRef{}, xerrors.Errorf("getting seal proof type: %w", err)
}
spt, err := m.currentSealProof(ctx)
if err != nil {
log.Errorf("%+v", err)
return
}
sid, err := m.sc.Next()
if err != nil {
return storage.SectorRef{}, xerrors.Errorf("generating sector number: %w", err)
}
sectorID := m.minerSector(spt, sid)
err = m.sealer.NewSector(ctx, sectorID)
if err != nil {
return storage.SectorRef{}, xerrors.Errorf("notifying sealer of the new sector: %w", err)
}
size, err := spt.SectorSize()
if err != nil {
log.Errorf("%+v", err)
return
}
sid, err := m.sc.Next()
if err != nil {
log.Errorf("%+v", err)
return
}
sectorID := m.minerSector(spt, sid)
err = m.sealer.NewSector(ctx, sectorID)
if err != nil {
log.Errorf("%+v", err)
return
}
pieces, err := m.pledgeSector(ctx, sectorID, []abi.UnpaddedPieceSize{}, abi.PaddedPieceSize(size).Unpadded())
if err != nil {
log.Errorf("%+v", err)
return
}
ps := make([]Piece, len(pieces))
for idx := range ps {
ps[idx] = Piece{
Piece: pieces[idx],
DealInfo: nil,
}
}
if err := m.newSectorCC(ctx, sid, ps); err != nil {
log.Errorf("%+v", err)
return
}
}()
return nil
log.Infof("Creating CC sector %d", sid)
return sectorID, m.sectors.Send(uint64(sid), SectorStartCC{
ID: sid,
SectorType: spt,
})
}

424
extern/storage-sealing/input.go vendored Normal file
View File

@ -0,0 +1,424 @@
package sealing
import (
"context"
"sort"
"time"
"golang.org/x/xerrors"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/go-padreader"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-statemachine"
"github.com/filecoin-project/specs-storage/storage"
sectorstorage "github.com/filecoin-project/lotus/extern/sector-storage"
"github.com/filecoin-project/lotus/extern/sector-storage/ffiwrapper"
)
func (m *Sealing) handleWaitDeals(ctx statemachine.Context, sector SectorInfo) error {
var used abi.UnpaddedPieceSize
for _, piece := range sector.Pieces {
used += piece.Piece.Size.Unpadded()
}
m.inputLk.Lock()
started, err := m.maybeStartSealing(ctx, sector, used)
if err != nil || started {
delete(m.openSectors, m.minerSectorID(sector.SectorNumber))
m.inputLk.Unlock()
return err
}
m.openSectors[m.minerSectorID(sector.SectorNumber)] = &openSector{
used: used,
maybeAccept: func(cid cid.Cid) error {
// todo check deal start deadline (configurable)
sid := m.minerSectorID(sector.SectorNumber)
m.assignedPieces[sid] = append(m.assignedPieces[sid], cid)
return ctx.Send(SectorAddPiece{})
},
}
go func() {
defer m.inputLk.Unlock()
if err := m.updateInput(ctx.Context(), sector.SectorType); err != nil {
log.Errorf("%+v", err)
}
}()
return nil
}
func (m *Sealing) maybeStartSealing(ctx statemachine.Context, sector SectorInfo, used abi.UnpaddedPieceSize) (bool, error) {
now := time.Now()
st := m.sectorTimers[m.minerSectorID(sector.SectorNumber)]
if st != nil {
if !st.Stop() { // timer expired, SectorStartPacking was/is being sent
// we send another SectorStartPacking in case one was sent in the handleAddPiece state
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "wait-timeout")
return true, ctx.Send(SectorStartPacking{})
}
}
ssize, err := sector.SectorType.SectorSize()
if err != nil {
return false, xerrors.Errorf("getting sector size")
}
maxDeals, err := getDealPerSectorLimit(ssize)
if err != nil {
return false, xerrors.Errorf("getting per-sector deal limit: %w", err)
}
if len(sector.dealIDs()) >= maxDeals {
// can't accept more deals
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "maxdeals")
return true, ctx.Send(SectorStartPacking{})
}
if used.Padded() == abi.PaddedPieceSize(ssize) {
// sector full
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "filled")
return true, ctx.Send(SectorStartPacking{})
}
if sector.CreationTime != 0 {
cfg, err := m.getConfig()
if err != nil {
m.inputLk.Unlock()
return false, xerrors.Errorf("getting storage config: %w", err)
}
// todo check deal age, start sealing if any deal has less than X (configurable) to start deadline
sealTime := time.Unix(sector.CreationTime, 0).Add(cfg.WaitDealsDelay)
if now.After(sealTime) {
m.inputLk.Unlock()
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "wait-timeout")
return true, ctx.Send(SectorStartPacking{})
}
m.sectorTimers[m.minerSectorID(sector.SectorNumber)] = time.AfterFunc(sealTime.Sub(now), func() {
log.Infow("starting to seal deal sector", "sector", sector.SectorNumber, "trigger", "wait-timer")
if err := ctx.Send(SectorStartPacking{}); err != nil {
log.Errorw("sending SectorStartPacking event failed", "sector", sector.SectorNumber, "error", err)
}
})
}
return false, nil
}
func (m *Sealing) handleAddPiece(ctx statemachine.Context, sector SectorInfo) error {
ssize, err := sector.SectorType.SectorSize()
if err != nil {
return err
}
res := SectorPieceAdded{}
m.inputLk.Lock()
pending, ok := m.assignedPieces[m.minerSectorID(sector.SectorNumber)]
if ok {
delete(m.assignedPieces, m.minerSectorID(sector.SectorNumber))
}
m.inputLk.Unlock()
if !ok {
// nothing to do here (might happen after a restart in AddPiece)
return ctx.Send(res)
}
var offset abi.UnpaddedPieceSize
pieceSizes := make([]abi.UnpaddedPieceSize, len(sector.Pieces))
for i, p := range sector.Pieces {
pieceSizes[i] = p.Piece.Size.Unpadded()
offset += p.Piece.Size.Unpadded()
}
maxDeals, err := getDealPerSectorLimit(ssize)
if err != nil {
return xerrors.Errorf("getting per-sector deal limit: %w", err)
}
for i, piece := range pending {
m.inputLk.Lock()
deal, ok := m.pendingPieces[piece]
m.inputLk.Unlock()
if !ok {
return xerrors.Errorf("piece %s assigned to sector %d not found", piece, sector.SectorNumber)
}
if len(sector.dealIDs())+(i+1) > maxDeals {
// todo: this is rather unlikely to happen, but in case it does, return the deal to waiting queue instead of failing it
deal.accepted(sector.SectorNumber, offset, xerrors.Errorf("too many deals assigned to sector %d, dropping deal", sector.SectorNumber))
continue
}
pads, padLength := ffiwrapper.GetRequiredPadding(offset.Padded(), deal.size.Padded())
if offset.Padded()+padLength+deal.size.Padded() > abi.PaddedPieceSize(ssize) {
// todo: this is rather unlikely to happen, but in case it does, return the deal to waiting queue instead of failing it
deal.accepted(sector.SectorNumber, offset, xerrors.Errorf("piece %s assigned to sector %d with not enough space", piece, sector.SectorNumber))
continue
}
offset += padLength.Unpadded()
for _, p := range pads {
ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx.Context(), DealSectorPriority),
m.minerSector(sector.SectorType, sector.SectorNumber),
pieceSizes,
p.Unpadded(),
NewNullReader(p.Unpadded()))
if err != nil {
err = xerrors.Errorf("writing padding piece: %w", err)
deal.accepted(sector.SectorNumber, offset, err)
return ctx.Send(SectorAddPieceFailed{err})
}
pieceSizes = append(pieceSizes, p.Unpadded())
res.NewPieces = append(res.NewPieces, Piece{
Piece: ppi,
})
}
ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx.Context(), DealSectorPriority),
m.minerSector(sector.SectorType, sector.SectorNumber),
pieceSizes,
deal.size,
deal.data)
if err != nil {
err = xerrors.Errorf("writing piece: %w", err)
deal.accepted(sector.SectorNumber, offset, err)
return ctx.Send(SectorAddPieceFailed{err})
}
log.Infow("deal added to a sector", "deal", deal.deal.DealID, "sector", sector.SectorNumber, "piece", ppi.PieceCID)
deal.accepted(sector.SectorNumber, offset, nil)
offset += deal.size
pieceSizes = append(pieceSizes, deal.size)
res.NewPieces = append(res.NewPieces, Piece{
Piece: ppi,
DealInfo: &deal.deal,
})
}
return ctx.Send(res)
}
func (m *Sealing) handleAddPieceFailed(ctx statemachine.Context, sector SectorInfo) error {
log.Errorf("No recovery plan for AddPiece failing")
// todo: cleanup sector / just go retry (requires adding offset param to AddPiece in sector-storage for this to be safe)
return nil
}
func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPieceSize, data storage.Data, deal DealInfo) (abi.SectorNumber, abi.PaddedPieceSize, error) {
log.Infof("Adding piece for deal %d (publish msg: %s)", deal.DealID, deal.PublishCid)
if (padreader.PaddedSize(uint64(size))) != size {
return 0, 0, xerrors.Errorf("cannot allocate unpadded piece")
}
sp, err := m.currentSealProof(ctx)
if err != nil {
return 0, 0, xerrors.Errorf("getting current seal proof type: %w", err)
}
ssize, err := sp.SectorSize()
if err != nil {
return 0, 0, err
}
if size > abi.PaddedPieceSize(ssize).Unpadded() {
return 0, 0, xerrors.Errorf("piece cannot fit into a sector")
}
if _, err := deal.DealProposal.Cid(); err != nil {
return 0, 0, xerrors.Errorf("getting proposal CID: %w", err)
}
m.inputLk.Lock()
if _, exist := m.pendingPieces[proposalCID(deal)]; exist {
m.inputLk.Unlock()
return 0, 0, xerrors.Errorf("piece for deal %s already pending", proposalCID(deal))
}
resCh := make(chan struct {
sn abi.SectorNumber
offset abi.UnpaddedPieceSize
err error
}, 1)
m.pendingPieces[proposalCID(deal)] = &pendingPiece{
size: size,
deal: deal,
data: data,
assigned: false,
accepted: func(sn abi.SectorNumber, offset abi.UnpaddedPieceSize, err error) {
resCh <- struct {
sn abi.SectorNumber
offset abi.UnpaddedPieceSize
err error
}{sn: sn, offset: offset, err: err}
},
}
go func() {
defer m.inputLk.Unlock()
if err := m.updateInput(ctx, sp); err != nil {
log.Errorf("%+v", err)
}
}()
res := <-resCh
return res.sn, res.offset.Padded(), res.err
}
// called with m.inputLk
func (m *Sealing) updateInput(ctx context.Context, sp abi.RegisteredSealProof) error {
ssize, err := sp.SectorSize()
if err != nil {
return err
}
type match struct {
sector abi.SectorID
deal cid.Cid
size abi.UnpaddedPieceSize
padding abi.UnpaddedPieceSize
}
var matches []match
toAssign := map[cid.Cid]struct{}{} // used to maybe create new sectors
// todo: this is distinctly O(n^2), may need to be optimized for tiny deals and large scale miners
// (unlikely to be a problem now)
for proposalCid, piece := range m.pendingPieces {
if piece.assigned {
continue // already assigned to a sector, skip
}
toAssign[proposalCid] = struct{}{}
for id, sector := range m.openSectors {
avail := abi.PaddedPieceSize(ssize).Unpadded() - sector.used
if piece.size <= avail { // (note: if we have enough space for the piece, we also have enough space for inter-piece padding)
matches = append(matches, match{
sector: id,
deal: proposalCid,
size: piece.size,
padding: avail % piece.size,
})
}
}
}
sort.Slice(matches, func(i, j int) bool {
if matches[i].padding != matches[j].padding { // less padding is better
return matches[i].padding < matches[j].padding
}
if matches[i].size != matches[j].size { // larger pieces are better
return matches[i].size < matches[j].size
}
return matches[i].sector.Number < matches[j].sector.Number // prefer older sectors
})
var assigned int
for _, mt := range matches {
if m.pendingPieces[mt.deal].assigned {
assigned++
continue
}
if _, found := m.openSectors[mt.sector]; !found {
continue
}
err := m.openSectors[mt.sector].maybeAccept(mt.deal)
if err != nil {
m.pendingPieces[mt.deal].accepted(mt.sector.Number, 0, err) // non-error case in handleAddPiece
}
m.pendingPieces[mt.deal].assigned = true
delete(toAssign, mt.deal)
if err != nil {
log.Errorf("sector %d rejected deal %s: %+v", mt.sector, mt.deal, err)
continue
}
delete(m.openSectors, mt.sector)
}
if len(toAssign) > 0 {
if err := m.tryCreateDealSector(ctx, sp); err != nil {
log.Errorw("Failed to create a new sector for deals", "error", err)
}
}
return nil
}
func (m *Sealing) tryCreateDealSector(ctx context.Context, sp abi.RegisteredSealProof) error {
cfg, err := m.getConfig()
if err != nil {
return xerrors.Errorf("getting storage config: %w", err)
}
if cfg.MaxSealingSectorsForDeals > 0 && m.stats.curSealing() >= cfg.MaxSealingSectorsForDeals {
return nil
}
if cfg.MaxWaitDealsSectors > 0 && m.stats.curStaging() >= cfg.MaxWaitDealsSectors {
return nil
}
// Now actually create a new sector
sid, err := m.sc.Next()
if err != nil {
return xerrors.Errorf("getting sector number: %w", err)
}
err = m.sealer.NewSector(ctx, m.minerSector(sp, sid))
if err != nil {
return xerrors.Errorf("initializing sector: %w", err)
}
log.Infow("Creating sector", "number", sid, "type", "deal", "proofType", sp)
return m.sectors.Send(uint64(sid), SectorStart{
ID: sid,
SectorType: sp,
})
}
func (m *Sealing) StartPacking(sid abi.SectorNumber) error {
return m.sectors.Send(uint64(sid), SectorStartPacking{})
}
func proposalCID(deal DealInfo) cid.Cid {
pc, err := deal.DealProposal.Cid()
if err != nil {
log.Errorf("DealProposal.Cid error: %+v", err)
return cid.Undef
}
return pc
}

325
extern/storage-sealing/sealing.go vendored
View File

@ -3,8 +3,6 @@ package sealing
import (
"context"
"errors"
"io"
"math"
"sync"
"time"
@ -15,7 +13,6 @@ import (
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
padreader "github.com/filecoin-project/go-padreader"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/go-state-types/big"
"github.com/filecoin-project/go-state-types/crypto"
@ -89,9 +86,13 @@ type Sealing struct {
sectors *statemachine.StateGroup
sc SectorIDCounter
verif ffiwrapper.Verifier
pcp PreCommitPolicy
pcp PreCommitPolicy
unsealedInfoMap UnsealedSectorMap
inputLk sync.Mutex
openSectors map[abi.SectorID]*openSector
sectorTimers map[abi.SectorID]*time.Timer
pendingPieces map[cid.Cid]*pendingPiece
assignedPieces map[abi.SectorID][]cid.Cid
upgradeLk sync.Mutex
toUpgrade map[abi.SectorNumber]struct{}
@ -113,17 +114,20 @@ type FeeConfig struct {
MaxTerminateGasFee abi.TokenAmount
}
type UnsealedSectorMap struct {
infos map[abi.SectorNumber]UnsealedSectorInfo
lk sync.Mutex
type openSector struct {
used abi.UnpaddedPieceSize // change to bitfield/rle when AddPiece gains offset support to better fill sectors
maybeAccept func(cid.Cid) error // called with inputLk
}
type UnsealedSectorInfo struct {
numDeals uint64
// stored should always equal sum of pieceSizes.Padded()
stored abi.PaddedPieceSize
pieceSizes []abi.UnpaddedPieceSize
ssize abi.SectorSize
type pendingPiece struct {
size abi.UnpaddedPieceSize
deal DealInfo
data storage.Data
assigned bool // assigned to a sector?
accepted func(abi.SectorNumber, abi.UnpaddedPieceSize, error)
}
func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds datastore.Batching, sealer sectorstorage.SectorManager, sc SectorIDCounter, verif ffiwrapper.Verifier, pcp PreCommitPolicy, gc GetSealingConfigFunc, notifee SectorStateNotifee, as AddrSel) *Sealing {
@ -137,12 +141,12 @@ func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds
sc: sc,
verif: verif,
pcp: pcp,
unsealedInfoMap: UnsealedSectorMap{
infos: make(map[abi.SectorNumber]UnsealedSectorInfo),
lk: sync.Mutex{},
},
toUpgrade: map[abi.SectorNumber]struct{}{},
openSectors: map[abi.SectorID]*openSector{},
sectorTimers: map[abi.SectorID]*time.Timer{},
pendingPieces: map[cid.Cid]*pendingPiece{},
assignedPieces: map[abi.SectorID][]cid.Cid{},
toUpgrade: map[abi.SectorNumber]struct{}{},
notifee: notifee,
addrSel: as,
@ -159,8 +163,6 @@ func New(api SealingAPI, fc FeeConfig, events Events, maddr address.Address, ds
s.sectors = statemachine.New(namespace.Wrap(ds, datastore.NewKey(SectorStorePrefix)), s, SectorInfo{})
s.unsealedInfoMap.lk.Lock() // released after initialized in .Run()
return s
}
@ -184,104 +186,6 @@ func (m *Sealing) Stop(ctx context.Context) error {
return nil
}
func (m *Sealing) AddPieceToAnySector(ctx context.Context, size abi.UnpaddedPieceSize, r io.Reader, d DealInfo) (abi.SectorNumber, abi.PaddedPieceSize, error) {
log.Infof("Adding piece for deal %d (publish msg: %s)", d.DealID, d.PublishCid)
if (padreader.PaddedSize(uint64(size))) != size {
return 0, 0, xerrors.Errorf("cannot allocate unpadded piece")
}
sp, err := m.currentSealProof(ctx)
if err != nil {
return 0, 0, xerrors.Errorf("getting current seal proof type: %w", err)
}
ssize, err := sp.SectorSize()
if err != nil {
return 0, 0, err
}
if size > abi.PaddedPieceSize(ssize).Unpadded() {
return 0, 0, xerrors.Errorf("piece cannot fit into a sector")
}
m.unsealedInfoMap.lk.Lock()
sid, pads, err := m.getSectorAndPadding(ctx, size)
if err != nil {
m.unsealedInfoMap.lk.Unlock()
return 0, 0, xerrors.Errorf("getting available sector: %w", err)
}
for _, p := range pads {
err = m.addPiece(ctx, sid, p.Unpadded(), NewNullReader(p.Unpadded()), nil)
if err != nil {
m.unsealedInfoMap.lk.Unlock()
return 0, 0, xerrors.Errorf("writing pads: %w", err)
}
}
offset := m.unsealedInfoMap.infos[sid].stored
err = m.addPiece(ctx, sid, size, r, &d)
if err != nil {
m.unsealedInfoMap.lk.Unlock()
return 0, 0, xerrors.Errorf("adding piece to sector: %w", err)
}
startPacking := m.unsealedInfoMap.infos[sid].numDeals >= getDealPerSectorLimit(ssize)
m.unsealedInfoMap.lk.Unlock()
if startPacking {
if err := m.StartPacking(sid); err != nil {
return 0, 0, xerrors.Errorf("start packing: %w", err)
}
}
return sid, offset, nil
}
// Caller should hold m.unsealedInfoMap.lk
func (m *Sealing) addPiece(ctx context.Context, sectorID abi.SectorNumber, size abi.UnpaddedPieceSize, r io.Reader, di *DealInfo) error {
log.Infof("Adding piece to sector %d", sectorID)
sp, err := m.currentSealProof(ctx)
if err != nil {
return xerrors.Errorf("getting current seal proof type: %w", err)
}
ssize, err := sp.SectorSize()
if err != nil {
return err
}
ppi, err := m.sealer.AddPiece(sectorstorage.WithPriority(ctx, DealSectorPriority), m.minerSector(sp, sectorID), m.unsealedInfoMap.infos[sectorID].pieceSizes, size, r)
if err != nil {
return xerrors.Errorf("writing piece: %w", err)
}
piece := Piece{
Piece: ppi,
DealInfo: di,
}
err = m.sectors.Send(uint64(sectorID), SectorAddPiece{NewPiece: piece})
if err != nil {
return err
}
ui := m.unsealedInfoMap.infos[sectorID]
num := m.unsealedInfoMap.infos[sectorID].numDeals
if di != nil {
num = num + 1
}
m.unsealedInfoMap.infos[sectorID] = UnsealedSectorInfo{
numDeals: num,
stored: ui.stored + piece.Piece.Size,
pieceSizes: append(ui.pieceSizes, piece.Piece.Size.Unpadded()),
ssize: ssize,
}
return nil
}
func (m *Sealing) Remove(ctx context.Context, sid abi.SectorNumber) error {
return m.sectors.Send(uint64(sid), SectorRemove{})
}
@ -298,183 +202,6 @@ func (m *Sealing) TerminatePending(ctx context.Context) ([]abi.SectorID, error)
return m.terminator.Pending(ctx)
}
// Caller should NOT hold m.unsealedInfoMap.lk
func (m *Sealing) StartPacking(sectorID abi.SectorNumber) error {
// locking here ensures that when the SectorStartPacking event is sent, the sector won't be picked up anywhere else
m.unsealedInfoMap.lk.Lock()
defer m.unsealedInfoMap.lk.Unlock()
// cannot send SectorStartPacking to sectors that have already been packed, otherwise it will cause the state machine to exit
if _, ok := m.unsealedInfoMap.infos[sectorID]; !ok {
log.Warnf("call start packing, but sector %v not in unsealedInfoMap.infos, maybe have called", sectorID)
return nil
}
log.Infof("Starting packing sector %d", sectorID)
err := m.sectors.Send(uint64(sectorID), SectorStartPacking{})
if err != nil {
return err
}
log.Infof("send Starting packing event success sector %d", sectorID)
delete(m.unsealedInfoMap.infos, sectorID)
return nil
}
// Caller should hold m.unsealedInfoMap.lk
func (m *Sealing) getSectorAndPadding(ctx context.Context, size abi.UnpaddedPieceSize) (abi.SectorNumber, []abi.PaddedPieceSize, error) {
for tries := 0; tries < 100; tries++ {
for k, v := range m.unsealedInfoMap.infos {
pads, padLength := ffiwrapper.GetRequiredPadding(v.stored, size.Padded())
if v.stored+size.Padded()+padLength <= abi.PaddedPieceSize(v.ssize) {
return k, pads, nil
}
}
if len(m.unsealedInfoMap.infos) > 0 {
log.Infow("tried to put a piece into an open sector, found none with enough space", "open", len(m.unsealedInfoMap.infos), "size", size, "tries", tries)
}
ns, ssize, err := m.newDealSector(ctx)
switch err {
case nil:
m.unsealedInfoMap.infos[ns] = UnsealedSectorInfo{
numDeals: 0,
stored: 0,
pieceSizes: nil,
ssize: ssize,
}
case errTooManySealing:
m.unsealedInfoMap.lk.Unlock()
select {
case <-time.After(2 * time.Second):
case <-ctx.Done():
m.unsealedInfoMap.lk.Lock()
return 0, nil, xerrors.Errorf("getting sector for piece: %w", ctx.Err())
}
m.unsealedInfoMap.lk.Lock()
continue
default:
return 0, nil, xerrors.Errorf("creating new sector: %w", err)
}
return ns, nil, nil
}
return 0, nil, xerrors.Errorf("failed to allocate piece to a sector")
}
var errTooManySealing = errors.New("too many sectors sealing")
// newDealSector creates a new sector for deal storage
func (m *Sealing) newDealSector(ctx context.Context) (abi.SectorNumber, abi.SectorSize, error) {
// First make sure we don't have too many 'open' sectors
cfg, err := m.getConfig()
if err != nil {
return 0, 0, xerrors.Errorf("getting config: %w", err)
}
if cfg.MaxSealingSectorsForDeals > 0 {
if m.stats.curSealing() > cfg.MaxSealingSectorsForDeals {
return 0, 0, ErrTooManySectorsSealing
}
}
if cfg.MaxWaitDealsSectors > 0 && uint64(len(m.unsealedInfoMap.infos)) >= cfg.MaxWaitDealsSectors {
// Too many sectors are sealing in parallel. Start sealing one, and retry
// allocating the piece to a sector (we're dropping the lock here, so in
// case other goroutines are also trying to create a sector, we retry in
// getSectorAndPadding instead of here - otherwise if we have lots of
// parallel deals in progress, we can start creating a ton of sectors
// with just a single deal in them)
var mostStored abi.PaddedPieceSize = math.MaxUint64
var best abi.SectorNumber = math.MaxUint64
for sn, info := range m.unsealedInfoMap.infos {
if info.stored+1 > mostStored+1 { // 18446744073709551615 + 1 = 0
best = sn
}
}
if best != math.MaxUint64 {
m.unsealedInfoMap.lk.Unlock()
err := m.StartPacking(best)
m.unsealedInfoMap.lk.Lock()
if err != nil {
log.Errorf("newDealSector StartPacking error: %+v", err)
// let's pretend this is fine
}
}
return 0, 0, errTooManySealing // will wait a bit and retry
}
spt, err := m.currentSealProof(ctx)
if err != nil {
return 0, 0, xerrors.Errorf("getting current seal proof type: %w", err)
}
// Now actually create a new sector
sid, err := m.sc.Next()
if err != nil {
return 0, 0, xerrors.Errorf("getting sector number: %w", err)
}
err = m.sealer.NewSector(context.TODO(), m.minerSector(spt, sid))
if err != nil {
return 0, 0, xerrors.Errorf("initializing sector: %w", err)
}
log.Infof("Creating sector %d", sid)
err = m.sectors.Send(uint64(sid), SectorStart{
ID: sid,
SectorType: spt,
})
if err != nil {
return 0, 0, xerrors.Errorf("starting the sector fsm: %w", err)
}
cf, err := m.getConfig()
if err != nil {
return 0, 0, xerrors.Errorf("getting the sealing delay: %w", err)
}
if cf.WaitDealsDelay > 0 {
timer := time.NewTimer(cf.WaitDealsDelay)
go func() {
<-timer.C
if err := m.StartPacking(sid); err != nil {
log.Errorf("starting sector %d: %+v", sid, err)
}
}()
}
ssize, err := spt.SectorSize()
return sid, ssize, err
}
// newSectorCC accepts a slice of pieces with no deal (junk data)
func (m *Sealing) newSectorCC(ctx context.Context, sid abi.SectorNumber, pieces []Piece) error {
spt, err := m.currentSealProof(ctx)
if err != nil {
return xerrors.Errorf("getting current seal proof type: %w", err)
}
log.Infof("Creating CC sector %d", sid)
return m.sectors.Send(uint64(sid), SectorStartCC{
ID: sid,
Pieces: pieces,
SectorType: spt,
})
}
func (m *Sealing) currentSealProof(ctx context.Context) (abi.RegisteredSealProof, error) {
mi, err := m.api.StateMinerInfo(ctx, m.maddr, nil)
if err != nil {
@ -512,9 +239,9 @@ func (m *Sealing) Address() address.Address {
return m.maddr
}
func getDealPerSectorLimit(size abi.SectorSize) uint64 {
func getDealPerSectorLimit(size abi.SectorSize) (int, error) {
if size < 64<<30 {
return 256
return 256, nil
}
return 512
return 512, nil
}

10
extern/storage-sealing/sector_state.go vendored
View File

@ -6,6 +6,8 @@ var ExistSectorStateList = map[SectorState]struct{}{
Empty: {},
WaitDeals: {},
Packing: {},
AddPiece: {},
AddPieceFailed: {},
GetTicket: {},
PreCommit1: {},
PreCommit2: {},
@ -43,8 +45,9 @@ const (
UndefinedSectorState SectorState = ""
// happy path
Empty SectorState = "Empty"
Empty SectorState = "Empty" // deprecated
WaitDeals SectorState = "WaitDeals" // waiting for more pieces (deals) to be added to the sector
AddPiece SectorState = "AddPiece" // put deal data (and padding if required) into the sector
Packing SectorState = "Packing" // sector not in sealStore, and not on chain
GetTicket SectorState = "GetTicket" // generate ticket
PreCommit1 SectorState = "PreCommit1" // do PreCommit1
@ -59,6 +62,7 @@ const (
Proving SectorState = "Proving"
// error modes
FailedUnrecoverable SectorState = "FailedUnrecoverable"
AddPieceFailed SectorState = "AddPieceFailed"
SealPreCommit1Failed SectorState = "SealPreCommit1Failed"
SealPreCommit2Failed SectorState = "SealPreCommit2Failed"
PreCommitFailed SectorState = "PreCommitFailed"
@ -85,7 +89,9 @@ const (
func toStatState(st SectorState) statSectorState {
switch st {
case Empty, WaitDeals, Packing, GetTicket, PreCommit1, PreCommit2, PreCommitting, PreCommitWait, WaitSeed, Committing, SubmitCommit, CommitWait, FinalizeSector:
case UndefinedSectorState, Empty, WaitDeals, AddPiece:
return sstStaging
case Packing, GetTicket, PreCommit1, PreCommit2, PreCommitting, PreCommitWait, WaitSeed, Committing, SubmitCommit, CommitWait, FinalizeSector:
return sstSealing
case Proving, Removed, Removing, Terminating, TerminateWait, TerminateFinality, TerminateFailed:
return sstProving

21
extern/storage-sealing/states_failed.go vendored
View File

@ -3,6 +3,7 @@ package sealing
import (
"time"
"github.com/hashicorp/go-multierror"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/chain/actors/builtin/market"
@ -352,6 +353,7 @@ func (m *Sealing) handleRecoverDealIDs(ctx statemachine.Context, sector SectorIn
}
var toFix []int
paddingPieces := 0
for i, p := range sector.Pieces {
// if no deal is associated with the piece, ensure that we added it as
@ -361,6 +363,7 @@ func (m *Sealing) handleRecoverDealIDs(ctx statemachine.Context, sector SectorIn
if !p.Piece.PieceCID.Equals(exp) {
return xerrors.Errorf("sector %d piece %d had non-zero PieceCID %+v", sector.SectorNumber, i, p.Piece.PieceCID)
}
paddingPieces++
continue
}
@ -396,6 +399,7 @@ func (m *Sealing) handleRecoverDealIDs(ctx statemachine.Context, sector SectorIn
}
}
failed := map[int]error{}
updates := map[int]abi.DealID{}
for _, i := range toFix {
p := sector.Pieces[i]
@ -414,12 +418,27 @@ func (m *Sealing) handleRecoverDealIDs(ctx statemachine.Context, sector SectorIn
}
res, err := m.dealInfo.GetCurrentDealInfo(ctx.Context(), tok, dp, *p.DealInfo.PublishCid)
if err != nil {
return xerrors.Errorf("recovering deal ID for publish deal message %s (sector %d, piece %d): %w", *p.DealInfo.PublishCid, sector.SectorNumber, i, err)
failed[i] = xerrors.Errorf("getting current deal info for piece %d: %w", i, err)
}
updates[i] = res.DealID
}
if len(failed) > 0 {
var merr error
for _, e := range failed {
merr = multierror.Append(merr, e)
}
if len(failed)+paddingPieces == len(sector.Pieces) {
log.Errorf("removing sector %d: all deals expired or unrecoverable: %+v", sector.SectorNumber, merr)
return ctx.Send(SectorRemove{})
}
// todo: try to remove bad pieces (hard; see the todo above)
return xerrors.Errorf("failed to recover some deals: %w", merr)
}
// Not much to do here, we can't go back in time to commit this sector
return ctx.Send(SectorUpdateDealIDs{Updates: updates})
}

42
extern/storage-sealing/states_sealing.go vendored
View File

@ -25,6 +25,24 @@ var DealSectorPriority = 1024
var MaxTicketAge = abi.ChainEpoch(builtin0.EpochsInDay * 2)
func (m *Sealing) handlePacking(ctx statemachine.Context, sector SectorInfo) error {
m.inputLk.Lock()
// make sure we not accepting deals into this sector
for _, c := range m.assignedPieces[m.minerSectorID(sector.SectorNumber)] {
pp := m.pendingPieces[c]
delete(m.pendingPieces, c)
if pp == nil {
log.Errorf("nil assigned pending piece %s", c)
continue
}
// todo: return to the sealing queue (this is extremely unlikely to happen)
pp.accepted(sector.SectorNumber, 0, xerrors.Errorf("sector entered packing state early"))
}
delete(m.openSectors, m.minerSectorID(sector.SectorNumber))
delete(m.assignedPieces, m.minerSectorID(sector.SectorNumber))
m.inputLk.Unlock()
log.Infow("performing filling up rest of the sector...", "sector", sector.SectorNumber)
var allocated abi.UnpaddedPieceSize
@ -52,7 +70,7 @@ func (m *Sealing) handlePacking(ctx statemachine.Context, sector SectorInfo) err
log.Warnf("Creating %d filler pieces for sector %d", len(fillerSizes), sector.SectorNumber)
}
fillerPieces, err := m.pledgeSector(sector.sealingCtx(ctx.Context()), m.minerSector(sector.SectorType, sector.SectorNumber), sector.existingPieceSizes(), fillerSizes...)
fillerPieces, err := m.padSector(sector.sealingCtx(ctx.Context()), m.minerSector(sector.SectorType, sector.SectorNumber), sector.existingPieceSizes(), fillerSizes...)
if err != nil {
return xerrors.Errorf("filling up the sector (%v): %w", fillerSizes, err)
}
@ -60,6 +78,28 @@ func (m *Sealing) handlePacking(ctx statemachine.Context, sector SectorInfo) err
return ctx.Send(SectorPacked{FillerPieces: fillerPieces})
}
func (m *Sealing) padSector(ctx context.Context, sectorID storage.SectorRef, existingPieceSizes []abi.UnpaddedPieceSize, sizes ...abi.UnpaddedPieceSize) ([]abi.PieceInfo, error) {
if len(sizes) == 0 {
return nil, nil
}
log.Infof("Pledge %d, contains %+v", sectorID, existingPieceSizes)
out := make([]abi.PieceInfo, len(sizes))
for i, size := range sizes {
ppi, err := m.sealer.AddPiece(ctx, sectorID, existingPieceSizes, size, NewNullReader(size))
if err != nil {
return nil, xerrors.Errorf("add piece: %w", err)
}
existingPieceSizes = append(existingPieceSizes, size)
out[i] = ppi
}
return out, nil
}
func checkTicketExpired(sector SectorInfo, epoch abi.ChainEpoch) bool {
return epoch-sector.TicketEpoch > MaxTicketAge // TODO: allow configuring expected seal durations
}

13
extern/storage-sealing/stats.go vendored
View File

@ -9,7 +9,8 @@ import (
type statSectorState int
const (
sstSealing statSectorState = iota
sstStaging statSectorState = iota
sstSealing
sstFailed
sstProving
nsst
@ -41,5 +42,13 @@ func (ss *SectorStats) curSealing() uint64 {
ss.lk.Lock()
defer ss.lk.Unlock()
return ss.totals[sstSealing] + ss.totals[sstFailed]
return ss.totals[sstStaging] + ss.totals[sstSealing] + ss.totals[sstFailed]
}
// return the number of sectors waiting to enter the sealing pipeline
func (ss *SectorStats) curStaging() uint64 {
ss.lk.Lock()
defer ss.lk.Unlock()
return ss.totals[sstStaging]
}

3
extern/storage-sealing/types.go vendored
View File

@ -72,7 +72,8 @@ type SectorInfo struct {
SectorType abi.RegisteredSealProof
// Packing
Pieces []Piece
CreationTime int64 // unix seconds
Pieces []Piece
// PreCommit1
TicketValue abi.SealRandomness

1
go.mod
View File

@ -50,6 +50,7 @@ require (
github.com/gbrlsnchs/jwt/v3 v3.0.0-beta.1
github.com/go-kit/kit v0.10.0
github.com/go-ole/go-ole v1.2.4 // indirect
github.com/golang/mock v1.4.4
github.com/google/uuid v1.1.2
github.com/gorilla/mux v1.7.4
github.com/gorilla/websocket v1.4.2

5
node/config/def.go
View File

@ -72,6 +72,11 @@ type SealingConfig struct {
WaitDealsDelay Duration
AlwaysKeepUnsealedCopy bool
// Keep this many sectors in sealing pipeline, start CC if needed
// todo TargetSealingSectors uint64
// todo TargetSectors - stop auto-pleding new sectors after this many sectors are sealed, default CC upgrade for deals sectors if above
}
type MinerFeeConfig struct {

8
node/impl/full/state.go
View File

@ -1097,9 +1097,9 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
var sectorWeight abi.StoragePower
if act, err := state.GetActor(market.Address); err != nil {
return types.EmptyInt, xerrors.Errorf("loading miner actor %s: %w", maddr, err)
return types.EmptyInt, xerrors.Errorf("loading market actor: %w", err)
} else if s, err := market.Load(store, act); err != nil {
return types.EmptyInt, xerrors.Errorf("loading market actor state %s: %w", maddr, err)
return types.EmptyInt, xerrors.Errorf("loading market actor state: %w", err)
} else if w, vw, err := s.VerifyDealsForActivation(maddr, pci.DealIDs, ts.Height(), pci.Expiration); err != nil {
return types.EmptyInt, xerrors.Errorf("verifying deals for activation: %w", err)
} else {
@ -1113,7 +1113,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
pledgeCollateral abi.TokenAmount
)
if act, err := state.GetActor(power.Address); err != nil {
return types.EmptyInt, xerrors.Errorf("loading miner actor: %w", err)
return types.EmptyInt, xerrors.Errorf("loading power actor: %w", err)
} else if s, err := power.Load(store, act); err != nil {
return types.EmptyInt, xerrors.Errorf("loading power actor state: %w", err)
} else if p, err := s.TotalPowerSmoothed(); err != nil {
@ -1127,7 +1127,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
rewardActor, err := state.GetActor(reward.Address)
if err != nil {
return types.EmptyInt, xerrors.Errorf("loading miner actor: %w", err)
return types.EmptyInt, xerrors.Errorf("loading reward actor: %w", err)
}
rewardState, err := reward.Load(store, rewardActor)

30
node/impl/storminer.go
View File

@ -121,8 +121,30 @@ func (sm *StorageMinerAPI) ActorSectorSize(ctx context.Context, addr address.Add
return mi.SectorSize, nil
}
func (sm *StorageMinerAPI) PledgeSector(ctx context.Context) error {
return sm.Miner.PledgeSector()
func (sm *StorageMinerAPI) PledgeSector(ctx context.Context) (abi.SectorID, error) {
sr, err := sm.Miner.PledgeSector(ctx)
if err != nil {
return abi.SectorID{}, err
}
// wait for the sector to enter the Packing state
// TODO: instead of polling implement some pubsub-type thing in storagefsm
for {
info, err := sm.Miner.GetSectorInfo(sr.ID.Number)
if err != nil {
return abi.SectorID{}, xerrors.Errorf("getting pledged sector info: %w", err)
}
if info.State != sealing.UndefinedSectorState {
return sr.ID, nil
}
select {
case <-time.After(10 * time.Millisecond):
case <-ctx.Done():
return abi.SectorID{}, ctx.Err()
}
}
}
func (sm *StorageMinerAPI) SectorsStatus(ctx context.Context, sid abi.SectorNumber, showOnChainInfo bool) (api.SectorInfo, error) {
@ -219,6 +241,10 @@ func (sm *StorageMinerAPI) SectorsList(context.Context) ([]abi.SectorNumber, err
out := make([]abi.SectorNumber, len(sectors))
for i, sector := range sectors {
if sector.State == sealing.UndefinedSectorState {
continue // sector ID not set yet
}
out[i] = sector.SectorNumber
}
return out, nil

5
storage/sealing.go
View File

@ -8,6 +8,7 @@ import (
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/specs-storage/storage"
sealing "github.com/filecoin-project/lotus/extern/storage-sealing"
)
@ -34,8 +35,8 @@ func (m *Miner) GetSectorInfo(sid abi.SectorNumber) (sealing.SectorInfo, error)
return m.sealing.GetSectorInfo(sid)
}
func (m *Miner) PledgeSector() error {
return m.sealing.PledgeSector()
func (m *Miner) PledgeSector(ctx context.Context) (storage.SectorRef, error) {
return m.sealing.PledgeSector(ctx)
}
func (m *Miner) ForceSectorState(ctx context.Context, id abi.SectorNumber, state sealing.SectorState) error {