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

Merge branch 'testnet/3' into feat/update-markets

Browse Source
This commit is contained in:
Łukasz Magiera 2020-02-04 07:17:18 +01:00
commit 0e81c1cf7e
63 changed files with 1325 additions and 309 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

91
Dockerfile
View File

@ -1,91 +0,0 @@
FROM golang:1.13.4-buster
MAINTAINER ldoublewood <ldoublewood@gmail.com>
ENV SRC_DIR /lotus
RUN apt-get update && apt-get install -y && apt-get install -y ca-certificates llvm clang mesa-opencl-icd ocl-icd-opencl-dev
RUN curl -sSf https://sh.rustup.rs | sh -s -- -y
# Get su-exec, a very minimal tool for dropping privileges,
# and tini, a very minimal init daemon for containers
ENV SUEXEC_VERSION v0.2
ENV TINI_VERSION v0.18.0
RUN set -x \
&& cd /tmp \
&& git clone https://github.com/ncopa/su-exec.git \
&& cd su-exec \
&& git checkout -q $SUEXEC_VERSION \
&& make \
&& cd /tmp \
&& wget -q -O tini https://github.com/krallin/tini/releases/download/$TINI_VERSION/tini \
&& chmod +x tini
# Download packages first so they can be cached.
COPY go.mod go.sum $SRC_DIR/
COPY extern/ $SRC_DIR/extern/
RUN cd $SRC_DIR \
&& go mod download
COPY Makefile $SRC_DIR
# Because extern/filecoin-ffi building script need to get version number from git
COPY .git/ $SRC_DIR/.git/
COPY .gitmodules $SRC_DIR/
# Download dependence first
RUN cd $SRC_DIR \
&& mkdir $SRC_DIR/build \
&& . $HOME/.cargo/env \
&& make clean \
&& make deps
COPY . $SRC_DIR
# Build the thing.
RUN cd $SRC_DIR \
&& . $HOME/.cargo/env \
&& make
# Now comes the actual target image, which aims to be as small as possible.
FROM busybox:1-glibc
MAINTAINER ldoublewood <ldoublewood@gmail.com>
# Get the executable binary and TLS CAs from the build container.
ENV SRC_DIR /lotus
COPY --from=0 $SRC_DIR/lotus /usr/local/bin/lotus
COPY --from=0 $SRC_DIR/lotus-storage-miner /usr/local/bin/lotus-storage-miner
COPY --from=0 /tmp/su-exec/su-exec /sbin/su-exec
COPY --from=0 /tmp/tini /sbin/tini
COPY --from=0 /etc/ssl/certs /etc/ssl/certs
# This shared lib (part of glibc) doesn't seem to be included with busybox.
COPY --from=0 /lib/x86_64-linux-gnu/libdl-2.28.so /lib/libdl.so.2
COPY --from=0 /lib/x86_64-linux-gnu/libutil-2.28.so /lib/libutil.so.1
COPY --from=0 /usr/lib/x86_64-linux-gnu/libOpenCL.so.1.0.0 /lib/libOpenCL.so.1
COPY --from=0 /lib/x86_64-linux-gnu/librt-2.28.so /lib/librt.so.1
COPY --from=0 /lib/x86_64-linux-gnu/libgcc_s.so.1 /lib/libgcc_s.so.1
# WS port
EXPOSE 1234
# P2P port
EXPOSE 5678
# Create the home directory and switch to a non-privileged user.
ENV HOME_PATH /data
ENV PARAMCACHE_PATH /var/tmp/filecoin-proof-parameters
RUN mkdir -p $HOME_PATH \
&& adduser -D -h $HOME_PATH -u 1000 -G users lotus \
&& chown lotus:users $HOME_PATH
VOLUME $HOME_PATH
VOLUME $PARAMCACHE_PATH
# Execute the daemon subcommand by default
CMD ["/sbin/tini", "--", "lotus", "daemon"]

2
README.md
View File

@ -2,7 +2,7 @@
# Project Lotus - 莲
Lotus is an experimental implementation of the Filecoin Distributed Storage Network. For more details about Filecoin, check out the [Filecoin Spec](https://github.com/filecoin-project/specs).
Lotus is an implementation of the Filecoin Distributed Storage Network. For more details about Filecoin, check out the [Filecoin Spec](https://github.com/filecoin-project/specs).
## Development

8
api/api_full.go
View File

@ -4,12 +4,12 @@ import (
"context"
"time"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-filestore"
"github.com/libp2p/go-libp2p-core/peer"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-fil-markets/storagemarket"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
@ -33,6 +33,7 @@ type FullNode interface {
ChainGetParentMessages(context.Context, cid.Cid) ([]Message, error)
ChainGetTipSetByHeight(context.Context, uint64, *types.TipSet) (*types.TipSet, error)
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
ChainHasObj(context.Context, cid.Cid) (bool, error)
ChainSetHead(context.Context, *types.TipSet) error
ChainGetGenesis(context.Context) (*types.TipSet, error)
ChainTipSetWeight(context.Context, *types.TipSet) (types.BigInt, error)
@ -109,6 +110,7 @@ type FullNode interface {
StateMinerPeerID(ctx context.Context, m address.Address, ts *types.TipSet) (peer.ID, error)
StateMinerElectionPeriodStart(ctx context.Context, actor address.Address, ts *types.TipSet) (uint64, error)
StateMinerSectorSize(context.Context, address.Address, *types.TipSet) (uint64, error)
StateMinerFaults(context.Context, address.Address, *types.TipSet) ([]uint64, error)
StatePledgeCollateral(context.Context, *types.TipSet) (types.BigInt, error)
StateWaitMsg(context.Context, cid.Cid) (*MsgWait, error)
StateListMiners(context.Context, *types.TipSet) ([]address.Address, error)
@ -123,6 +125,8 @@ type FullNode interface {
StateMinerSectorCount(context.Context, address.Address, *types.TipSet) (MinerSectors, error)
StateCompute(context.Context, uint64, []*types.Message, *types.TipSet) (cid.Cid, error)
MsigGetAvailableBalance(context.Context, address.Address, *types.TipSet) (types.BigInt, error)
MarketEnsureAvailable(context.Context, address.Address, types.BigInt) error
// MarketFreeBalance

3
api/api_storage.go
View File

@ -22,11 +22,11 @@ const (
WaitSeed // waiting for seed
Committing
CommitWait // waiting for message to land on chain
FinalizeSector
Proving
_ // reserved
_
_
_
// recovery handling
// Reseal
@ -64,6 +64,7 @@ var SectorStates = []string{
WaitSeed: "WaitSeed",
Committing: "Committing",
CommitWait: "CommitWait",
FinalizeSector: "FinalizeSector",
Proving: "Proving",
SealFailed: "SealFailed",

61
api/apibstore/apibstore.go Normal file
View File

@ -0,0 +1,61 @@
package apibstore
import (
"context"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"golang.org/x/xerrors"
)
type ChainIO interface {
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
ChainHasObj(context.Context, cid.Cid) (bool, error)
}
type apiBStore struct {
api ChainIO
}
func (a *apiBStore) DeleteBlock(cid.Cid) error {
return xerrors.New("not supported")
}
func (a *apiBStore) Has(c cid.Cid) (bool, error) {
return a.api.ChainHasObj(context.TODO(), c)
}
func (a *apiBStore) Get(c cid.Cid) (blocks.Block, error) {
bb, err := a.api.ChainReadObj(context.TODO(), c)
if err != nil {
return nil, err
}
return blocks.NewBlockWithCid(bb, c)
}
func (a *apiBStore) GetSize(c cid.Cid) (int, error) {
bb, err := a.api.ChainReadObj(context.TODO(), c)
if err != nil {
return 0, err
}
return len(bb), nil
}
func (a *apiBStore) Put(blocks.Block) error {
return xerrors.New("not supported")
}
func (a *apiBStore) PutMany([]blocks.Block) error {
return xerrors.New("not supported")
}
func (a *apiBStore) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
return nil, xerrors.New("not supported")
}
func (a *apiBStore) HashOnRead(enabled bool) {
return
}
var _ blockstore.Blockstore = &apiBStore{}

16
api/apistruct/struct.go
View File

@ -50,6 +50,7 @@ type FullNodeStruct struct {
ChainGetParentMessages func(context.Context, cid.Cid) ([]api.Message, error) `perm:"read"`
ChainGetTipSetByHeight func(context.Context, uint64, *types.TipSet) (*types.TipSet, error) `perm:"read"`
ChainReadObj func(context.Context, cid.Cid) ([]byte, error) `perm:"read"`
ChainHasObj func(context.Context, cid.Cid) (bool, error) `perm:"read"`
ChainSetHead func(context.Context, *types.TipSet) error `perm:"admin"`
ChainGetGenesis func(context.Context) (*types.TipSet, error) `perm:"read"`
ChainTipSetWeight func(context.Context, *types.TipSet) (types.BigInt, error) `perm:"read"`
@ -99,6 +100,7 @@ type FullNodeStruct struct {
StateMinerPeerID func(ctx context.Context, m address.Address, ts *types.TipSet) (peer.ID, error) `perm:"read"`
StateMinerElectionPeriodStart func(ctx context.Context, actor address.Address, ts *types.TipSet) (uint64, error) `perm:"read"`
StateMinerSectorSize func(context.Context, address.Address, *types.TipSet) (uint64, error) `perm:"read"`
StateMinerFaults func(context.Context, address.Address, *types.TipSet) ([]uint64, error) `perm:"read"`
StateCall func(context.Context, *types.Message, *types.TipSet) (*api.MethodCall, error) `perm:"read"`
StateReplay func(context.Context, *types.TipSet, cid.Cid) (*api.ReplayResults, error) `perm:"read"`
StateGetActor func(context.Context, address.Address, *types.TipSet) (*types.Actor, error) `perm:"read"`
@ -118,6 +120,8 @@ type FullNodeStruct struct {
StateListMessages func(ctx context.Context, match *types.Message, ts *types.TipSet, toht uint64) ([]cid.Cid, error) `perm:"read"`
StateCompute func(context.Context, uint64, []*types.Message, *types.TipSet) (cid.Cid, error) `perm:"read"`
MsigGetAvailableBalance func(context.Context, address.Address, *types.TipSet) (types.BigInt, error) `perm:"read"`
MarketEnsureAvailable func(context.Context, address.Address, types.BigInt) error `perm:"sign"`
PaychGet func(ctx context.Context, from, to address.Address, ensureFunds types.BigInt) (*api.ChannelInfo, error) `perm:"sign"`
@ -338,6 +342,10 @@ func (c *FullNodeStruct) ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte,
return c.Internal.ChainReadObj(ctx, obj)
}
func (c *FullNodeStruct) ChainHasObj(ctx context.Context, o cid.Cid) (bool, error) {
return c.Internal.ChainHasObj(ctx, o)
}
func (c *FullNodeStruct) ChainSetHead(ctx context.Context, ts *types.TipSet) error {
return c.Internal.ChainSetHead(ctx, ts)
}
@ -410,6 +418,10 @@ func (c *FullNodeStruct) StateMinerSectorSize(ctx context.Context, actor address
return c.Internal.StateMinerSectorSize(ctx, actor, ts)
}
func (c *FullNodeStruct) StateMinerFaults(ctx context.Context, actor address.Address, ts *types.TipSet) ([]uint64, error) {
return c.Internal.StateMinerFaults(ctx, actor, ts)
}
func (c *FullNodeStruct) StateCall(ctx context.Context, msg *types.Message, ts *types.TipSet) (*api.MethodCall, error) {
return c.Internal.StateCall(ctx, msg, ts)
}
@ -477,6 +489,10 @@ func (c *FullNodeStruct) StateCompute(ctx context.Context, height uint64, msgs [
return c.Internal.StateCompute(ctx, height, msgs, ts)
}
func (c *FullNodeStruct) MsigGetAvailableBalance(ctx context.Context, a address.Address, ts *types.TipSet) (types.BigInt, error) {
return c.Internal.MsigGetAvailableBalance(ctx, a, ts)
}
func (c *FullNodeStruct) MarketEnsureAvailable(ctx context.Context, addr address.Address, amt types.BigInt) error {
return c.Internal.MarketEnsureAvailable(ctx, addr, amt)
}

2
build/forks.go
View File

@ -5,3 +5,5 @@ const ForkBlizzardHeight = 6288
const ForkFrigidHeight = 7950
const ForkBootyBayHeight = 11000
const ForkMissingSnowballs = 34000

4
build/parameters.go
View File

@ -2,4 +2,6 @@ package build
import rice "github.com/GeertJohan/go.rice"
var ParametersJson = rice.MustFindBox("proof-params").MustBytes("parameters.json")
func ParametersJson() []byte {
return rice.MustFindBox("proof-params").MustBytes("parameters.json")
}

4
build/params_debug.go
View File

@ -2,6 +2,10 @@
package build
func init() {
InsecurePoStValidation = true
}
var SectorSizes = []uint64{1024}
// Seconds

2
build/version.go
View File

@ -5,7 +5,7 @@ import "fmt"
var CurrentCommit string
// BuildVersion is the local build version, set by build system
const BuildVersion = "0.2.6"
const BuildVersion = "0.2.7"
var UserVersion = BuildVersion + CurrentCommit

9
chain/actors/actor_init.go
View File

@ -27,6 +27,8 @@ const (
GasCreateActor = 100
)
var BuiltInActors map[cid.Cid]bool
func init() {
n, err := cbor.WrapObject(map[string]string{}, mh.SHA2_256, -1)
@ -160,12 +162,7 @@ func (ia InitActor) Exec(act *types.Actor, vmctx types.VMContext, p *ExecParams)
}
func IsBuiltinActor(code cid.Cid) bool {
switch code {
case StorageMarketCodeCid, StoragePowerCodeCid, StorageMinerCodeCid, StorageMiner2CodeCid, AccountCodeCid, InitCodeCid, MultisigCodeCid, PaymentChannelCodeCid:
return true
default:
return false
}
return BuiltInActors[code]
}
func IsSingletonActor(code cid.Cid) bool {

43
chain/actors/actor_miner2.go
View File

@ -33,7 +33,10 @@ func (sma StorageMinerActor2) Exports() []interface{} {
//8: sma.DePledge,
9: sma.GetOwner,
10: sma.GetWorkerAddr,
11: sma.GetPower, // TODO: Remove
11: withUpdates(
update{0, sma.GetPower},
update{build.ForkMissingSnowballs, sma.GetPower2},
), // FORK
12: sma.GetPeerID,
13: sma.GetSectorSize,
14: sma.UpdatePeerID,
@ -401,6 +404,20 @@ func (sma StorageMinerActor2) GetPower(act *types.Actor, vmctx types.VMContext,
if err != nil {
return nil, err
}
return self.Power.Bytes(), nil
}
func (sma StorageMinerActor2) GetPower2(act *types.Actor, vmctx types.VMContext, params *struct{}) ([]byte, ActorError) {
_, self, err := loadState(vmctx)
if err != nil {
return nil, err
}
if self.SlashedAt != 0 {
return types.NewInt(0).Bytes(), nil
}
return self.Power.Bytes(), nil
}
@ -615,6 +632,11 @@ func (sma StorageMinerActor2) CheckMiner(act *types.Actor, vmctx types.VMContext
// Slash for being late
self.SlashedAt = vmctx.BlockHeight()
oldPower := self.Power
if vmctx.BlockHeight() > build.ForkMissingSnowballs {
self.Power = types.NewInt(0)
}
nstate, err := vmctx.Storage().Put(self)
if err != nil {
@ -625,7 +647,7 @@ func (sma StorageMinerActor2) CheckMiner(act *types.Actor, vmctx types.VMContext
}
var out bytes.Buffer
if err := self.Power.MarshalCBOR(&out); err != nil {
if err := oldPower.MarshalCBOR(&out); err != nil {
return nil, aerrors.HandleExternalError(err, "marshaling return value")
}
return out.Bytes(), nil
@ -710,6 +732,23 @@ func (sma StorageMinerActor2) SlashConsensusFault(act *types.Actor, vmctx types.
return nil, aerrors.Wrap(err, "failed to burn funds")
}
if vmctx.BlockHeight() > build.ForkMissingSnowballs {
oldstate, self, err := loadState(vmctx)
if err != nil {
return nil, aerrors.Wrap(err, "failed to load state for slashing")
}
self.Power = types.NewInt(0)
ncid, err := vmctx.Storage().Put(self)
if err != nil {
return nil, err
}
if err := vmctx.Storage().Commit(oldstate, ncid); err != nil {
return nil, err
}
}
// TODO: this still allows the miner to commit sectors and submit posts,
// their users could potentially be unaffected, but the miner will never be
// able to mine a block again

50
chain/actors/actor_storagepower.go
View File

@ -648,6 +648,8 @@ func checkProofSubmissionsAtH(vmctx types.VMContext, self *StoragePowerState, he
return aerrors.HandleExternalError(err, "failed to load proving bucket")
}
forRemoval := make([]address.Address, 0)
err = bhamt.ForEach(vmctx.Context(), func(k string, val interface{}) error {
_, span := trace.StartSpan(vmctx.Context(), "StoragePowerActor.CheckProofSubmissions.loop")
defer span.End()
@ -657,6 +659,18 @@ func checkProofSubmissionsAtH(vmctx types.VMContext, self *StoragePowerState, he
return aerrors.Escalate(err, "parsing miner address")
}
if vmctx.BlockHeight() > build.ForkMissingSnowballs {
has, aerr := MinerSetHas(vmctx, self.Miners, maddr)
if aerr != nil {
return aerr
}
if !has {
forRemoval = append(forRemoval, maddr)
}
}
span.AddAttributes(trace.StringAttribute("miner", maddr.String()))
params, err := SerializeParams(&CheckMinerParams{NetworkPower: self.TotalStorage})
@ -690,6 +704,24 @@ func checkProofSubmissionsAtH(vmctx types.VMContext, self *StoragePowerState, he
return aerrors.HandleExternalError(err, "iterating miners in proving bucket")
}
if vmctx.BlockHeight() > build.ForkMissingSnowballs && len(forRemoval) > 0 {
nBucket, err := MinerSetRemove(vmctx.Context(), vmctx, bucket, forRemoval...)
if err != nil {
return aerrors.Wrap(err, "could not remove miners from set")
}
eerr := buckets.Set(bucketID, nBucket)
if err != nil {
return aerrors.HandleExternalError(eerr, "could not set the bucket")
}
ncid, eerr := buckets.Flush()
if err != nil {
return aerrors.HandleExternalError(eerr, "could not flush buckets")
}
self.ProvingBuckets = ncid
}
return nil
}
@ -764,19 +796,21 @@ func MinerSetAdd(ctx context.Context, vmctx types.VMContext, rcid cid.Cid, maddr
return c, nil
}
func MinerSetRemove(ctx context.Context, vmctx types.VMContext, rcid cid.Cid, maddr address.Address) (cid.Cid, aerrors.ActorError) {
func MinerSetRemove(ctx context.Context, vmctx types.VMContext, rcid cid.Cid, maddrs ...address.Address) (cid.Cid, aerrors.ActorError) {
nd, err := hamt.LoadNode(ctx, vmctx.Ipld(), rcid)
if err != nil {
return cid.Undef, aerrors.HandleExternalError(err, "failed to load miner set")
}
mkey := string(maddr.Bytes())
switch nd.Delete(ctx, mkey) {
default:
return cid.Undef, aerrors.HandleExternalError(err, "failed to delete miner from set")
case hamt.ErrNotFound:
return cid.Undef, aerrors.New(1, "miner not found in set on delete")
case nil:
for _, maddr := range maddrs {
mkey := string(maddr.Bytes())
switch nd.Delete(ctx, mkey) {
default:
return cid.Undef, aerrors.HandleExternalError(err, "failed to delete miner from set")
case hamt.ErrNotFound:
return cid.Undef, aerrors.New(1, "miner not found in set on delete")
case nil:
}
}
if err := nd.Flush(ctx); err != nil {

11
chain/actors/actors.go
View File

@ -51,4 +51,15 @@ func init() {
MultisigCodeCid = mustSum("fil/1/multisig")
InitCodeCid = mustSum("fil/1/init")
PaymentChannelCodeCid = mustSum("fil/1/paych")
BuiltInActors = map[cid.Cid]bool{
StorageMarketCodeCid: true,
StoragePowerCodeCid: true,
StorageMinerCodeCid: true,
StorageMiner2CodeCid: true,
AccountCodeCid: true,
InitCodeCid: true,
MultisigCodeCid: true,
PaymentChannelCodeCid: true,
}
}

2
chain/actors/forks.go
View File

@ -26,7 +26,7 @@ func withUpdates(updates ...update) interface{} {
vmctx := args[1].Interface().(types.VMContext)
for _, u := range updates {
if vmctx.BlockHeight() >= u.start {
if vmctx.BlockHeight() > u.start {
return reflect.ValueOf(u.method).Call(args)
}
}

34
chain/gen/gen.go
View File

@ -6,10 +6,11 @@ import (
"crypto/sha256"
"encoding/binary"
"fmt"
"github.com/filecoin-project/lotus/chain/vm"
"io/ioutil"
"sync/atomic"
"github.com/filecoin-project/lotus/chain/vm"
ffi "github.com/filecoin-project/filecoin-ffi"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
@ -56,6 +57,8 @@ type ChainGen struct {
Timestamper func(*types.TipSet, uint64) uint64
GetMessages func(*ChainGen) ([]*types.SignedMessage, error)
w *wallet.Wallet
eppProvs map[address.Address]ElectionPoStProver
@ -210,10 +213,11 @@ func NewGenerator() (*ChainGen, error) {
genesis: genb.Genesis,
w: w,
Miners: minercfg.MinerAddrs,
eppProvs: mgen,
banker: banker,
receivers: receievers,
GetMessages: getRandomMessages,
Miners: minercfg.MinerAddrs,
eppProvs: mgen,
banker: banker,
receivers: receievers,
CurTipset: gents,
@ -224,6 +228,14 @@ func NewGenerator() (*ChainGen, error) {
return gen, nil
}
func (cg *ChainGen) SetStateManager(sm *stmgr.StateManager) {
cg.sm = sm
}
func (cg *ChainGen) ChainStore() *store.ChainStore {
return cg.cs
}
func (cg *ChainGen) Genesis() *types.BlockHeader {
return cg.genesis
}
@ -295,7 +307,7 @@ func (cg *ChainGen) NextTipSet() (*MinedTipSet, error) {
func (cg *ChainGen) NextTipSetFromMiners(base *types.TipSet, miners []address.Address) (*MinedTipSet, error) {
var blks []*types.FullBlock
msgs, err := cg.getRandomMessages()
msgs, err := cg.GetMessages(cg)
if err != nil {
return nil, xerrors.Errorf("get random messages: %w", err)
}
@ -359,7 +371,15 @@ func (cg *ChainGen) ResyncBankerNonce(ts *types.TipSet) error {
return nil
}
func (cg *ChainGen) getRandomMessages() ([]*types.SignedMessage, error) {
func (cg *ChainGen) Banker() address.Address {
return cg.banker
}
func (cg *ChainGen) Wallet() *wallet.Wallet {
return cg.w
}
func getRandomMessages(cg *ChainGen) ([]*types.SignedMessage, error) {
msgs := make([]*types.SignedMessage, cg.msgsPerBlock)
for m := range msgs {
msg := types.Message{

109
chain/stmgr/forks.go
View File

@ -7,6 +7,7 @@ import (
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/types"
"github.com/ipfs/go-cid"
hamt "github.com/ipfs/go-hamt-ipld"
blockstore "github.com/ipfs/go-ipfs-blockstore"
@ -14,33 +15,107 @@ import (
"golang.org/x/xerrors"
)
var ForksAtHeight = map[uint64]func(context.Context, *StateManager, cid.Cid) (cid.Cid, error){
build.ForkBlizzardHeight: func(ctx context.Context, sm *StateManager, pstate cid.Cid) (cid.Cid, error) {
log.Warnw("Executing blizzard fork logic")
nstate, err := fixBlizzardAMTBug(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("blizzard bug fix failed: %w", err)
}
return nstate, nil
},
build.ForkFrigidHeight: func(ctx context.Context, sm *StateManager, pstate cid.Cid) (cid.Cid, error) {
log.Warnw("Executing frigid fork logic")
nstate, err := fixBlizzardAMTBug(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("frigid bug fix failed: %w", err)
}
return nstate, nil
},
build.ForkBootyBayHeight: func(ctx context.Context, sm *StateManager, pstate cid.Cid) (cid.Cid, error) {
log.Warnw("Executing booty bay fork logic")
nstate, err := fixBlizzardAMTBug(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("booty bay bug fix failed: %w", err)
}
return nstate, nil
},
build.ForkMissingSnowballs: func(ctx context.Context, sm *StateManager, pstate cid.Cid) (cid.Cid, error) {
log.Warnw("Adding more snow to the world")
nstate, err := fixTooFewSnowballs(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("missing snowballs bug fix failed: %w", err)
}
return nstate, nil
},
}
func (sm *StateManager) handleStateForks(ctx context.Context, pstate cid.Cid, height, parentH uint64) (_ cid.Cid, err error) {
for i := parentH; i < height; i++ {
switch i {
case build.ForkBlizzardHeight:
log.Warnw("Executing blizzard fork logic", "height", i)
pstate, err = fixBlizzardAMTBug(ctx, sm, pstate)
f, ok := ForksAtHeight[i]
if ok {
nstate, err := f(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("blizzard bug fix failed: %w", err)
}
case build.ForkFrigidHeight:
log.Warnw("Executing frigid fork logic", "height", i)
pstate, err = fixBlizzardAMTBug(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("frigid bug fix failed: %w", err)
}
case build.ForkBootyBayHeight:
log.Warnw("Executing booty bay fork logic", "height", i)
pstate, err = fixBlizzardAMTBug(ctx, sm, pstate)
if err != nil {
return cid.Undef, xerrors.Errorf("booty bay bug fix failed: %w", err)
return cid.Undef, err
}
pstate = nstate
}
}
return pstate, nil
}
func fixTooFewSnowballs(ctx context.Context, sm *StateManager, pstate cid.Cid) (cid.Cid, error) {
cst := hamt.CSTFromBstore(sm.cs.Blockstore())
st, err := state.LoadStateTree(cst, pstate)
if err != nil {
return cid.Undef, err
}
spa, err := st.GetActor(actors.StoragePowerAddress)
if err != nil {
return cid.Undef, xerrors.Errorf("failed to get storage power actor: %w", err)
}
var spast actors.StoragePowerState
if err := cst.Get(ctx, spa.Head, &spast); err != nil {
return cid.Undef, err
}
miners, err := actors.MinerSetList(ctx, cst, spast.Miners)
if err != nil {
return cid.Undef, err
}
sum := types.NewInt(0)
for _, m := range miners {
mact, err := st.GetActor(m)
if err != nil {
return cid.Undef, xerrors.Errorf("getting miner actor to fix: %w", err)
}
var mstate actors.StorageMinerActorState
if err := cst.Get(ctx, mact.Head, &mstate); err != nil {
return cid.Undef, xerrors.Errorf("failed to load miner actor state: %w", err)
}
if mstate.SlashedAt != 0 {
continue
}
sum = types.BigAdd(sum, mstate.Power)
}
spast.TotalStorage = sum
nspahead, err := cst.Put(ctx, &spast)
if err != nil {
return cid.Undef, err
}
spa.Head = nspahead
return st.Flush(ctx)
}
/*
1) Iterate through each miner in the chain:
1.1) Fixup their sector set and proving set

231
chain/stmgr/forks_test.go Normal file
View File

@ -0,0 +1,231 @@
package stmgr_test
import (
"context"
"fmt"
"io"
"testing"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/actors/aerrors"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/stmgr"
. "github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/chain/vm"
"github.com/ipfs/go-cid"
hamt "github.com/ipfs/go-hamt-ipld"
blockstore "github.com/ipfs/go-ipfs-blockstore"
logging "github.com/ipfs/go-log"
mh "github.com/multiformats/go-multihash"
cbg "github.com/whyrusleeping/cbor-gen"
)
func init() {
build.SectorSizes = []uint64{1024}
build.MinimumMinerPower = 1024
}
const testForkHeight = 40
type testActor struct {
}
type testActorState struct {
HasUpgraded uint64
}
func (tas *testActorState) MarshalCBOR(w io.Writer) error {
return cbg.CborWriteHeader(w, cbg.MajUnsignedInt, tas.HasUpgraded)
}
func (tas *testActorState) UnmarshalCBOR(r io.Reader) error {
t, v, err := cbg.CborReadHeader(r)
if err != nil {
return err
}
if t != cbg.MajUnsignedInt {
return fmt.Errorf("wrong type in test actor state")
}
tas.HasUpgraded = v
return nil
}
func (ta *testActor) Exports() []interface{} {
return []interface{}{
1: ta.Constructor,
2: ta.TestMethod,
}
}
func (ta *testActor) Constructor(act *types.Actor, vmctx types.VMContext, params *struct{}) ([]byte, aerrors.ActorError) {
c, err := vmctx.Storage().Put(&testActorState{11})
if err != nil {
return nil, err
}
fmt.Println("NEW ACTOR ADDRESS IS: ", vmctx.Message().To.String())
return nil, vmctx.Storage().Commit(actors.EmptyCBOR, c)
}
func (ta *testActor) TestMethod(act *types.Actor, vmctx types.VMContext, params *struct{}) ([]byte, aerrors.ActorError) {
var st testActorState
if err := vmctx.Storage().Get(vmctx.Storage().GetHead(), &st); err != nil {
return nil, err
}
if vmctx.BlockHeight() > testForkHeight {
if st.HasUpgraded != 55 {
return nil, aerrors.Fatal("fork updating applied in wrong order")
}
} else {
if st.HasUpgraded != 11 {
return nil, aerrors.Fatal("fork updating happened too early")
}
}
return nil, nil
}
func TestForkHeightTriggers(t *testing.T) {
logging.SetAllLoggers(logging.LevelInfo)
ctx := context.TODO()
cg, err := gen.NewGenerator()
if err != nil {
t.Fatal(err)
}
sm := NewStateManager(cg.ChainStore())
inv := vm.NewInvoker()
pref := cid.NewPrefixV1(cid.Raw, mh.IDENTITY)
actcid, err := pref.Sum([]byte("testactor"))
if err != nil {
t.Fatal(err)
}
actors.BuiltInActors[actcid] = true
// predicting the address here... may break if other assumptions change
taddr, err := address.NewIDAddress(1000)
if err != nil {
t.Fatal(err)
}
stmgr.ForksAtHeight[testForkHeight] = func(ctx context.Context, sm *StateManager, pstate cid.Cid) (cid.Cid, error) {
cst := hamt.CSTFromBstore(sm.ChainStore().Blockstore())
st, err := state.LoadStateTree(cst, pstate)
if err != nil {
return cid.Undef, err
}
act, err := st.GetActor(taddr)
if err != nil {
return cid.Undef, err
}
var tas testActorState
if err := cst.Get(ctx, act.Head, &tas); err != nil {
return cid.Undef, err
}
tas.HasUpgraded = 55
ns, err := cst.Put(ctx, &tas)
if err != nil {
return cid.Undef, err
}
act.Head = ns
if err := st.SetActor(taddr, act); err != nil {
return cid.Undef, err
}
return st.Flush(ctx)
}
inv.Register(actcid, &testActor{}, &testActorState{})
sm.SetVMConstructor(func(c cid.Cid, h uint64, r vm.Rand, a address.Address, b blockstore.Blockstore, s *types.VMSyscalls) (*vm.VM, error) {
nvm, err := vm.NewVM(c, h, r, a, b, s)
if err != nil {
return nil, err
}
nvm.SetInvoker(inv)
return nvm, nil
})
cg.SetStateManager(sm)
var msgs []*types.SignedMessage
enc, err := actors.SerializeParams(&actors.ExecParams{Code: actcid})
if err != nil {
t.Fatal(err)
}
m := &types.Message{
From: cg.Banker(),
To: actors.InitAddress,
Method: actors.IAMethods.Exec,
Params: enc,
GasLimit: types.NewInt(10000),
GasPrice: types.NewInt(0),
}
sig, err := cg.Wallet().Sign(ctx, cg.Banker(), m.Cid().Bytes())
if err != nil {
t.Fatal(err)
}
msgs = append(msgs, &types.SignedMessage{
Signature: *sig,
Message: *m,
})
nonce := uint64(1)
cg.GetMessages = func(cg *gen.ChainGen) ([]*types.SignedMessage, error) {
if len(msgs) > 0 {
fmt.Println("added construct method")
m := msgs
msgs = nil
return m, nil
}
m := &types.Message{
From: cg.Banker(),
To: taddr,
Method: 2,
Params: nil,
Nonce: nonce,
GasLimit: types.NewInt(10000),
GasPrice: types.NewInt(0),
}
nonce++
sig, err := cg.Wallet().Sign(ctx, cg.Banker(), m.Cid().Bytes())
if err != nil {
return nil, err
}
return []*types.SignedMessage{
&types.SignedMessage{
Signature: *sig,
Message: *m,
},
}, nil
}
for i := 0; i < 50; i++ {
_, err = cg.NextTipSet()
if err != nil {
t.Fatal(err)
}
}
}

9
chain/stmgr/stmgr.go
View File

@ -18,6 +18,7 @@ import (
bls "github.com/filecoin-project/filecoin-ffi"
"github.com/ipfs/go-cid"
hamt "github.com/ipfs/go-hamt-ipld"
blockstore "github.com/ipfs/go-ipfs-blockstore"
logging "github.com/ipfs/go-log/v2"
"go.opencensus.io/trace"
)
@ -30,10 +31,12 @@ type StateManager struct {
stCache map[string][]cid.Cid
compWait map[string]chan struct{}
stlk sync.Mutex
newVM func(cid.Cid, uint64, vm.Rand, address.Address, blockstore.Blockstore, *types.VMSyscalls) (*vm.VM, error)
}
func NewStateManager(cs *store.ChainStore) *StateManager {
return &StateManager{
newVM: vm.NewVM,
cs: cs,
stCache: make(map[string][]cid.Cid),
compWait: make(map[string]chan struct{}),
@ -139,7 +142,7 @@ func (sm *StateManager) computeTipSetState(ctx context.Context, blks []*types.Bl
r := store.NewChainRand(sm.cs, cids, blks[0].Height)
vmi, err := vm.NewVM(pstate, blks[0].Height, r, address.Undef, sm.cs.Blockstore(), sm.cs.VMSys())
vmi, err := sm.newVM(pstate, blks[0].Height, r, address.Undef, sm.cs.Blockstore(), sm.cs.VMSys())
if err != nil {
return cid.Undef, cid.Undef, xerrors.Errorf("instantiating VM failed: %w", err)
}
@ -636,3 +639,7 @@ func (sm *StateManager) ValidateChain(ctx context.Context, ts *types.TipSet) err
return nil
}
func (sm *StateManager) SetVMConstructor(nvm func(cid.Cid, uint64, vm.Rand, address.Address, blockstore.Blockstore, *types.VMSyscalls) (*vm.VM, error)) {
sm.newVM = nvm
}

17
chain/stmgr/utils.go
View File

@ -2,6 +2,8 @@ package stmgr
import (
"context"
amt2 "github.com/filecoin-project/go-amt-ipld/v2"
"github.com/filecoin-project/lotus/chain/actors/aerrors"
ffi "github.com/filecoin-project/filecoin-ffi"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
@ -253,6 +255,21 @@ func GetMinerSlashed(ctx context.Context, sm *StateManager, ts *types.TipSet, ma
return mas.SlashedAt, nil
}
func GetMinerFaults(ctx context.Context, sm *StateManager, ts *types.TipSet, maddr address.Address) ([]uint64, error) {
var mas actors.StorageMinerActorState
_, err := sm.LoadActorState(ctx, maddr, &mas, ts)
if err != nil {
return nil, xerrors.Errorf("(get ssize) failed to load miner actor state: %w", err)
}
ss, lerr := amt2.LoadAMT(amt.WrapBlockstore(sm.cs.Blockstore()), mas.Sectors)
if lerr != nil {
return nil, aerrors.HandleExternalError(lerr, "could not load proving set node")
}
return mas.FaultSet.All(2 * ss.Count)
}
func GetStorageDeal(ctx context.Context, sm *StateManager, dealId uint64, ts *types.TipSet) (*actors.OnChainDeal, error) {
var state actors.StorageMarketState
if _, err := sm.LoadActorState(ctx, actors.StorageMarketAddress, &state, ts); err != nil {

26
chain/types/blockheader.go
View File

@ -34,31 +34,31 @@ type EPostProof struct {
}
type BlockHeader struct {
Miner address.Address
Miner address.Address // 0
Ticket *Ticket
Ticket *Ticket // 1
EPostProof EPostProof
EPostProof EPostProof // 2
Parents []cid.Cid
Parents []cid.Cid // 3
ParentWeight BigInt
ParentWeight BigInt // 4
Height uint64
Height uint64 // 5
ParentStateRoot cid.Cid
ParentStateRoot cid.Cid // 6
ParentMessageReceipts cid.Cid
ParentMessageReceipts cid.Cid // 7
Messages cid.Cid
Messages cid.Cid // 8
BLSAggregate Signature
BLSAggregate Signature // 9
Timestamp uint64
Timestamp uint64 // 10
BlockSig *Signature
BlockSig *Signature // 11
ForkSignaling uint64
ForkSignaling uint64 // 12
}
func (b *BlockHeader) ToStorageBlock() (block.Block, error) {

22
chain/vm/invoker.go
View File

@ -23,21 +23,21 @@ type invoker struct {
type invokeFunc func(act *types.Actor, vmctx types.VMContext, params []byte) ([]byte, aerrors.ActorError)
type nativeCode []invokeFunc
func newInvoker() *invoker {
func NewInvoker() *invoker {
inv := &invoker{
builtInCode: make(map[cid.Cid]nativeCode),
builtInState: make(map[cid.Cid]reflect.Type),
}
// add builtInCode using: register(cid, singleton)
inv.register(actors.InitCodeCid, actors.InitActor{}, actors.InitActorState{})
inv.register(actors.CronCodeCid, actors.CronActor{}, actors.CronActorState{})
inv.register(actors.StoragePowerCodeCid, actors.StoragePowerActor{}, actors.StoragePowerState{})
inv.register(actors.StorageMarketCodeCid, actors.StorageMarketActor{}, actors.StorageMarketState{})
inv.register(actors.StorageMinerCodeCid, actors.StorageMinerActor{}, actors.StorageMinerActorState{})
inv.register(actors.StorageMiner2CodeCid, actors.StorageMinerActor2{}, actors.StorageMinerActorState{})
inv.register(actors.MultisigCodeCid, actors.MultiSigActor{}, actors.MultiSigActorState{})
inv.register(actors.PaymentChannelCodeCid, actors.PaymentChannelActor{}, actors.PaymentChannelActorState{})
inv.Register(actors.InitCodeCid, actors.InitActor{}, actors.InitActorState{})
inv.Register(actors.CronCodeCid, actors.CronActor{}, actors.CronActorState{})
inv.Register(actors.StoragePowerCodeCid, actors.StoragePowerActor{}, actors.StoragePowerState{})
inv.Register(actors.StorageMarketCodeCid, actors.StorageMarketActor{}, actors.StorageMarketState{})
inv.Register(actors.StorageMinerCodeCid, actors.StorageMinerActor{}, actors.StorageMinerActorState{})
inv.Register(actors.StorageMiner2CodeCid, actors.StorageMinerActor2{}, actors.StorageMinerActorState{})
inv.Register(actors.MultisigCodeCid, actors.MultiSigActor{}, actors.MultiSigActorState{})
inv.Register(actors.PaymentChannelCodeCid, actors.PaymentChannelActor{}, actors.PaymentChannelActorState{})
return inv
}
@ -60,7 +60,7 @@ func (inv *invoker) Invoke(act *types.Actor, vmctx types.VMContext, method uint6
}
func (inv *invoker) register(c cid.Cid, instance Invokee, state interface{}) {
func (inv *invoker) Register(c cid.Cid, instance Invokee, state interface{}) {
code, err := inv.transform(instance)
if err != nil {
panic(err)
@ -165,7 +165,7 @@ func DumpActorState(code cid.Cid, b []byte) (interface{}, error) {
return nil, nil
}
i := newInvoker() // TODO: register builtins in init block
i := NewInvoker() // TODO: register builtins in init block
typ, ok := i.builtInState[code]
if !ok {

6
chain/vm/vm.go
View File

@ -327,7 +327,7 @@ func NewVM(base cid.Cid, height uint64, r Rand, maddr address.Address, cbs block
buf: buf,
blockHeight: height,
blockMiner: maddr,
inv: newInvoker(),
inv: NewInvoker(),
rand: r, // TODO: Probably should be a syscall
Syscalls: syscalls,
}, nil
@ -671,6 +671,10 @@ func Transfer(from, to *types.Actor, amt types.BigInt) error {
return nil
}
func (vm *VM) SetInvoker(i *invoker) {
vm.inv = i
}
func deductFunds(act *types.Actor, amt types.BigInt) error {
if act.Balance.LessThan(amt) {
return fmt.Errorf("not enough funds")

119
cli/chain.go
View File

@ -1,10 +1,13 @@
package cli
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
"os/exec"
"strconv"
"strings"
"time"
@ -28,6 +31,7 @@ var chainCmd = &cli.Command{
chainSetHeadCmd,
chainListCmd,
chainGetCmd,
chainBisectCmd,
chainExportCmd,
slashConsensusFault,
},
@ -388,12 +392,125 @@ func printTipSet(format string, ts *types.TipSet) {
blks += fmt.Sprintf("%s: %s,", b.Cid(), b.Miner)
}
blks += " ]"
sCids := make([]string, 0, len(blks))
for _, c := range ts.Cids() {
sCids = append(sCids, c.String())
}
format = strings.ReplaceAll(format, "<tipset>", strings.Join(sCids, ","))
format = strings.ReplaceAll(format, "<blocks>", blks)
format = strings.ReplaceAll(format, "<weight>", fmt.Sprint(ts.Blocks()[0].ParentWeight))
fmt.Println(format)
}
var chainBisectCmd = &cli.Command{
Name: "bisect",
Usage: "bisect chain for an event",
Description: `Bisect the chain state tree:
lotus chain bisect [min height] [max height] '1/2/3/state/path' 'shell command' 'args'
Returns the first tipset in which condition is true
v
[start] FFFFFFFTTT [end]
Example: find height at which deal ID 100 000 appeared
- lotus chain bisect 1 32000 '@Ha:t03/1' jq -e '.[2] > 100000'
For special path elements see 'chain get' help
`,
Action: func(cctx *cli.Context) error {
api, closer, err := GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := ReqContext(cctx)
if cctx.Args().Len() < 4 {
return xerrors.New("need at least 4 args")
}
start, err := strconv.ParseUint(cctx.Args().Get(0), 10, 64)
if err != nil {
return err
}
end, err := strconv.ParseUint(cctx.Args().Get(1), 10, 64)
if err != nil {
return err
}
subPath := cctx.Args().Get(2)
highest, err := api.ChainGetTipSetByHeight(ctx, end, nil)
if err != nil {
return err
}
prev := highest.Height()
for {
mid := (start + end) / 2
if end - start == 1 {
mid = end
start = end
}
midTs, err := api.ChainGetTipSetByHeight(ctx, mid, highest)
if err != nil {
return err
}
path := "/ipld/" + midTs.ParentState().String() + "/" + subPath
fmt.Printf("* Testing %d (%d - %d) (%s): ", mid, start, end, path)
nd, err := api.ChainGetNode(ctx, path)
if err != nil {
return err
}
b, err := json.MarshalIndent(nd, "", "\t")
if err != nil {
return err
}
cmd := exec.CommandContext(ctx, cctx.Args().Get(3), cctx.Args().Slice()[4:]...)
cmd.Stdin = bytes.NewReader(b)
var out bytes.Buffer
cmd.Stdout = &out
switch cmd.Run().(type) {
case nil:
// it's lower
end = mid
highest = midTs
fmt.Println("true")
case *exec.ExitError:
start = mid
fmt.Println("false")
default:
return err
}
if start == end {
if strings.TrimSpace(out.String()) == "true" {
fmt.Println(midTs.Height())
} else {
fmt.Println(prev)
}
return nil
}
prev = mid
}
},
}
var chainExportCmd = &cli.Command{
Name: "export",
Usage: "export chain to a car file",
@ -462,7 +579,7 @@ var slashConsensusFault = &cli.Command{
return xerrors.Errorf("getting block 1: %w", err)
}
c2, err := cid.Parse(cctx.Args().Get(0))
c2, err := cid.Parse(cctx.Args().Get(1))
if err != nil {
return xerrors.Errorf("parsing cid 2: %w", err)
}

2
cli/params.go
View File

@ -24,7 +24,7 @@ var fetchParamCmd = &cli.Command{
return err
}
sectorSize := uint64(sectorSizeInt)
err = paramfetch.GetParams(build.ParametersJson, sectorSize)
err = paramfetch.GetParams(build.ParametersJson(), sectorSize)
if err != nil {
return xerrors.Errorf("fetching proof parameters: %w", err)
}

8
cmd/lotus-bench/main.go
View File

@ -146,7 +146,7 @@ func main() {
Miner: maddr,
SectorSize: sectorSize,
WorkerThreads: 2,
Dir: sbdir,
Paths: sectorbuilder.SimplePath(sbdir),
}
if robench == "" {
@ -155,7 +155,7 @@ func main() {
}
}
if err := paramfetch.GetParams(build.ParametersJson, sectorSize); err != nil {
if err := paramfetch.GetParams(build.ParametersJson(), sectorSize); err != nil {
return xerrors.Errorf("getting params: %w", err)
}
sb, err := sectorbuilder.New(cfg, mds)
@ -174,7 +174,7 @@ func main() {
r := rand.New(rand.NewSource(100 + int64(i)))
pi, err := sb.AddPiece(dataSize, i, r, nil)
pi, err := sb.AddPiece(context.TODO(), dataSize, i, r, nil)
if err != nil {
return err
}
@ -225,7 +225,7 @@ func main() {
if !c.Bool("skip-unseal") {
log.Info("Unsealing sector")
rc, err := sb.ReadPieceFromSealedSector(1, 0, dataSize, ticket.TicketBytes[:], commD[:])
rc, err := sb.ReadPieceFromSealedSector(context.TODO(), 1, 0, dataSize, ticket.TicketBytes[:], commD[:])
if err != nil {
return err
}

1
cmd/lotus-chainwatch/sync.go
View File

@ -41,7 +41,6 @@ func runSyncer(ctx context.Context, api api.FullNode, st *storage) {
go subMpool(ctx, api, st)
go subBlocks(ctx, api, st)
}
}
}
}()

4
cmd/lotus-seal-worker/sub.go
View File

@ -35,13 +35,13 @@ func acceptJobs(ctx context.Context, api lapi.StorageMiner, endpoint string, aut
SectorSize: ssize,
Miner: act,
WorkerThreads: 1,
Dir: repo,
Paths: sectorbuilder.SimplePath(repo),
})
if err != nil {
return err
}
if err := paramfetch.GetParams(build.ParametersJson, ssize); err != nil {
if err := paramfetch.GetParams(build.ParametersJson(), ssize); err != nil {
return xerrors.Errorf("get params: %w", err)
}

17
cmd/lotus-seal-worker/transfer.go
View File

@ -1,12 +1,14 @@
package main
import (
"fmt"
"io"
"mime"
"net/http"
"os"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
files "github.com/ipfs/go-ipfs-files"
"golang.org/x/xerrors"
"gopkg.in/cheggaaa/pb.v1"
@ -26,7 +28,7 @@ func (w *worker) sizeForType(typ string) int64 {
func (w *worker) fetch(typ string, sectorID uint64) error {
outname := filepath.Join(w.repo, typ, w.sb.SectorName(sectorID))
url := w.minerEndpoint + "/remote/" + typ + "/" + w.sb.SectorName(sectorID)
url := w.minerEndpoint + "/remote/" + typ + "/" + fmt.Sprint(sectorID)
log.Infof("Fetch %s %s", typ, url)
req, err := http.NewRequest("GET", url, nil)
@ -76,21 +78,24 @@ func (w *worker) fetch(typ string, sectorID uint64) error {
}
func (w *worker) push(typ string, sectorID uint64) error {
filename := filepath.Join(w.repo, typ, w.sb.SectorName(sectorID))
filename, err := w.sb.SectorPath(fs.DataType(typ), sectorID)
if err != nil {
return err
}
url := w.minerEndpoint + "/remote/" + typ + "/" + w.sb.SectorName(sectorID)
url := w.minerEndpoint + "/remote/" + typ + "/" + fmt.Sprint(sectorID)
log.Infof("Push %s %s", typ, url)
stat, err := os.Stat(filename)
stat, err := os.Stat(string(filename))
if err != nil {
return err
}
var r io.Reader
if stat.IsDir() {
r, err = tarutil.TarDirectory(filename)
r, err = tarutil.TarDirectory(string(filename))
} else {
r, err = os.OpenFile(filename, os.O_RDONLY, 0644)
r, err = os.OpenFile(string(filename), os.O_RDONLY, 0644)
}
if err != nil {
return xerrors.Errorf("opening push reader: %w", err)

4
cmd/lotus-seed/main.go
View File

@ -196,7 +196,7 @@ var aggregateSectorDirsCmd = &cli.Command{
agsb, err := sectorbuilder.New(&sectorbuilder.Config{
Miner: maddr,
SectorSize: ssize,
Dir: destdir,
Paths: sectorbuilder.SimplePath(destdir),
WorkerThreads: 2,
}, namespace.Wrap(agmds, datastore.NewKey("/sectorbuilder")))
if err != nil {
@ -257,7 +257,7 @@ var aggregateSectorDirsCmd = &cli.Command{
sb, err := sectorbuilder.New(&sectorbuilder.Config{
Miner: maddr,
SectorSize: genm.SectorSize,
Dir: dir,
Paths: sectorbuilder.SimplePath(dir),
WorkerThreads: 2,
}, namespace.Wrap(mds, datastore.NewKey("/sectorbuilder")))
if err != nil {

6
cmd/lotus-seed/seed/seed.go
View File

@ -32,7 +32,7 @@ func PreSeal(maddr address.Address, ssize uint64, offset uint64, sectors int, sb
Miner: maddr,
SectorSize: ssize,
FallbackLastID: offset,
Dir: sbroot,
Paths: sectorbuilder.SimplePath(sbroot),
WorkerThreads: 2,
}
@ -59,7 +59,7 @@ func PreSeal(maddr address.Address, ssize uint64, offset uint64, sectors int, sb
return nil, err
}
pi, err := sb.AddPiece(size, sid, rand.Reader, nil)
pi, err := sb.AddPiece(context.TODO(), size, sid, rand.Reader, nil)
if err != nil {
return nil, err
}
@ -76,7 +76,7 @@ func PreSeal(maddr address.Address, ssize uint64, offset uint64, sectors int, sb
return nil, xerrors.Errorf("commit: %w", err)
}
if err := sb.TrimCache(sid); err != nil {
if err := sb.TrimCache(context.TODO(), sid); err != nil {
return nil, xerrors.Errorf("trim cache: %w", err)
}

47
cmd/lotus-shed/bigint.go Normal file
View File

@ -0,0 +1,47 @@
package main
import (
"encoding/base64"
"encoding/hex"
"fmt"
"github.com/filecoin-project/lotus/chain/types"
"gopkg.in/urfave/cli.v2"
)
var bigIntParseCmd = &cli.Command{
Name: "bigint",
Description: "parse encoded big ints",
Flags: []cli.Flag{
&cli.StringFlag{
Name: "enc",
Value: "base64",
Usage: "specify input encoding to parse",
},
},
Action: func(cctx *cli.Context) error {
val := cctx.Args().Get(0)
var dec []byte
switch cctx.String("enc") {
case "base64":
d, err := base64.StdEncoding.DecodeString(val)
if err != nil {
return fmt.Errorf("decoding base64 value: %w", err)
}
dec = d
case "hex":
d, err := hex.DecodeString(val)
if err != nil {
return fmt.Errorf("decoding hex value: %w", err)
}
dec = d
default:
return fmt.Errorf("unrecognized encoding: %s", cctx.String("enc"))
}
iv := types.BigFromBytes(dec)
fmt.Println(iv.String())
return nil
},
}

1
cmd/lotus-shed/main.go
View File

@ -20,6 +20,7 @@ func main() {
keyinfoCmd,
peerkeyCmd,
noncefix,
bigIntParseCmd,
}
app := &cli.App{

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

@ -50,15 +50,27 @@ var infoCmd = &cli.Command{
return err
}
percI := types.BigDiv(types.BigMul(pow.MinerPower, types.NewInt(1000)), pow.TotalPower)
fmt.Printf("Power: %s / %s (%0.4f%%)\n", pow.MinerPower.SizeStr(), pow.TotalPower.SizeStr(), float64(percI.Int64())/100000*10000)
percI := types.BigDiv(types.BigMul(pow.MinerPower, types.NewInt(1000000)), pow.TotalPower)
fmt.Printf("Power: %s / %s (%0.4f%%)\n", pow.MinerPower.SizeStr(), pow.TotalPower.SizeStr(), float64(percI.Int64())/10000)
secCounts, err := api.StateMinerSectorCount(ctx, maddr, nil)
if err != nil {
return err
}
faults, err := api.StateMinerFaults(ctx, maddr, nil)
if err != nil {
return err
}
fmt.Printf("\tCommitted: %s\n", types.BigMul(types.NewInt(secCounts.Sset), types.NewInt(sizeByte)).SizeStr())
fmt.Printf("\tProving: %s\n", types.BigMul(types.NewInt(secCounts.Pset), types.NewInt(sizeByte)).SizeStr())
if len(faults) == 0 {
fmt.Printf("\tProving: %s\n", types.BigMul(types.NewInt(secCounts.Pset), types.NewInt(sizeByte)).SizeStr())
} else {
fmt.Printf("\tProving: %s (%s Faulty, %.2f%%)\n",
types.BigMul(types.NewInt(secCounts.Pset-uint64(len(faults))), types.NewInt(sizeByte)).SizeStr(),
types.BigMul(types.NewInt(uint64(len(faults))), types.NewInt(sizeByte)).SizeStr(),
float64(10000*uint64(len(faults))/secCounts.Pset)/100.)
}
// TODO: indicate whether the post worker is in use
wstat, err := nodeApi.WorkerStats(ctx)

8
cmd/lotus-storage-miner/init.go
View File

@ -97,7 +97,7 @@ var initCmd = &cli.Command{
}
log.Info("Checking proof parameters")
if err := paramfetch.GetParams(build.ParametersJson, ssize); err != nil {
if err := paramfetch.GetParams(build.ParametersJson(), ssize); err != nil {
return xerrors.Errorf("fetching proof parameters: %w", err)
}
@ -177,7 +177,7 @@ var initCmd = &cli.Command{
oldsb, err := sectorbuilder.New(&sectorbuilder.Config{
SectorSize: ssize,
WorkerThreads: 2,
Dir: pssb,
Paths: sectorbuilder.SimplePath(pssb),
}, namespace.Wrap(oldmds, datastore.NewKey("/sectorbuilder")))
if err != nil {
return xerrors.Errorf("failed to open up preseal sectorbuilder: %w", err)
@ -186,7 +186,7 @@ var initCmd = &cli.Command{
nsb, err := sectorbuilder.New(&sectorbuilder.Config{
SectorSize: ssize,
WorkerThreads: 2,
Dir: lr.Path(),
Paths: sectorbuilder.SimplePath(lr.Path()),
}, namespace.Wrap(mds, datastore.NewKey("/sectorbuilder")))
if err != nil {
return xerrors.Errorf("failed to open up sectorbuilder: %w", err)
@ -369,7 +369,7 @@ func storageMinerInit(ctx context.Context, cctx *cli.Context, api lapi.FullNode,
return err
}
sbcfg, err := modules.SectorBuilderConfig(lr.Path(), 2, false, false)(mds, api)
sbcfg, err := modules.SectorBuilderConfig(sectorbuilder.SimplePath(lr.Path()), 2, false, false)(mds, api)
if err != nil {
return xerrors.Errorf("getting genesis miner sector builder config: %w", err)
}

2
cmd/lotus/daemon.go
View File

@ -100,7 +100,7 @@ var DaemonCmd = &cli.Command{
return xerrors.Errorf("repo init error: %w", err)
}
if err := paramfetch.GetParams(build.ParametersJson, 0); err != nil {
if err := paramfetch.GetParams(build.ParametersJson(), 0); err != nil {
return xerrors.Errorf("fetching proof parameters: %w", err)
}

112
cmd/lotus/debug_advance.go Normal file
View File

@ -0,0 +1,112 @@
// +build debug
package main
import (
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/actors"
"github.com/filecoin-project/lotus/chain/gen"
"github.com/filecoin-project/lotus/chain/types"
lcli "github.com/filecoin-project/lotus/cli"
"github.com/filecoin-project/lotus/miner"
"golang.org/x/xerrors"
"gopkg.in/urfave/cli.v2"
)
func init() {
AdvanceBlockCmd = &cli.Command{
Name: "advance-block",
Action: func(cctx *cli.Context) error {
api, closer, err := lcli.GetFullNodeAPI(cctx)
if err != nil {
return err
}
defer closer()
ctx := lcli.ReqContext(cctx)
head, err := api.ChainHead(ctx)
if err != nil {
return err
}
pending, err := api.MpoolPending(ctx, head)
if err != nil {
return err
}
msgs, err := miner.SelectMessages(ctx, api.StateGetActor, head, pending)
if len(msgs) > build.BlockMessageLimit {
log.Error("SelectMessages returned too many messages: ", len(msgs))
msgs = msgs[:build.BlockMessageLimit]
}
addr, _ := address.NewIDAddress(101)
var ticket *types.Ticket
{
vrfBase := head.MinTicket().VRFProof
ret, err := api.StateCall(ctx, &types.Message{
From: addr,
To: addr,
Method: actors.MAMethods.GetWorkerAddr,
}, head)
if err != nil {
return xerrors.Errorf("failed to get miner worker addr: %w", err)
}
if ret.ExitCode != 0 {
return xerrors.Errorf("failed to get miner worker addr (exit code %d)", ret.ExitCode)
}
w, err := address.NewFromBytes(ret.Return)
if err != nil {
return xerrors.Errorf("GetWorkerAddr returned malformed address: %w", err)
}
t, err := gen.ComputeVRF(ctx, api.WalletSign, w, addr, gen.DSepTicket, vrfBase)
if err != nil {
return xerrors.Errorf("compute vrf failed: %w", err)
}
ticket = &types.Ticket{
VRFProof: t,
}
}
epostp := &types.EPostProof{
Proof: []byte("valid proof"),
Candidates: []types.EPostTicket{
{
ChallengeIndex: 0,
SectorID: 1,
},
},
}
{
r, err := api.ChainGetRandomness(ctx, head.Key(), int64(head.Height()+1)-build.EcRandomnessLookback)
if err != nil {
return xerrors.Errorf("chain get randomness: %w", err)
}
mworker, err := api.StateMinerWorker(ctx, addr, head)
if err != nil {
return xerrors.Errorf("failed to get miner worker: %w", err)
}
vrfout, err := gen.ComputeVRF(ctx, api.WalletSign, mworker, addr, gen.DSepElectionPost, r)
if err != nil {
return xerrors.Errorf("failed to compute VRF: %w", err)
}
epostp.PostRand = vrfout
}
uts := head.MinTimestamp() + uint64(build.BlockDelay)
nheight := head.Height() + 1
blk, err := api.MinerCreateBlock(ctx, addr, head, ticket, epostp, msgs, nheight, uts)
if err != nil {
return xerrors.Errorf("creating block: %w", err)
}
return api.SyncSubmitBlock(ctx, blk)
},
}
}

6
cmd/lotus/main.go
View File

@ -14,12 +14,18 @@ import (
"github.com/filecoin-project/lotus/tracing"
)
var AdvanceBlockCmd *cli.Command
func main() {
lotuslog.SetupLogLevels()
local := []*cli.Command{
DaemonCmd,
}
if AdvanceBlockCmd != nil {
local = append(local, AdvanceBlockCmd)
}
jaeger := tracing.SetupJaegerTracing("lotus")
defer func() {
if jaeger != nil {

6
documentation/en/.library.json
View File

@ -32,6 +32,12 @@
"github": "en/install-lotus-ubuntu.md",
"value": null
},
{
"title": "Fedora Installation",
"slug": "en+install-lotus-fedora",
"github": "en/install-lotus-fedora.md",
"value": null
},
{
"title": "MacOS Installation",
"slug": "en+install-lotus-macos",

6
documentation/en/storing-data.md
View File

@ -49,6 +49,12 @@ Store a **Data CID** with a miner:
lotus client deal <Data CID> <miner> <price> <duration>
```
Check the status of a deal:
```sh
lotus client list-deals
```
- Price is in attoFIL.
- The `duration`, which represents how long the miner will keep your file hosted, is represented in blocks. Each block represents 45 seconds.

2
go.mod
View File

@ -20,7 +20,7 @@ require (
github.com/filecoin-project/go-data-transfer v0.0.0-20191219005021-4accf56bd2ce
github.com/filecoin-project/go-fil-markets v0.0.0-20200124235616-d94a1cf0beaa
github.com/filecoin-project/go-paramfetch v0.0.1
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200123143044-d9cc96c53c55
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200203173614-42d67726bb62
github.com/filecoin-project/go-statestore v0.1.0
github.com/gbrlsnchs/jwt/v3 v3.0.0-beta.1
github.com/go-ole/go-ole v1.2.4 // indirect

4
go.sum
View File

@ -117,8 +117,8 @@ github.com/filecoin-project/go-paramfetch v0.0.0-20200102181131-b20d579f2878/go.
github.com/filecoin-project/go-paramfetch v0.0.1 h1:gV7bs5YaqlgpGFMiLxInGK2L1FyCXUE0rimz4L7ghoE=
github.com/filecoin-project/go-paramfetch v0.0.1/go.mod h1:fZzmf4tftbwf9S37XRifoJlz7nCjRdIrMGLR07dKLCc=
github.com/filecoin-project/go-sectorbuilder v0.0.1/go.mod h1:3OZ4E3B2OuwhJjtxR4r7hPU9bCfB+A+hm4alLEsaeDc=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200123143044-d9cc96c53c55 h1:XChPRKPZL+/N6a3ccLmjCJ7JrR+SFLFJDllv0BkxW4I=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200123143044-d9cc96c53c55/go.mod h1:ahsryULdwYoZ94K09HcfqX3QBwevWVldENSV/EdCbNg=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200203173614-42d67726bb62 h1:/+xdjMkIdiRs6vA2lJU56iqtEcl9BQgYXi8b2KuuYCg=
github.com/filecoin-project/go-sectorbuilder v0.0.2-0.20200203173614-42d67726bb62/go.mod h1:jNGVCDihkMFnraYVLH1xl4ceZQVxx/u4dOORrTKeRi0=
github.com/filecoin-project/go-statestore v0.1.0 h1:t56reH59843TwXHkMcwyuayStBIiWBRilQjQ+5IiwdQ=
github.com/filecoin-project/go-statestore v0.1.0/go.mod h1:LFc9hD+fRxPqiHiaqUEZOinUJB4WARkRfNl10O7kTnI=
github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=

2
markets/retrievaladapter/provider.go
View File

@ -31,7 +31,7 @@ func (rpn *retrievalProviderNode) UnsealSector(ctx context.Context, sectorID uin
if err != nil {
return nil, err
}
return rpn.sb.ReadPieceFromSealedSector(sectorID, offset, length, si.Ticket.TicketBytes, si.CommD)
return rpn.sb.ReadPieceFromSealedSector(ctx, sectorID, offset, length, si.Ticket.TicketBytes, si.CommD)
}
func (rpn *retrievalProviderNode) SavePaymentVoucher(ctx context.Context, paymentChannel address.Address, voucher *retrievaltypes.SignedVoucher, proof []byte, expectedAmount retrievaltoken.TokenAmount) (retrievaltoken.TokenAmount, error) {

16
node/builder.go
View File

@ -7,6 +7,7 @@ import (
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
blockstore "github.com/ipfs/go-ipfs-blockstore"
logging "github.com/ipfs/go-log"
ci "github.com/libp2p/go-libp2p-core/crypto"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/peer"
@ -54,6 +55,8 @@ import (
"github.com/filecoin-project/lotus/storage/sectorblocks"
)
var log = logging.Logger("builder")
// special is a type used to give keys to modules which
// can't really be identified by the returned type
type special struct{ id int }
@ -343,15 +346,20 @@ func ConfigStorageMiner(c interface{}, lr repo.LockedRepo) Option {
return Error(xerrors.Errorf("invalid config from repo, got: %T", c))
}
path := cfg.SectorBuilder.Path
if path == "" {
path = lr.Path()
scfg := sectorbuilder.SimplePath(lr.Path())
if cfg.SectorBuilder.Path == "" {
if len(cfg.SectorBuilder.Storage) > 0 {
scfg = cfg.SectorBuilder.Storage
}
} else {
scfg = sectorbuilder.SimplePath(cfg.SectorBuilder.Path)
log.Warn("LEGACY SectorBuilder.Path FOUND IN CONFIG. Please use the new storage config")
}
return Options(
ConfigCommon(&cfg.Common),
Override(new(*sectorbuilder.Config), modules.SectorBuilderConfig(path,
Override(new(*sectorbuilder.Config), modules.SectorBuilderConfig(scfg,
cfg.SectorBuilder.WorkerCount,
cfg.SectorBuilder.DisableLocalPreCommit,
cfg.SectorBuilder.DisableLocalCommit)),

5
node/config/def.go
View File

@ -3,6 +3,8 @@ package config
import (
"encoding"
"time"
"github.com/filecoin-project/go-sectorbuilder/fs"
)
// Common is common config between full node and miner
@ -54,7 +56,8 @@ type Metrics struct {
// // Storage Miner
type SectorBuilder struct {
Path string
Path string // TODO: remove // FORK (-ish)
Storage []fs.PathConfig
WorkerCount uint
DisableLocalPreCommit bool

4
node/impl/full/chain.go
View File

@ -170,6 +170,10 @@ func (a *ChainAPI) ChainReadObj(ctx context.Context, obj cid.Cid) ([]byte, error
return blk.RawData(), nil
}
func (a *ChainAPI) ChainHasObj(ctx context.Context, obj cid.Cid) (bool, error) {
return a.Chain.Blockstore().Has(obj)
}
func (a *ChainAPI) ChainSetHead(ctx context.Context, ts *types.TipSet) error {
return a.Chain.SetHead(ts)
}

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

@ -3,6 +3,7 @@ package full
import (
"bytes"
"context"
"fmt"
"strconv"
"github.com/filecoin-project/go-amt-ipld"
@ -84,6 +85,10 @@ func (a *StateAPI) StateMinerSectorSize(ctx context.Context, actor address.Addre
return stmgr.GetMinerSectorSize(ctx, a.StateManager, ts, actor)
}
func (a *StateAPI) StateMinerFaults(ctx context.Context, addr address.Address, ts *types.TipSet) ([]uint64, error) {
return stmgr.GetMinerFaults(ctx, a.StateManager, ts, addr)
}
func (a *StateAPI) StatePledgeCollateral(ctx context.Context, ts *types.TipSet) (types.BigInt, error) {
param, err := actors.SerializeParams(&actors.PledgeCollateralParams{Size: types.NewInt(0)})
if err != nil {
@ -404,3 +409,32 @@ func (a *StateAPI) StateListMessages(ctx context.Context, match *types.Message,
func (a *StateAPI) StateCompute(ctx context.Context, height uint64, msgs []*types.Message, ts *types.TipSet) (cid.Cid, error) {
return stmgr.ComputeState(ctx, a.StateManager, height, msgs, ts)
}
func (a *StateAPI) MsigGetAvailableBalance(ctx context.Context, addr address.Address, ts *types.TipSet) (types.BigInt, error) {
if ts == nil {
ts = a.Chain.GetHeaviestTipSet()
}
var st actors.MultiSigActorState
act, err := a.StateManager.LoadActorState(ctx, addr, &st, ts)
if err != nil {
return types.EmptyInt, xerrors.Errorf("failed to load multisig actor state: %w", err)
}
if act.Code != actors.MultisigCodeCid {
return types.EmptyInt, fmt.Errorf("given actor was not a multisig")
}
if st.UnlockDuration == 0 {
return act.Balance, nil
}
offset := ts.Height() - st.StartingBlock
if offset > st.UnlockDuration {
return act.Balance, nil
}
minBalance := types.BigDiv(st.InitialBalance, types.NewInt(st.UnlockDuration))
minBalance = types.BigMul(minBalance, types.NewInt(offset))
return types.BigSub(act.Balance, minBalance), nil
}

40
node/impl/storminer.go
View File

@ -9,14 +9,15 @@ import (
"os"
"strconv"
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/gorilla/mux"
files "github.com/ipfs/go-ipfs-files"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/api/apistruct"
"github.com/filecoin-project/lotus/lib/tarutil"
"github.com/filecoin-project/lotus/miner"
"github.com/filecoin-project/lotus/storage"
@ -44,8 +45,8 @@ func (sm *StorageMinerAPI) ServeRemote(w http.ResponseWriter, r *http.Request) {
mux := mux.NewRouter()
mux.HandleFunc("/remote/{type}/{sname}", sm.remoteGetSector).Methods("GET")
mux.HandleFunc("/remote/{type}/{sname}", sm.remotePutSector).Methods("PUT")
mux.HandleFunc("/remote/{type}/{id}", sm.remoteGetSector).Methods("GET")
mux.HandleFunc("/remote/{type}/{id}", sm.remotePutSector).Methods("PUT")
log.Infof("SERVEGETREMOTE %s", r.URL)
@ -55,14 +56,21 @@ func (sm *StorageMinerAPI) ServeRemote(w http.ResponseWriter, r *http.Request) {
func (sm *StorageMinerAPI) remoteGetSector(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
path, err := sm.SectorBuilder.GetPath(vars["type"], vars["sname"])
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
log.Error("parsing sector id: ", err)
w.WriteHeader(500)
return
}
path, err := sm.SectorBuilder.SectorPath(fs.DataType(vars["type"]), id)
if err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
stat, err := os.Stat(path)
stat, err := os.Stat(string(path))
if err != nil {
log.Error(err)
w.WriteHeader(500)
@ -71,10 +79,10 @@ func (sm *StorageMinerAPI) remoteGetSector(w http.ResponseWriter, r *http.Reques
var rd io.Reader
if stat.IsDir() {
rd, err = tarutil.TarDirectory(path)
rd, err = tarutil.TarDirectory(string(path))
w.Header().Set("Content-Type", "application/x-tar")
} else {
rd, err = os.OpenFile(path, os.O_RDONLY, 0644)
rd, err = os.OpenFile(string(path), os.O_RDONLY, 0644)
w.Header().Set("Content-Type", "application/octet-stream")
}
if err != nil {
@ -93,7 +101,15 @@ func (sm *StorageMinerAPI) remoteGetSector(w http.ResponseWriter, r *http.Reques
func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Request) {
vars := mux.Vars(r)
path, err := sm.SectorBuilder.GetPath(vars["type"], vars["sname"])
id, err := strconv.ParseUint(vars["id"], 10, 64)
if err != nil {
log.Error("parsing sector id: ", err)
w.WriteHeader(500)
return
}
// This is going to get better with worker-to-worker transfers
path, err := sm.SectorBuilder.AllocSectorPath(fs.DataType(vars["type"]), id, true)
if err != nil {
log.Error(err)
w.WriteHeader(500)
@ -107,7 +123,7 @@ func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Reques
return
}
if err := os.RemoveAll(path); err != nil {
if err := os.RemoveAll(string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return
@ -115,13 +131,13 @@ func (sm *StorageMinerAPI) remotePutSector(w http.ResponseWriter, r *http.Reques
switch mediatype {
case "application/x-tar":
if err := tarutil.ExtractTar(r.Body, path); err != nil {
if err := tarutil.ExtractTar(r.Body, string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return
}
default:
if err := files.WriteTo(files.NewReaderFile(r.Body), path); err != nil {
if err := files.WriteTo(files.NewReaderFile(r.Body), string(path)); err != nil {
log.Error(err)
w.WriteHeader(500)
return

27
node/modules/storageminer.go
View File

@ -17,6 +17,7 @@ import (
storageimpl "github.com/filecoin-project/go-fil-markets/storagemarket/impl"
paramfetch "github.com/filecoin-project/go-paramfetch"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
"github.com/filecoin-project/go-statestore"
"github.com/ipfs/go-bitswap"
"github.com/ipfs/go-bitswap/network"
@ -57,14 +58,14 @@ func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
}
func GetParams(sbc *sectorbuilder.Config) error {
if err := paramfetch.GetParams(build.ParametersJson, sbc.SectorSize); err != nil {
if err := paramfetch.GetParams(build.ParametersJson(), sbc.SectorSize); err != nil {
return xerrors.Errorf("fetching proof parameters: %w", err)
}
return nil
}
func SectorBuilderConfig(storagePath string, threads uint, noprecommit, nocommit bool) func(dtypes.MetadataDS, api.FullNode) (*sectorbuilder.Config, error) {
func SectorBuilderConfig(storage []fs.PathConfig, threads uint, noprecommit, nocommit bool) func(dtypes.MetadataDS, api.FullNode) (*sectorbuilder.Config, error) {
return func(ds dtypes.MetadataDS, api api.FullNode) (*sectorbuilder.Config, error) {
minerAddr, err := minerAddrFromDS(ds)
if err != nil {
@ -76,9 +77,11 @@ func SectorBuilderConfig(storagePath string, threads uint, noprecommit, nocommit
return nil, err
}
sp, err := homedir.Expand(storagePath)
if err != nil {
return nil, err
for i := range storage {
storage[i].Path, err = homedir.Expand(storage[i].Path)
if err != nil {
return nil, err
}
}
if threads > math.MaxUint8 {
@ -93,7 +96,7 @@ func SectorBuilderConfig(storagePath string, threads uint, noprecommit, nocommit
NoPreCommit: noprecommit,
NoCommit: nocommit,
Dir: sp,
Paths: storage,
}
return sb, nil
@ -106,15 +109,23 @@ func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h
return nil, err
}
sm, err := storage.NewMiner(api, maddr, h, ds, sb, tktFn)
ctx := helpers.LifecycleCtx(mctx, lc)
worker, err := api.StateMinerWorker(ctx, maddr, nil)
if err != nil {
return nil, err
}
ctx := helpers.LifecycleCtx(mctx, lc)
fps := storage.NewFPoStScheduler(api, sb, maddr, worker)
sm, err := storage.NewMiner(api, maddr, worker, h, ds, sb, tktFn)
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
go fps.Run(ctx)
return sm.Run(ctx)
},
OnStop: sm.Stop,

2
node/node_test.go
View File

@ -232,7 +232,7 @@ func builder(t *testing.T, nFull int, storage []int) ([]test.TestNode, []test.Te
SectorSize: 1024,
WorkerThreads: 2,
Miner: genMiner,
Dir: psd,
Paths: sectorbuilder.SimplePath(psd),
}, namespace.Wrap(mds, datastore.NewKey("/sectorbuilder")))
if err != nil {
t.Fatal(err)

125
storage/fpost_run.go
View File

@ -14,7 +14,7 @@ import (
"github.com/filecoin-project/lotus/chain/types"
)
func (s *fpostScheduler) failPost(eps uint64) {
func (s *FPoStScheduler) failPost(eps uint64) {
s.failLk.Lock()
if eps > s.failed {
s.failed = eps
@ -22,7 +22,7 @@ func (s *fpostScheduler) failPost(eps uint64) {
s.failLk.Unlock()
}
func (s *fpostScheduler) doPost(ctx context.Context, eps uint64, ts *types.TipSet) {
func (s *FPoStScheduler) doPost(ctx context.Context, eps uint64, ts *types.TipSet) {
ctx, abort := context.WithCancel(ctx)
s.abort = abort
@ -31,7 +31,7 @@ func (s *fpostScheduler) doPost(ctx context.Context, eps uint64, ts *types.TipSe
go func() {
defer abort()
ctx, span := trace.StartSpan(ctx, "fpostScheduler.doPost")
ctx, span := trace.StartSpan(ctx, "FPoStScheduler.doPost")
defer span.End()
proof, err := s.runPost(ctx, eps, ts)
@ -50,58 +50,91 @@ func (s *fpostScheduler) doPost(ctx context.Context, eps uint64, ts *types.TipSe
}()
}
func (s *fpostScheduler) checkFaults(ctx context.Context, ssi sectorbuilder.SortedPublicSectorInfo) ([]uint64, error) {
func (s *FPoStScheduler) declareFaults(ctx context.Context, fc uint64, params *actors.DeclareFaultsParams) error {
log.Warnf("DECLARING %d FAULTS", fc)
enc, aerr := actors.SerializeParams(params)
if aerr != nil {
return xerrors.Errorf("could not serialize declare faults parameters: %w", aerr)
}
msg := &types.Message{
To: s.actor,
From: s.worker,
Method: actors.MAMethods.DeclareFaults,
Params: enc,
Value: types.NewInt(0),
GasLimit: types.NewInt(10000000), // i dont know help
GasPrice: types.NewInt(1),
}
sm, err := s.api.MpoolPushMessage(ctx, msg)
if err != nil {
return xerrors.Errorf("pushing faults message to mpool: %w", err)
}
rec, err := s.api.StateWaitMsg(ctx, sm.Cid())
if err != nil {
return xerrors.Errorf("waiting for declare faults: %w", err)
}
if rec.Receipt.ExitCode != 0 {
return xerrors.Errorf("declare faults exit %d", rec.Receipt.ExitCode)
}
log.Infof("Faults declared successfully")
return nil
}
func (s *FPoStScheduler) checkFaults(ctx context.Context, ssi sectorbuilder.SortedPublicSectorInfo) ([]uint64, error) {
faults := s.sb.Scrub(ssi)
var faultIDs []uint64
declaredFaults := map[uint64]struct{}{}
{
chainFaults, err := s.api.StateMinerFaults(ctx, s.actor, nil)
if err != nil {
return nil, xerrors.Errorf("checking on-chain faults: %w", err)
}
for _, fault := range chainFaults {
declaredFaults[fault] = struct{}{}
}
}
if len(faults) > 0 {
params := &actors.DeclareFaultsParams{Faults: types.NewBitField()}
for _, fault := range faults {
log.Warnf("fault detected: sector %d: %s", fault.SectorID, fault.Err)
faultIDs = append(faultIDs, fault.SectorID)
if _, ok := declaredFaults[fault.SectorID]; ok {
continue
}
// TODO: omit already declared (with finality in mind though..)
log.Warnf("new fault detected: sector %d: %s", fault.SectorID, fault.Err)
declaredFaults[fault.SectorID] = struct{}{}
params.Faults.Set(fault.SectorID)
}
log.Warnf("DECLARING %d FAULTS", len(faults))
enc, aerr := actors.SerializeParams(params)
if aerr != nil {
return nil, xerrors.Errorf("could not serialize declare faults parameters: %w", aerr)
}
msg := &types.Message{
To: s.actor,
From: s.worker,
Method: actors.MAMethods.DeclareFaults,
Params: enc,
Value: types.NewInt(0),
GasLimit: types.NewInt(10000000), // i dont know help
GasPrice: types.NewInt(1),
}
sm, err := s.api.MpoolPushMessage(ctx, msg)
pc, err := params.Faults.Count()
if err != nil {
return nil, xerrors.Errorf("pushing faults message to mpool: %w", err)
return nil, xerrors.Errorf("counting faults: %w", err)
}
if pc > 0 {
if err := s.declareFaults(ctx, pc, params); err != nil {
return nil, err
}
}
}
rec, err := s.api.StateWaitMsg(ctx, sm.Cid())
if err != nil {
return nil, xerrors.Errorf("waiting for declare faults: %w", err)
}
if rec.Receipt.ExitCode != 0 {
return nil, xerrors.Errorf("declare faults exit %d", rec.Receipt.ExitCode)
}
log.Infof("Faults declared successfully")
faultIDs := make([]uint64, 0, len(declaredFaults))
for fault := range declaredFaults {
faultIDs = append(faultIDs, fault)
}
return faultIDs, nil
}
func (s *fpostScheduler) runPost(ctx context.Context, eps uint64, ts *types.TipSet) (*actors.SubmitFallbackPoStParams, error) {
func (s *FPoStScheduler) runPost(ctx context.Context, eps uint64, ts *types.TipSet) (*actors.SubmitFallbackPoStParams, error) {
ctx, span := trace.StartSpan(ctx, "storage.runPost")
defer span.End()
@ -161,7 +194,7 @@ func (s *fpostScheduler) runPost(ctx context.Context, eps uint64, ts *types.TipS
}, nil
}
func (s *fpostScheduler) sortedSectorInfo(ctx context.Context, ts *types.TipSet) (sectorbuilder.SortedPublicSectorInfo, error) {
func (s *FPoStScheduler) sortedSectorInfo(ctx context.Context, ts *types.TipSet) (sectorbuilder.SortedPublicSectorInfo, error) {
sset, err := s.api.StateMinerProvingSet(ctx, s.actor, ts)
if err != nil {
return sectorbuilder.SortedPublicSectorInfo{}, xerrors.Errorf("failed to get proving set for miner (tsH: %d): %w", ts.Height(), err)
@ -184,7 +217,7 @@ func (s *fpostScheduler) sortedSectorInfo(ctx context.Context, ts *types.TipSet)
return sectorbuilder.NewSortedPublicSectorInfo(sbsi), nil
}
func (s *fpostScheduler) submitPost(ctx context.Context, proof *actors.SubmitFallbackPoStParams) error {
func (s *FPoStScheduler) submitPost(ctx context.Context, proof *actors.SubmitFallbackPoStParams) error {
ctx, span := trace.StartSpan(ctx, "storage.commitPost")
defer span.End()
@ -211,5 +244,19 @@ func (s *fpostScheduler) submitPost(ctx context.Context, proof *actors.SubmitFal
log.Infof("Submitted fallback post: %s", sm.Cid())
go func() {
rec, err := s.api.StateWaitMsg(context.TODO(), sm.Cid())
if err != nil {
log.Error(err)
return
}
if rec.Receipt.ExitCode == 0 {
return
}
log.Errorf("Submitting fallback post %s failed: exit %d", sm.Cid(), rec.Receipt.ExitCode)
}()
return nil
}

22
storage/fpost_sched.go
View File

@ -19,7 +19,7 @@ const Inactive = 0
const StartConfidence = 4 // TODO: config
type fpostScheduler struct {
type FPoStScheduler struct {
api storageMinerApi
sb sectorbuilder.Interface
@ -36,7 +36,11 @@ type fpostScheduler struct {
failLk sync.Mutex
}
func (s *fpostScheduler) run(ctx context.Context) {
func NewFPoStScheduler(api storageMinerApi, sb sectorbuilder.Interface, actor address.Address, worker address.Address) *FPoStScheduler {
return &FPoStScheduler{api: api, sb: sb, actor: actor, worker: worker}
}
func (s *FPoStScheduler) Run(ctx context.Context) {
notifs, err := s.api.ChainNotify(ctx)
if err != nil {
return
@ -61,11 +65,11 @@ func (s *fpostScheduler) run(ctx context.Context) {
select {
case changes, ok := <-notifs:
if !ok {
log.Warn("fpostScheduler notifs channel closed")
log.Warn("FPoStScheduler notifs channel closed")
return
}
ctx, span := trace.StartSpan(ctx, "fpostScheduler.headChange")
ctx, span := trace.StartSpan(ctx, "FPoStScheduler.headChange")
var lowest, highest *types.TipSet = s.cur, nil
@ -95,7 +99,7 @@ func (s *fpostScheduler) run(ctx context.Context) {
}
}
func (s *fpostScheduler) revert(ctx context.Context, newLowest *types.TipSet) error {
func (s *FPoStScheduler) revert(ctx context.Context, newLowest *types.TipSet) error {
if s.cur == newLowest {
return nil
}
@ -113,9 +117,9 @@ func (s *fpostScheduler) revert(ctx context.Context, newLowest *types.TipSet) er
return nil
}
func (s *fpostScheduler) update(ctx context.Context, new *types.TipSet) error {
func (s *FPoStScheduler) update(ctx context.Context, new *types.TipSet) error {
if new == nil {
return xerrors.Errorf("no new tipset in fpostScheduler.update")
return xerrors.Errorf("no new tipset in FPoStScheduler.update")
}
newEPS, start, err := s.shouldFallbackPost(ctx, new)
if err != nil {
@ -142,7 +146,7 @@ func (s *fpostScheduler) update(ctx context.Context, new *types.TipSet) error {
return nil
}
func (s *fpostScheduler) abortActivePoSt() {
func (s *FPoStScheduler) abortActivePoSt() {
if s.activeEPS == Inactive {
return // noop
}
@ -157,7 +161,7 @@ func (s *fpostScheduler) abortActivePoSt() {
s.abort = nil
}
func (s *fpostScheduler) shouldFallbackPost(ctx context.Context, ts *types.TipSet) (uint64, bool, error) {
func (s *FPoStScheduler) shouldFallbackPost(ctx context.Context, ts *types.TipSet) (uint64, bool, error) {
eps, err := s.api.StateMinerElectionPeriodStart(ctx, s.actor, ts)
if err != nil {
return 0, false, xerrors.Errorf("getting ElectionPeriodStart: %w", err)

25
storage/miner.go
View File

@ -50,6 +50,7 @@ type storageMinerApi interface {
StateGetActor(ctx context.Context, actor address.Address, ts *types.TipSet) (*types.Actor, error)
StateGetReceipt(context.Context, cid.Cid, *types.TipSet) (*types.MessageReceipt, error)
StateMarketStorageDeal(context.Context, uint64, *types.TipSet) (*actors.OnChainDeal, error)
StateMinerFaults(context.Context, address.Address, *types.TipSet) ([]uint64, error)
MpoolPushMessage(context.Context, *types.Message) (*types.SignedMessage, error)
@ -65,7 +66,7 @@ type storageMinerApi interface {
WalletHas(context.Context, address.Address) (bool, error)
}
func NewMiner(api storageMinerApi, addr address.Address, h host.Host, ds datastore.Batching, sb sectorbuilder.Interface, tktFn sealing.TicketFn) (*Miner, error) {
func NewMiner(api storageMinerApi, maddr, worker address.Address, h host.Host, ds datastore.Batching, sb sectorbuilder.Interface, tktFn sealing.TicketFn) (*Miner, error) {
m := &Miner{
api: api,
h: h,
@ -73,7 +74,8 @@ func NewMiner(api storageMinerApi, addr address.Address, h host.Host, ds datasto
ds: ds,
tktFn: tktFn,
maddr: addr,
maddr: maddr,
worker: worker,
}
return m, nil
@ -84,16 +86,6 @@ func (m *Miner) Run(ctx context.Context) error {
return xerrors.Errorf("miner preflight checks failed: %w", err)
}
fps := &fpostScheduler{
api: m.api,
sb: m.sb,
actor: m.maddr,
worker: m.worker,
}
go fps.run(ctx)
evts := events.NewEvents(ctx, m.api)
m.sealing = sealing.New(m.api, evts, m.maddr, m.worker, m.ds, m.sb, m.tktFn)
@ -108,14 +100,7 @@ func (m *Miner) Stop(ctx context.Context) error {
}
func (m *Miner) runPreflightChecks(ctx context.Context) error {
worker, err := m.api.StateMinerWorker(ctx, m.maddr, nil)
if err != nil {
return err
}
m.worker = worker
has, err := m.api.WalletHas(ctx, worker)
has, err := m.api.WalletHas(ctx, m.worker)
if err != nil {
return xerrors.Errorf("failed to check wallet for worker key: %w", err)
}

27
storage/sbmock/sbmock.go
View File

@ -13,6 +13,7 @@ import (
ffi "github.com/filecoin-project/filecoin-ffi"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
"golang.org/x/xerrors"
)
@ -60,7 +61,7 @@ func (sb *SBMock) RateLimit() func() {
}
}
func (sb *SBMock) AddPiece(size uint64, sectorId uint64, r io.Reader, existingPieces []uint64) (sectorbuilder.PublicPieceInfo, error) {
func (sb *SBMock) AddPiece(ctx context.Context, size uint64, sectorId uint64, r io.Reader, existingPieces []uint64) (sectorbuilder.PublicPieceInfo, error) {
sb.lk.Lock()
ss, ok := sb.sectors[sectorId]
if !ok {
@ -284,7 +285,7 @@ func (sb *SBMock) GenerateEPostCandidates(sectorInfo sectorbuilder.SortedPublicS
return out, nil
}
func (sb *SBMock) ReadPieceFromSealedSector(sectorID uint64, offset uint64, size uint64, ticket []byte, commD []byte) (io.ReadCloser, error) {
func (sb *SBMock) ReadPieceFromSealedSector(ctx context.Context, sectorID uint64, offset uint64, size uint64, ticket []byte, commD []byte) (io.ReadCloser, error) {
if len(sb.sectors[sectorID].pieces) > 1 {
panic("implme")
}
@ -301,7 +302,7 @@ func (sb *SBMock) StageFakeData() (uint64, []sectorbuilder.PublicPieceInfo, erro
buf := make([]byte, usize)
rand.Read(buf)
pi, err := sb.AddPiece(usize, sid, bytes.NewReader(buf), nil)
pi, err := sb.AddPiece(context.TODO(), usize, sid, bytes.NewReader(buf), nil)
if err != nil {
return 0, nil, err
}
@ -309,6 +310,26 @@ func (sb *SBMock) StageFakeData() (uint64, []sectorbuilder.PublicPieceInfo, erro
return sid, []sectorbuilder.PublicPieceInfo{pi}, nil
}
func (sb *SBMock) FinalizeSector(context.Context, uint64) error {
return nil
}
func (sb *SBMock) DropStaged(context.Context, uint64) error {
return nil
}
func (sb *SBMock) SectorPath(typ fs.DataType, sectorID uint64) (fs.SectorPath, error) {
panic("implement me")
}
func (sb *SBMock) AllocSectorPath(typ fs.DataType, sectorID uint64, cache bool) (fs.SectorPath, error) {
panic("implement me")
}
func (sb *SBMock) ReleaseSector(fs.DataType, fs.SectorPath) {
panic("implement me")
}
func (m mockVerif) VerifyElectionPost(ctx context.Context, sectorSize uint64, sectorInfo sectorbuilder.SortedPublicSectorInfo, challengeSeed []byte, proof []byte, candidates []sectorbuilder.EPostCandidate, proverID address.Address) (bool, error) {
panic("implement me")
}

10
storage/sealing/fsm.go
View File

@ -48,10 +48,14 @@ var fsmPlanners = []func(events []statemachine.Event, state *SectorInfo) error{
),
api.Committing: planCommitting,
api.CommitWait: planOne(
on(SectorProving{}, api.Proving),
on(SectorProving{}, api.FinalizeSector),
on(SectorCommitFailed{}, api.CommitFailed),
),
api.FinalizeSector: planOne(
on(SectorFinalized{}, api.Proving),
),
api.Proving: planOne(
on(SectorFaultReported{}, api.FaultReported),
on(SectorFaulty{}, api.Faulty),
@ -150,6 +154,8 @@ func (m *Sealing) plan(events []statemachine.Event, state *SectorInfo) (func(sta
return m.handleCommitting, nil
case api.CommitWait:
return m.handleCommitWait, nil
case api.FinalizeSector:
return m.handleFinalizeSector, nil
case api.Proving:
// TODO: track sector health / expiration
log.Infof("Proving sector %d", state.SectorID)
@ -205,6 +211,8 @@ func planCommitting(events []statemachine.Event, state *SectorInfo) error {
state.State = api.SealCommitFailed
case SectorSealFailed:
state.State = api.CommitFailed
case SectorCommitFailed:
state.State = api.CommitFailed
default:
return xerrors.Errorf("planCommitting got event of unknown type %T, events: %+v", event.User, events)
}

8
storage/sealing/fsm_events.go
View File

@ -120,6 +120,14 @@ type SectorProving struct{}
func (evt SectorProving) apply(*SectorInfo) {}
type SectorFinalized struct{}
func (evt SectorFinalized) apply(*SectorInfo) {}
type SectorFinalizeFailed struct{ error }
func (evt SectorFinalizeFailed) apply(*SectorInfo) {}
// Failed state recovery
type SectorRetrySeal struct{}

20
storage/sealing/fsm_test.go
View File

@ -50,6 +50,9 @@ func TestHappyPath(t *testing.T) {
require.Equal(m.t, m.state.State, api.CommitWait)
m.planSingle(SectorProving{})
require.Equal(m.t, m.state.State, api.FinalizeSector)
m.planSingle(SectorFinalized{})
require.Equal(m.t, m.state.State, api.Proving)
}
@ -81,5 +84,22 @@ func TestSeedRevert(t *testing.T) {
require.Equal(m.t, m.state.State, api.CommitWait)
m.planSingle(SectorProving{})
require.Equal(m.t, m.state.State, api.FinalizeSector)
m.planSingle(SectorFinalized{})
require.Equal(m.t, m.state.State, api.Proving)
}
func TestPlanCommittingHandlesSectorCommitFailed(t *testing.T) {
m := test{
s: &Sealing{},
t: t,
state: &SectorInfo{State: api.Committing},
}
events := []statemachine.Event{{SectorCommitFailed{}}}
require.NoError(t, planCommitting(events, m.state))
require.Equal(t, api.SectorStates[api.CommitFailed], api.SectorStates[m.state.State])
}

36
storage/sealing/garbage.go
View File

@ -5,7 +5,9 @@ import (
"context"
"io"
"math"
"math/bits"
"math/rand"
"runtime"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
"golang.org/x/xerrors"
@ -14,17 +16,32 @@ import (
"github.com/filecoin-project/lotus/chain/types"
)
func (m *Sealing) pledgeReader(size uint64, parts uint64) io.Reader {
parts = 1 << bits.Len64(parts) // round down to nearest power of 2
if size/parts < 127 {
parts = size / 127
}
piece := sectorbuilder.UserBytesForSectorSize((size/127 + size) / parts)
readers := make([]io.Reader, parts)
for i := range readers {
readers[i] = io.LimitReader(rand.New(rand.NewSource(42+int64(i))), int64(piece))
}
return io.MultiReader(readers...)
}
func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPieceSizes []uint64, sizes ...uint64) ([]Piece, error) {
if len(sizes) == 0 {
return nil, nil
}
log.Infof("Pledge %d, contains %+v", sectorID, existingPieceSizes)
deals := make([]actors.StorageDealProposal, len(sizes))
for i, size := range sizes {
release := m.sb.RateLimit()
commP, err := sectorbuilder.GeneratePieceCommitment(io.LimitReader(rand.New(rand.NewSource(42)), int64(size)), size)
release()
commP, err := m.fastPledgeCommitment(size, uint64(runtime.NumCPU()))
if err != nil {
return nil, err
}
@ -44,6 +61,8 @@ func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPie
deals[i] = sdp
}
log.Infof("Publishing deals for %d", sectorID)
params, aerr := actors.SerializeParams(&actors.PublishStorageDealsParams{
Deals: deals,
})
@ -63,7 +82,7 @@ func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPie
if err != nil {
return nil, err
}
r, err := m.api.StateWaitMsg(ctx, smsg.Cid())
r, err := m.api.StateWaitMsg(ctx, smsg.Cid()) // TODO: more finality
if err != nil {
return nil, err
}
@ -78,12 +97,13 @@ func (m *Sealing) pledgeSector(ctx context.Context, sectorID uint64, existingPie
return nil, xerrors.New("got unexpected number of DealIDs from PublishStorageDeals")
}
out := make([]Piece, len(sizes))
log.Infof("Deals for sector %d: %+v", sectorID, resp.DealIDs)
out := make([]Piece, len(sizes))
for i, size := range sizes {
ppi, err := m.sb.AddPiece(size, sectorID, io.LimitReader(rand.New(rand.NewSource(42)), int64(size)), existingPieceSizes)
ppi, err := m.sb.AddPiece(ctx, size, sectorID, m.pledgeReader(size, uint64(runtime.NumCPU())), existingPieceSizes)
if err != nil {
return nil, err
return nil, xerrors.Errorf("add piece: %w", err)
}
existingPieceSizes = append(existingPieceSizes, size)

3
storage/sealing/sealing.go
View File

@ -112,7 +112,7 @@ func (m *Sealing) AllocatePiece(size uint64) (sectorID uint64, offset uint64, er
func (m *Sealing) 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{})
ppi, err := m.sb.AddPiece(ctx, size, sectorID, r, []uint64{})
if err != nil {
return xerrors.Errorf("adding piece to sector: %w", err)
}
@ -121,6 +121,7 @@ func (m *Sealing) SealPiece(ctx context.Context, size uint64, r io.Reader, secto
}
func (m *Sealing) newSector(ctx context.Context, sid uint64, dealID uint64, ppi sectorbuilder.PublicPieceInfo) error {
log.Infof("Start sealing %d", sid)
return m.sectors.Send(sid, SectorStart{
id: sid,
pieces: []Piece{

18
storage/sealing/states.go
View File

@ -4,6 +4,7 @@ import (
"context"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/go-sectorbuilder/fs"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/build"
@ -232,6 +233,23 @@ func (m *Sealing) handleCommitWait(ctx statemachine.Context, sector SectorInfo)
return ctx.Send(SectorProving{})
}
func (m *Sealing) handleFinalizeSector(ctx statemachine.Context, sector SectorInfo) error {
// TODO: Maybe wait for some finality
if err := m.sb.FinalizeSector(ctx.Context(), sector.SectorID); err != nil {
if !xerrors.Is(err, fs.ErrNoSuitablePath) {
return ctx.Send(SectorFinalizeFailed{xerrors.Errorf("finalize sector: %w", err)})
}
log.Warnf("finalize sector: %v", err)
}
if err := m.sb.DropStaged(ctx.Context(), sector.SectorID); err != nil {
return ctx.Send(SectorFinalizeFailed{xerrors.Errorf("drop staged: %w", err)})
}
return ctx.Send(SectorFinalized{})
}
func (m *Sealing) handleFaulty(ctx statemachine.Context, sector SectorInfo) error {
// TODO: check if the fault has already been reported, and that this sector is even valid

43
storage/sealing/utils.go
View File

@ -1,7 +1,12 @@
package sealing
import (
"io"
"math/bits"
"math/rand"
"sync"
"github.com/hashicorp/go-multierror"
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
)
@ -42,6 +47,44 @@ func fillersFromRem(toFill uint64) ([]uint64, error) {
return out, nil
}
func (m *Sealing) fastPledgeCommitment(size uint64, parts uint64) (commP [sectorbuilder.CommLen]byte, err error) {
parts = 1 << bits.Len64(parts) // round down to nearest power of 2
if size/parts < 127 {
parts = size / 127
}
piece := sectorbuilder.UserBytesForSectorSize((size + size/127) / parts)
out := make([]sectorbuilder.PublicPieceInfo, parts)
var lk sync.Mutex
var wg sync.WaitGroup
wg.Add(int(parts))
for i := uint64(0); i < parts; i++ {
go func(i uint64) {
defer wg.Done()
commP, perr := sectorbuilder.GeneratePieceCommitment(io.LimitReader(rand.New(rand.NewSource(42+int64(i))), int64(piece)), piece)
lk.Lock()
if perr != nil {
err = multierror.Append(err, perr)
}
out[i] = sectorbuilder.PublicPieceInfo{
Size: piece,
CommP: commP,
}
lk.Unlock()
}(i)
}
wg.Wait()
if err != nil {
return [32]byte{}, err
}
return sectorbuilder.GenerateDataCommitment(m.sb.SectorSize(), out)
}
func (m *Sealing) ListSectors() ([]SectorInfo, error) {
var sectors []SectorInfo
if err := m.sectors.List(&sectors); err != nil {

18
storage/sealing/utils_test.go
View File

@ -1,6 +1,7 @@
package sealing
import (
"github.com/filecoin-project/lotus/storage/sbmock"
"testing"
"github.com/stretchr/testify/assert"
@ -42,5 +43,20 @@ func TestFillersFromRem(t *testing.T) {
ub = sectorbuilder.UserBytesForSectorSize(uint64(9) << i)
testFill(t, ub, []uint64{ub1, ub4})
}
}
func TestFastPledge(t *testing.T) {
sz := uint64(16 << 20)
s := Sealing{sb: sbmock.NewMockSectorBuilder(0, sz)}
if _, err := s.fastPledgeCommitment(sectorbuilder.UserBytesForSectorSize(sz), 5); err != nil {
t.Fatalf("%+v", err)
}
sz = uint64(1024)
s = Sealing{sb: sbmock.NewMockSectorBuilder(0, sz)}
if _, err := s.fastPledgeCommitment(sectorbuilder.UserBytesForSectorSize(sz), 64); err != nil {
t.Fatalf("%+v", err)
}
}

14
tools/dockers/docker-examples/basic-miner-busybox/Dockerfile
View File

@ -1,4 +1,4 @@
FROM golang:1.13.4-buster
FROM golang:1.13-buster
MAINTAINER ldoublewood <ldoublewood@gmail.com>
ENV SRC_DIR /lotus
@ -44,10 +44,12 @@ RUN cd $SRC_DIR \
COPY . $SRC_DIR
ARG MAKE_TARGET=all
# Build the thing.
RUN cd $SRC_DIR \
&& . $HOME/.cargo/env \
&& make
&& make $MAKE_TARGET
# Now comes the actual target image, which aims to be as small as possible.
FROM busybox:1-glibc
@ -56,7 +58,7 @@ MAINTAINER ldoublewood <ldoublewood@gmail.com>
# Get the executable binary and TLS CAs from the build container.
ENV SRC_DIR /lotus
COPY --from=0 $SRC_DIR/lotus /usr/local/bin/lotus
COPY --from=0 $SRC_DIR/lotus-storage-miner /usr/local/bin/lotus-storage-miner
COPY --from=0 $SRC_DIR/lotus-* /usr/local/bin/
COPY --from=0 /tmp/su-exec/su-exec /sbin/su-exec
COPY --from=0 /tmp/tini /sbin/tini
COPY --from=0 /etc/ssl/certs /etc/ssl/certs
@ -79,13 +81,15 @@ EXPOSE 5678
ENV HOME_PATH /data
ENV PARAMCACHE_PATH /var/tmp/filecoin-proof-parameters
RUN mkdir -p $HOME_PATH \
RUN mkdir -p $HOME_PATH $PARAMCACHE_PATH \
&& adduser -D -h $HOME_PATH -u 1000 -G users lotus \
&& chown lotus:users $HOME_PATH
&& chown lotus:users $HOME_PATH $PARAMCACHE_PATH
VOLUME $HOME_PATH
VOLUME $PARAMCACHE_PATH
USER lotus
# Execute the daemon subcommand by default
CMD ["/sbin/tini", "--", "lotus", "daemon"]