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

Merge branch 'next' into inmem-journal

Browse Source
This commit is contained in:
Raúl Kripalani
2020-07-20 10:38:58 +01:00
parent becbff0b13 db92b99ec1
commit 4d2d8b2d11
84 changed files with 4902 additions and 4114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/hello/cbor_gen.go
+5 -1
View File
@@ -7,7 +7,7 @@ import (
"io"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/ipfs/go-cid"
cid "github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
xerrors "golang.org/x/xerrors"
)
@@ -67,6 +67,8 @@ func (t *HelloMessage) MarshalCBOR(w io.Writer) error {
}
func (t *HelloMessage) UnmarshalCBOR(r io.Reader) error {
*t = HelloMessage{}
br := cbg.GetPeeker(r)
scratch := make([]byte, 8)
@@ -197,6 +199,8 @@ func (t *LatencyMessage) MarshalCBOR(w io.Writer) error {
}
func (t *LatencyMessage) UnmarshalCBOR(r io.Reader) error {
*t = LatencyMessage{}
br := cbg.GetPeeker(r)
scratch := make([]byte, 8)
node/impl/client/client.go
+1 -1
View File
@@ -296,7 +296,7 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (*api.ImportRes
}, nil
}
func (a *API) ClientRemoveImport(ctx context.Context, importID int64) error {
func (a *API) ClientRemoveImport(ctx context.Context, importID int) error {
return a.imgr().Remove(importID)
}
node/impl/full/chain.go
+1 -2
View File
@@ -11,7 +11,6 @@ import (
"sync"
"github.com/filecoin-project/go-amt-ipld/v2"
commcid "github.com/filecoin-project/go-fil-commcid"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/crypto"
"github.com/filecoin-project/specs-actors/actors/util/adt"
@@ -206,7 +205,7 @@ func (a *ChainAPI) ChainStatObj(ctx context.Context, obj cid.Cid, base cid.Cid)
var collect = true
walker := func(ctx context.Context, c cid.Cid) ([]*ipld.Link, error) {
if c.Prefix().MhType == uint64(commcid.FC_SEALED_V1) || c.Prefix().MhType == uint64(commcid.FC_UNSEALED_V1) {
if c.Prefix().Codec == cid.FilCommitmentSealed || c.Prefix().Codec == cid.FilCommitmentUnsealed {
return []*ipld.Link{}, nil
}
node/impl/full/multisig.go
+40
View File
@@ -2,6 +2,7 @@ package full
import (
"context"
"github.com/filecoin-project/specs-actors/actors/abi/big"
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/lotus/api"
@@ -135,6 +136,33 @@ func (a *MsigAPI) MsigPropose(ctx context.Context, msig address.Address, to addr
return smsg.Cid(), nil
}
func (a *MsigAPI) MsigSwapPropose(ctx context.Context, msig address.Address, src address.Address, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
enc, actErr := serializeSwapParams(oldAdd, newAdd)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigPropose(ctx, msig, msig, big.Zero(), src, uint64(builtin.MethodsMultisig.SwapSigner), enc)
}
func (a *MsigAPI) MsigSwapApprove(ctx context.Context, msig address.Address, src address.Address, txID uint64, proposer address.Address, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
enc, actErr := serializeSwapParams(oldAdd, newAdd)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigApprove(ctx, msig, txID, proposer, msig, big.Zero(), src, uint64(builtin.MethodsMultisig.SwapSigner), enc)
}
func (a *MsigAPI) MsigSwapCancel(ctx context.Context, msig address.Address, src address.Address, txID uint64, oldAdd address.Address, newAdd address.Address) (cid.Cid, error) {
enc, actErr := serializeSwapParams(oldAdd, newAdd)
if actErr != nil {
return cid.Undef, actErr
}
return a.MsigCancel(ctx, msig, txID, msig, big.Zero(), src, uint64(builtin.MethodsMultisig.SwapSigner), enc)
}
func (a *MsigAPI) MsigApprove(ctx context.Context, msig address.Address, txID uint64, proposer address.Address, to address.Address, amt types.BigInt, src address.Address, method uint64, params []byte) (cid.Cid, error) {
return a.msigApproveOrCancel(ctx, api.MsigApprove, msig, txID, proposer, to, amt, src, method, params)
}
@@ -223,3 +251,15 @@ func (a *MsigAPI) msigApproveOrCancel(ctx context.Context, operation api.MsigPro
return smsg.Cid(), nil
}
func serializeSwapParams(old address.Address, new address.Address) ([]byte, error) {
enc, actErr := actors.SerializeParams(&samsig.SwapSignerParams{
From: old,
To: new,
})
if actErr != nil {
return nil, actErr
}
return enc, nil
}
node/impl/full/state.go
+262 -68
View File
@@ -3,6 +3,7 @@ package full
import (
"bytes"
"context"
"errors"
"fmt"
"strconv"
@@ -15,6 +16,7 @@ import (
"github.com/filecoin-project/go-address"
"github.com/filecoin-project/go-amt-ipld/v2"
"github.com/filecoin-project/go-bitfield"
"github.com/filecoin-project/sector-storage/ffiwrapper"
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/abi/big"
@@ -25,6 +27,7 @@ import (
"github.com/filecoin-project/specs-actors/actors/builtin/power"
"github.com/filecoin-project/specs-actors/actors/builtin/reward"
"github.com/filecoin-project/specs-actors/actors/builtin/verifreg"
"github.com/filecoin-project/specs-actors/actors/util/adt"
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/chain/actors"
@@ -40,6 +43,8 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
)
var errBreakForeach = errors.New("break")
type StateAPI struct {
fx.In
@@ -65,19 +70,42 @@ func (a *StateAPI) StateMinerSectors(ctx context.Context, addr address.Address,
return stmgr.GetMinerSectorSet(ctx, a.StateManager, ts, addr, filter, filterOut)
}
func (a *StateAPI) StateMinerProvingSet(ctx context.Context, addr address.Address, tsk types.TipSetKey) ([]*api.ChainSectorInfo, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
func (a *StateAPI) StateMinerActiveSectors(ctx context.Context, maddr address.Address, tsk types.TipSetKey) ([]*api.ChainSectorInfo, error) {
var out []*api.ChainSectorInfo
err := a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(maddr,
a.StateManager.WithActorState(ctx, func(store adt.Store, mas *miner.State) error {
var allActive []*abi.BitField
err := a.StateManager.WithDeadlines(
a.StateManager.WithEachDeadline(
a.StateManager.WithEachPartition(func(store adt.Store, partIdx uint64, partition *miner.Partition) error {
active, err := partition.ActiveSectors()
if err != nil {
return xerrors.Errorf("partition.ActiveSectors: %w", err)
}
allActive = append(allActive, active)
return nil
})))(store, mas)
if err != nil {
return xerrors.Errorf("with deadlines: %w", err)
}
active, err := bitfield.MultiMerge(allActive...)
if err != nil {
return xerrors.Errorf("merging active sector bitfields: %w", err)
}
out, err = stmgr.LoadSectorsFromSet(ctx, a.Chain.Blockstore(), mas.Sectors, active, false)
return err
})))
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
return nil, xerrors.Errorf("getting active sectors from partitions: %w", err)
}
var mas miner.State
_, err = a.StateManager.LoadActorState(ctx, addr, &mas, ts)
if err != nil {
return nil, xerrors.Errorf("(get sset) failed to load miner actor state: %w", err)
}
return stmgr.GetProvingSetRaw(ctx, a.StateManager, mas)
return out, nil
}
func (a *StateAPI) StateMinerInfo(ctx context.Context, actor address.Address, tsk types.TipSetKey) (api.MinerInfo, error) {
@@ -93,12 +121,30 @@ func (a *StateAPI) StateMinerInfo(ctx context.Context, actor address.Address, ts
return api.NewApiMinerInfo(mi), nil
}
func (a *StateAPI) StateMinerDeadlines(ctx context.Context, m address.Address, tsk types.TipSetKey) (*miner.Deadlines, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
return stmgr.GetMinerDeadlines(ctx, a.StateManager, ts, m)
func (a *StateAPI) StateMinerDeadlines(ctx context.Context, m address.Address, tsk types.TipSetKey) ([]*miner.Deadline, error) {
var out []*miner.Deadline
return out, a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(m,
a.StateManager.WithActorState(ctx,
a.StateManager.WithDeadlines(
a.StateManager.WithEachDeadline(
func(store adt.Store, idx uint64, deadline *miner.Deadline) error {
out = append(out, deadline)
return nil
})))))
}
func (a *StateAPI) StateMinerPartitions(ctx context.Context, m address.Address, dlIdx uint64, tsk types.TipSetKey) ([]*miner.Partition, error) {
var out []*miner.Partition
return out, a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(m,
a.StateManager.WithActorState(ctx,
a.StateManager.WithDeadlines(
a.StateManager.WithDeadline(dlIdx,
a.StateManager.WithEachPartition(func(store adt.Store, partIdx uint64, partition *miner.Partition) error {
out = append(out, partition)
return nil
}))))))
}
func (a *StateAPI) StateMinerProvingDeadline(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*miner.DeadlineInfo, error) {
@@ -113,19 +159,32 @@ func (a *StateAPI) StateMinerProvingDeadline(ctx context.Context, addr address.A
return nil, xerrors.Errorf("(get sset) failed to load miner actor state: %w", err)
}
return miner.ComputeProvingPeriodDeadline(mas.ProvingPeriodStart, ts.Height()), nil
return mas.DeadlineInfo(ts.Height()).NextNotElapsed(), nil
}
func (a *StateAPI) StateMinerFaults(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.BitField, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
out := abi.NewBitField()
err := a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(addr,
a.StateManager.WithActorState(ctx,
a.StateManager.WithDeadlines(
a.StateManager.WithEachDeadline(
a.StateManager.WithEachPartition(func(store adt.Store, idx uint64, partition *miner.Partition) (err error) {
out, err = bitfield.MergeBitFields(out, partition.Faults)
return err
}))))))
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
return nil, err
}
return stmgr.GetMinerFaults(ctx, a.StateManager, ts, addr)
return out, err
}
func (a *StateAPI) StateAllMinerFaults(ctx context.Context, lookback abi.ChainEpoch, endTsk types.TipSetKey) ([]*api.Fault, error) {
endTs, err := a.Chain.GetTipSetFromKey(endTsk)
return nil, xerrors.Errorf("fixme")
/*endTs, err := a.Chain.GetTipSetFromKey(endTsk)
if err != nil {
return nil, xerrors.Errorf("loading end tipset %s: %w", endTsk, err)
}
@@ -162,15 +221,26 @@ func (a *StateAPI) StateAllMinerFaults(ctx context.Context, lookback abi.ChainEp
}
}
return allFaults, nil
return allFaults, nil*/
}
func (a *StateAPI) StateMinerRecoveries(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.BitField, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
out := abi.NewBitField()
err := a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(addr,
a.StateManager.WithActorState(ctx,
a.StateManager.WithDeadlines(
a.StateManager.WithEachDeadline(
a.StateManager.WithEachPartition(func(store adt.Store, idx uint64, partition *miner.Partition) (err error) {
out, err = bitfield.MergeBitFields(out, partition.Recoveries)
return err
}))))))
if err != nil {
return nil, xerrors.Errorf("loading tipset %s: %w", tsk, err)
return nil, err
}
return stmgr.GetMinerRecoveries(ctx, a.StateManager, ts, addr)
return out, err
}
func (a *StateAPI) StateMinerPower(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*api.MinerPower, error) {
@@ -598,11 +668,51 @@ func (a *StateAPI) StateChangedActors(ctx context.Context, old cid.Cid, new cid.
}
func (a *StateAPI) StateMinerSectorCount(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MinerSectors, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
var out api.MinerSectors
err := a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(addr,
a.StateManager.WithActorState(ctx, func(store adt.Store, mas *miner.State) error {
var allActive []*abi.BitField
err := a.StateManager.WithDeadlines(
a.StateManager.WithEachDeadline(
a.StateManager.WithEachPartition(func(store adt.Store, partIdx uint64, partition *miner.Partition) error {
active, err := partition.ActiveSectors()
if err != nil {
return xerrors.Errorf("partition.ActiveSectors: %w", err)
}
allActive = append(allActive, active)
return nil
})))(store, mas)
if err != nil {
return xerrors.Errorf("with deadlines: %w", err)
}
active, err := bitfield.MultiMerge(allActive...)
if err != nil {
return xerrors.Errorf("merging active sector bitfields: %w", err)
}
out.Active, err = active.Count()
if err != nil {
return xerrors.Errorf("counting active sectors: %w", err)
}
sarr, err := adt.AsArray(store, mas.Sectors)
if err != nil {
return err
}
out.Sectors = sarr.Length()
return nil
})))
if err != nil {
return api.MinerSectors{}, xerrors.Errorf("loading tipset %s: %w", tsk, err)
return api.MinerSectors{}, err
}
return stmgr.SectorSetSizes(ctx, a.StateManager, addr, ts)
return out, nil
}
func (a *StateAPI) StateSectorPreCommitInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (miner.SectorPreCommitOnChainInfo, error) {
@@ -621,6 +731,103 @@ func (a *StateAPI) StateSectorGetInfo(ctx context.Context, maddr address.Address
return stmgr.MinerSectorInfo(ctx, a.StateManager, maddr, n, ts)
}
type sectorPartitionCb func(store adt.Store, mas *miner.State, di uint64, pi uint64, part *miner.Partition) error
func (a *StateAPI) sectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tsk types.TipSetKey, cb sectorPartitionCb) error {
return a.StateManager.WithParentStateTsk(tsk,
a.StateManager.WithActor(maddr,
a.StateManager.WithActorState(ctx, func(store adt.Store, mas *miner.State) error {
return a.StateManager.WithDeadlines(func(store adt.Store, deadlines *miner.Deadlines) error {
err := a.StateManager.WithEachDeadline(func(store adt.Store, di uint64, deadline *miner.Deadline) error {
return a.StateManager.WithEachPartition(func(store adt.Store, pi uint64, partition *miner.Partition) error {
set, err := partition.Sectors.IsSet(uint64(sectorNumber))
if err != nil {
return xerrors.Errorf("is set: %w", err)
}
if set {
if err := cb(store, mas, di, pi, partition); err != nil {
return err
}
return errBreakForeach
}
return nil
})(store, di, deadline)
})(store, deadlines)
if err == errBreakForeach {
err = nil
}
return err
})(store, mas)
})))
}
func (a *StateAPI) StateSectorExpiration(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tsk types.TipSetKey) (*api.SectorExpiration, error) {
var onTimeEpoch, earlyEpoch abi.ChainEpoch
err := a.sectorPartition(ctx, maddr, sectorNumber, tsk, func(store adt.Store, mas *miner.State, di uint64, pi uint64, part *miner.Partition) error {
quant := mas.QuantEndOfDeadline()
expirations, err := miner.LoadExpirationQueue(store, part.ExpirationsEpochs, quant)
if err != nil {
return xerrors.Errorf("loading expiration queue: %w", err)
}
var eset miner.ExpirationSet
return expirations.Array.ForEach(&eset, func(epoch int64) error {
set, err := eset.OnTimeSectors.IsSet(uint64(sectorNumber))
if err != nil {
return xerrors.Errorf("checking if sector is in onTime set: %w", err)
}
if set {
onTimeEpoch = abi.ChainEpoch(epoch)
}
set, err = eset.EarlySectors.IsSet(uint64(sectorNumber))
if err != nil {
return xerrors.Errorf("checking if sector is in early set: %w", err)
}
if set {
earlyEpoch = abi.ChainEpoch(epoch)
}
return nil
})
})
if err != nil {
return nil, err
}
if onTimeEpoch == 0 {
return nil, xerrors.Errorf("expiration for sector %d not found", sectorNumber)
}
return &api.SectorExpiration{
OnTime: onTimeEpoch,
Early: earlyEpoch,
}, nil
}
func (a *StateAPI) StateSectorPartition(ctx context.Context, maddr address.Address, sectorNumber abi.SectorNumber, tsk types.TipSetKey) (*api.SectorLocation, error) {
var found *api.SectorLocation
err := a.sectorPartition(ctx, maddr, sectorNumber, tsk, func(store adt.Store, mas *miner.State, di, pi uint64, partition *miner.Partition) error {
found = &api.SectorLocation{
Deadline: di,
Partition: pi,
}
return errBreakForeach
})
if err != nil {
return nil, err
}
if found == nil {
}
return found, nil
}
func (a *StateAPI) StateListMessages(ctx context.Context, match *types.Message, tsk types.TipSetKey, toheight abi.ChainEpoch) ([]cid.Cid, error) {
ts, err := a.Chain.GetTipSetFromKey(tsk)
if err != nil {
@@ -729,39 +936,27 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
return types.EmptyInt, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
as := store.ActorStore(ctx, a.Chain.Blockstore())
var minerState miner.State
{
act, err := a.StateManager.GetActor(maddr, ts)
if err != nil {
return types.EmptyInt, err
}
if err := as.Get(ctx, act.Head, &minerState); err != nil {
return types.EmptyInt, err
}
}
var powerState power.State
{
act, err := a.StateManager.GetActor(builtin.StoragePowerActorAddr, ts)
if err != nil {
return types.EmptyInt, err
}
if err := as.Get(ctx, act.Head, &powerState); err != nil {
return types.EmptyInt, err
}
}
var rewardState reward.State
{
act, err := a.StateManager.GetActor(builtin.RewardActorAddr, ts)
if err != nil {
return types.EmptyInt, err
err = a.StateManager.WithParentStateTsk(tsk, func(state *state.StateTree) error {
if err := a.StateManager.WithActor(maddr, a.StateManager.WithActorState(ctx, &minerState))(state); err != nil {
return xerrors.Errorf("getting miner state: %w", err)
}
if err := as.Get(ctx, act.Head, &rewardState); err != nil {
return types.EmptyInt, err
if err := a.StateManager.WithActor(builtin.StoragePowerActorAddr, a.StateManager.WithActorState(ctx, &powerState))(state); err != nil {
return xerrors.Errorf("getting power state: %w", err)
}
if err := a.StateManager.WithActor(builtin.RewardActorAddr, a.StateManager.WithActorState(ctx, &rewardState))(state); err != nil {
return xerrors.Errorf("getting reward state: %w", err)
}
return nil
})
if err != nil {
return types.EmptyInt, err
}
var dealWeights market.VerifyDealsForActivationReturn
@@ -805,7 +1000,7 @@ func (a *StateAPI) StateMinerInitialPledgeCollateral(ctx context.Context, maddr
}
sectorWeight := miner.QAPowerForWeight(ssize, duration, dealWeights.DealWeight, dealWeights.VerifiedDealWeight)
initialPledge := miner.InitialPledgeForPower(sectorWeight, powerState.TotalQualityAdjPower, powerState.TotalPledgeCollateral, rewardState.ThisEpochReward, circSupply)
initialPledge := miner.InitialPledgeForPower(sectorWeight, powerState.TotalQualityAdjPower, reward.BaselinePowerAt(ts.Height()), powerState.TotalPledgeCollateral, rewardState.ThisEpochReward, circSupply)
return types.BigDiv(types.BigMul(initialPledge, initialPledgeNum), initialPledgeDen), nil
}
@@ -816,24 +1011,23 @@ func (a *StateAPI) StateMinerAvailableBalance(ctx context.Context, maddr address
return types.EmptyInt, xerrors.Errorf("loading tipset %s: %w", tsk, err)
}
act, err := a.StateManager.GetActor(maddr, ts)
if err != nil {
return types.EmptyInt, err
}
var act *types.Actor
var mas miner.State
if err := a.StateManager.WithParentState(ts, a.StateManager.WithActor(maddr, func(actor *types.Actor) error {
act = actor
return a.StateManager.WithActorState(ctx, &mas)(actor)
})); err != nil {
return types.BigInt{}, xerrors.Errorf("getting miner state: %w", err)
}
as := store.ActorStore(ctx, a.Chain.Blockstore())
var st miner.State
if err := as.Get(ctx, act.Head, &st); err != nil {
return types.EmptyInt, err
}
vested, err := st.CheckVestedFunds(as, ts.Height())
vested, err := mas.CheckVestedFunds(as, ts.Height())
if err != nil {
return types.EmptyInt, err
}
return types.BigAdd(st.GetAvailableBalance(act.Balance), vested), nil
return types.BigAdd(mas.GetAvailableBalance(act.Balance), vested), nil
}
// StateVerifiedClientStatus returns the data cap for the given address.
node/impl/full/wallet.go
+5 -1
View File
@@ -36,7 +36,11 @@ func (a *WalletAPI) WalletList(ctx context.Context) ([]address.Address, error) {
}
func (a *WalletAPI) WalletBalance(ctx context.Context, addr address.Address) (types.BigInt, error) {
return a.StateManager.GetBalance(addr, nil)
var bal types.BigInt
return bal, a.StateManager.WithParentStateTsk(types.EmptyTSK, a.StateManager.WithActor(addr, func(act *types.Actor) error {
bal = act.Balance
return nil
}))
}
func (a *WalletAPI) WalletSign(ctx context.Context, k address.Address, msg []byte) (*crypto.Signature, error) {
node/impl/paych/paych.go
+3 -1
View File
@@ -63,6 +63,8 @@ func (a *PaychAPI) PaychNewPayment(ctx context.Context, from, to address.Address
for i, v := range vouchers {
sv, err := a.paychVoucherCreate(ctx, ch.Channel, paych.SignedVoucher{
ChannelAddr: ch.Channel,
Amount: v.Amount,
Lane: uint64(lane),
@@ -161,7 +163,7 @@ func (a *PaychAPI) PaychVoucherAdd(ctx context.Context, ch address.Address, sv *
// actual additional value of this voucher will only be the difference between
// the two.
func (a *PaychAPI) PaychVoucherCreate(ctx context.Context, pch address.Address, amt types.BigInt, lane uint64) (*paych.SignedVoucher, error) {
return a.paychVoucherCreate(ctx, pch, paych.SignedVoucher{Amount: amt, Lane: lane})
return a.paychVoucherCreate(ctx, pch, paych.SignedVoucher{ChannelAddr: pch, Amount: amt, Lane: lane})
}
func (a *PaychAPI) paychVoucherCreate(ctx context.Context, pch address.Address, voucher paych.SignedVoucher) (*paych.SignedVoucher, error) {
node/modules/client.go
+7 -1
View File
@@ -5,6 +5,7 @@ import (
"time"
"github.com/filecoin-project/lotus/lib/bufbstore"
"golang.org/x/xerrors"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/libp2p/go-libp2p-core/host"
@@ -36,7 +37,12 @@ import (
)
func ClientMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
mds, err := importmgr.NewMultiDstore(r, "/client")
ds, err := r.Datastore("/client")
if err != nil {
return nil, xerrors.Errorf("getting datastore out of reop: %w", err)
}
mds, err := importmgr.NewMultiDstore(ds)
if err != nil {
return nil, err
}
node/modules/storageminer.go
+1 -7
View File
@@ -460,13 +460,7 @@ func SectorStorage(mctx helpers.MetricsCtx, lc fx.Lifecycle, ls stores.LocalStor
}
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
if err := sst.Close(); err != nil {
log.Errorf("%+v", err)
}
return nil
},
OnStop: sst.Close,
})
return sst, nil
node/repo/fsrepo.go
+3 -5
View File
@@ -226,11 +226,9 @@ type fsLockedRepo struct {
repoType RepoType
closer io.Closer
ds map[string]datastore.Batching
multiDs map[string]map[int64]datastore.Batching
dsErr error
dsOnce sync.Once
dsLk sync.Mutex
ds map[string]datastore.Batching
dsErr error
dsOnce sync.Once
storageLk sync.Mutex
configLk sync.Mutex
node/repo/fsrepo_ds.go
+6 -171
View File
@@ -1,13 +1,11 @@
package repo
import (
"fmt"
"github.com/ipfs/go-datastore"
"golang.org/x/xerrors"
"io/ioutil"
"os"
"path/filepath"
"strconv"
"github.com/ipfs/go-datastore"
"golang.org/x/xerrors"
dgbadger "github.com/dgraph-io/badger/v2"
badger "github.com/ipfs/go-ds-badger2"
@@ -24,9 +22,7 @@ var fsDatastores = map[string]dsCtor{
// Those need to be fast for large writes... but also need a really good GC :c
"staging": badgerDs, // miner specific
}
var fsMultiDatastores = map[string]dsCtor{
"client": badgerDs, // client specific
}
@@ -68,78 +64,11 @@ func (fsr *fsLockedRepo) openDatastores() (map[string]datastore.Batching, error)
return out, nil
}
func (fsr *fsLockedRepo) openMultiDatastores() (map[string]map[int64]datastore.Batching, error) {
out := map[string]map[int64]datastore.Batching{}
for p, ctor := range fsMultiDatastores {
path := fsr.join(filepath.Join(fsDatastore, p))
if err := os.MkdirAll(path, 0755); err != nil {
return nil, xerrors.Errorf("mkdir %s: %w", path, err)
}
di, err := ioutil.ReadDir(path)
if err != nil {
return nil, xerrors.Errorf("readdir '%s': %w", path, err)
}
out[p] = map[int64]datastore.Batching{}
for _, info := range di {
path := filepath.Join(path, info.Name())
prefix := datastore.NewKey(p)
id, err := strconv.ParseInt(info.Name(), 10, 64)
if err != nil {
log.Errorf("error parsing multi-datastore id for '%s': %w", path, err)
continue
}
// TODO: optimization: don't init datastores we don't need
ds, err := ctor(path)
if err != nil {
return nil, xerrors.Errorf("opening datastore %s: %w", prefix, err)
}
ds = measure.New("fsrepo."+p+"."+info.Name(), ds)
out[p][id] = ds
}
}
return out, nil
}
func (fsr *fsLockedRepo) openMultiDatastore(ns string, idx int64) (datastore.Batching, error) {
ctor, ok := fsMultiDatastores[ns]
if !ok {
return nil, xerrors.Errorf("no multi-datastore with namespace '%s'", ns)
}
si := fmt.Sprintf("%d", idx)
path := fsr.join(filepath.Join(fsDatastore, ns, si))
ds, err := ctor(path)
if err != nil {
return nil, xerrors.Errorf("opening datastore %s: %w", path, err)
}
ds = measure.New("fsrepo."+ns+"."+si, ds)
return ds, nil
}
func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
fsr.dsOnce.Do(func() {
var err error
fsr.ds, err = fsr.openDatastores()
if err != nil {
fsr.dsErr = err
return
}
fsr.multiDs, fsr.dsErr = fsr.openMultiDatastores()
fsr.ds, fsr.dsErr = fsr.openDatastores()
})
if fsr.dsErr != nil {
return nil, fsr.dsErr
}
@@ -147,99 +76,5 @@ func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
if ok {
return ds, nil
}
k := datastore.NewKey(ns)
parts := k.List()
if len(parts) != 2 {
return nil, xerrors.Errorf("expected multi-datastore namespace to have 2 parts")
}
fsr.dsLk.Lock()
defer fsr.dsLk.Unlock()
mds, ok := fsr.multiDs[parts[0]]
if !ok {
return nil, xerrors.Errorf("no multi-datastore with namespace %s", ns)
}
idx, err := strconv.ParseInt(parts[1], 10, 64)
if err != nil {
return nil, xerrors.Errorf("parsing mult-datastore index('%s'): %w", parts[1], err)
}
ds, ok = mds[idx]
if !ok {
ds, err = fsr.openMultiDatastore(parts[0], idx)
if err != nil {
return nil, xerrors.Errorf("opening multi-datastore: %w", err)
}
mds[idx] = ds
}
return ds, nil
}
func (fsr *fsLockedRepo) ListDatastores(ns string) ([]int64, error) {
k := datastore.NewKey(ns)
parts := k.List()
if len(parts) != 1 {
return nil, xerrors.Errorf("expected multi-datastore namespace to have 1 part")
}
fsr.dsLk.Lock()
defer fsr.dsLk.Unlock()
mds, ok := fsr.multiDs[parts[0]]
if !ok {
return nil, xerrors.Errorf("no multi-datastore with namespace %s", ns)
}
out := make([]int64, 0, len(mds))
for i := range mds {
out = append(out, i)
}
return out, nil
}
func (fsr *fsLockedRepo) DeleteDatastore(ns string) error {
k := datastore.NewKey(ns)
parts := k.List()
if len(parts) != 2 {
return xerrors.Errorf("expected multi-datastore namespace to have 2 parts")
}
mds, ok := fsr.multiDs[parts[0]]
if !ok {
return xerrors.Errorf("no multi-datastore with namespace %s", ns)
}
idx, err := strconv.ParseInt(parts[1], 10, 64)
if err != nil {
return xerrors.Errorf("parsing mult-datastore index('%s'): %w", parts[1], err)
}
fsr.dsLk.Lock()
defer fsr.dsLk.Unlock()
ds, ok := mds[idx]
if !ok {
return xerrors.Errorf("no multi-datastore with at index (namespace %s)", ns)
}
delete(mds, idx)
if err := ds.Close(); err != nil {
return xerrors.Errorf("closing datastore: %w", err)
}
path := fsr.join(filepath.Join(fsDatastore, parts[0], parts[1]))
log.Warnw("removing sub-datastore", "path", path, "namespace", ns)
if err := os.RemoveAll(path); err != nil {
return xerrors.Errorf("remove '%s': %w", path, err)
}
return nil
return nil, xerrors.Errorf("no such datastore: %s", ns)
}
node/repo/importmgr/mgr.go
+5 -5
View File
@@ -42,7 +42,7 @@ type StoreMeta struct {
Labels map[string]string
}
func (m *Mgr) NewStore() (int64, *Store, error) {
func (m *Mgr) NewStore() (int, *Store, error) {
id := m.mds.Next()
st, err := m.mds.Get(id)
if err != nil {
@@ -60,7 +60,7 @@ func (m *Mgr) NewStore() (int64, *Store, error) {
return id, st, err
}
func (m *Mgr) AddLabel(id int64, key, value string) error { // source, file path, data CID..
func (m *Mgr) AddLabel(id int, key, value string) error { // source, file path, data CID..
meta, err := m.ds.Get(datastore.NewKey(fmt.Sprintf("%d", id)))
if err != nil {
return xerrors.Errorf("getting metadata form datastore: %w", err)
@@ -81,11 +81,11 @@ func (m *Mgr) AddLabel(id int64, key, value string) error { // source, file path
return m.ds.Put(datastore.NewKey(fmt.Sprintf("%d", id)), meta)
}
func (m *Mgr) List() []int64 {
func (m *Mgr) List() []int {
return m.mds.List()
}
func (m *Mgr) Info(id int64) (*StoreMeta, error) {
func (m *Mgr) Info(id int) (*StoreMeta, error) {
meta, err := m.ds.Get(datastore.NewKey(fmt.Sprintf("%d", id)))
if err != nil {
return nil, xerrors.Errorf("getting metadata form datastore: %w", err)
@@ -99,7 +99,7 @@ func (m *Mgr) Info(id int64) (*StoreMeta, error) {
return &sm, nil
}
func (m *Mgr) Remove(id int64) error {
func (m *Mgr) Remove(id int) error {
if err := m.mds.Delete(id); err != nil {
return xerrors.Errorf("removing import: %w", err)
}
node/repo/importmgr/multistore.go
+109 -45
View File
@@ -1,44 +1,47 @@
package importmgr
import (
"encoding/json"
"fmt"
"path"
"sort"
"sync"
"sync/atomic"
"github.com/hashicorp/go-multierror"
"go.uber.org/multierr"
"golang.org/x/xerrors"
"github.com/ipfs/go-datastore"
ktds "github.com/ipfs/go-datastore/keytransform"
"github.com/ipfs/go-datastore/query"
)
type dsProvider interface {
Datastore(namespace string) (datastore.Batching, error)
ListDatastores(namespace string) ([]int64, error)
DeleteDatastore(namespace string) error
}
type MultiStore struct {
provider dsProvider
namespace string
ds datastore.Batching
open map[int64]*Store
next int64
open map[int]*Store
next int
lk sync.RWMutex
}
func NewMultiDstore(provider dsProvider, namespace string) (*MultiStore, error) {
ids, err := provider.ListDatastores(namespace)
if err != nil {
return nil, xerrors.Errorf("listing datastores: %w", err)
var dsListKey = datastore.NewKey("/list")
var dsMultiKey = datastore.NewKey("/multi")
func NewMultiDstore(ds datastore.Batching) (*MultiStore, error) {
listBytes, err := ds.Get(dsListKey)
if xerrors.Is(err, datastore.ErrNotFound) {
listBytes, _ = json.Marshal([]int{})
} else if err != nil {
return nil, xerrors.Errorf("could not read multistore list: %w", err)
}
var ids []int
if err := json.Unmarshal(listBytes, &ids); err != nil {
return nil, xerrors.Errorf("could not unmarshal multistore list: %w", err)
}
mds := &MultiStore{
provider: provider,
namespace: namespace,
open: map[int64]*Store{},
ds: ds,
open: map[int]*Store{},
}
for _, i := range ids {
@@ -55,15 +58,33 @@ func NewMultiDstore(provider dsProvider, namespace string) (*MultiStore, error)
return mds, nil
}
func (mds *MultiStore) path(i int64) string {
return path.Join("/", mds.namespace, fmt.Sprintf("%d", i))
func (mds *MultiStore) Next() int {
mds.lk.Lock()
defer mds.lk.Unlock()
mds.next++
return mds.next
}
func (mds *MultiStore) Next() int64 {
return atomic.AddInt64(&mds.next, 1)
func (mds *MultiStore) updateStores() error {
stores := make([]int, 0, len(mds.open))
for k := range mds.open {
stores = append(stores, k)
}
sort.Ints(stores)
listBytes, err := json.Marshal(stores)
if err != nil {
return xerrors.Errorf("could not marshal list: %w", err)
}
err = mds.ds.Put(dsListKey, listBytes)
if err != nil {
return xerrors.Errorf("could not save stores list: %w", err)
}
return nil
}
func (mds *MultiStore) Get(i int64) (*Store, error) {
func (mds *MultiStore) Get(i int) (*Store, error) {
mds.lk.Lock()
defer mds.lk.Unlock()
@@ -72,40 +93,85 @@ func (mds *MultiStore) Get(i int64) (*Store, error) {
return store, nil
}
ds, err := mds.provider.Datastore(mds.path(i))
wds := ktds.Wrap(mds.ds, ktds.PrefixTransform{
Prefix: dsMultiKey.ChildString(fmt.Sprintf("%d", i)),
})
var err error
mds.open[i], err = openStore(wds)
if err != nil {
return nil, err
return nil, xerrors.Errorf("could not open new store: %w", err)
}
mds.open[i], err = openStore(ds)
return mds.open[i], err
err = mds.updateStores()
if err != nil {
return nil, xerrors.Errorf("updating stores: %w", err)
}
return mds.open[i], nil
}
func (mds *MultiStore) List() []int64 {
func (mds *MultiStore) List() []int {
mds.lk.RLock()
defer mds.lk.RUnlock()
out := make([]int64, 0, len(mds.open))
out := make([]int, 0, len(mds.open))
for i := range mds.open {
out = append(out, i)
}
sort.Ints(out)
return out
}
func (mds *MultiStore) Delete(i int64) error {
func (mds *MultiStore) Delete(i int) error {
mds.lk.Lock()
defer mds.lk.Unlock()
store, ok := mds.open[i]
if ok {
if err := store.Close(); err != nil {
return xerrors.Errorf("closing sub-datastore %d: %w", i, err)
}
delete(mds.open, i)
if !ok {
return nil
}
delete(mds.open, i)
err := store.Close()
if err != nil {
return xerrors.Errorf("closing store: %w", err)
}
return mds.provider.DeleteDatastore(mds.path(i))
err = mds.updateStores()
if err != nil {
return xerrors.Errorf("updating stores: %w", err)
}
qres, err := store.ds.Query(query.Query{KeysOnly: true})
if err != nil {
return xerrors.Errorf("query error: %w", err)
}
defer qres.Close() //nolint:errcheck
b, err := store.ds.Batch()
if err != nil {
return xerrors.Errorf("batch error: %w", err)
}
for r := range qres.Next() {
if r.Error != nil {
_ = b.Commit()
return xerrors.Errorf("iterator error: %w", err)
}
err := b.Delete(datastore.NewKey(r.Key))
if err != nil {
_ = b.Commit()
return xerrors.Errorf("adding to batch: %w", err)
}
}
err = b.Commit()
if err != nil {
return xerrors.Errorf("committing: %w", err)
}
return nil
}
func (mds *MultiStore) Close() error {
@@ -113,12 +179,10 @@ func (mds *MultiStore) Close() error {
defer mds.lk.Unlock()
var err error
for i, store := range mds.open {
cerr := store.Close()
if cerr != nil {
err = multierror.Append(err, xerrors.Errorf("closing sub-datastore %d: %w", i, cerr))
}
for _, s := range mds.open {
err = multierr.Append(err, s.Close())
}
mds.open = make(map[int]*Store)
return err
}
node/repo/interface.go
-2
View File
@@ -36,8 +36,6 @@ type LockedRepo interface {
// Returns datastore defined in this repo.
Datastore(namespace string) (datastore.Batching, error)
ListDatastores(namespace string) ([]int64, error)
DeleteDatastore(namespace string) error
// Returns config in this repo
Config() (interface{}, error)