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

Client Import manager

Browse Source
This commit is contained in:
Łukasz Magiera 2020-07-07 01:39:30 +02:00
parent 92e4507cf7
commit 8942967223
12 changed files with 433 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
node/builder.go
View File

@ -232,7 +232,6 @@ func Online() Option {
Override(new(dtypes.ChainGCBlockstore), modules.ChainGCBlockstore),
Override(new(dtypes.ChainExchange), modules.ChainExchange),
Override(new(dtypes.ChainBlockService), modules.ChainBlockservice),
Override(new(dtypes.ClientDAG), testing.MemoryClientDag),
// Filecoin services
Override(new(*chain.Syncer), modules.NewSyncer),
@ -436,9 +435,9 @@ func Repo(r repo.Repo) Option {
Override(new(dtypes.MetadataDS), modules.Datastore),
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
Override(new(dtypes.ClientMultiDstore), modules.ClientMultiDatastore),
Override(new(dtypes.ClientFilestore), modules.ClientFstore),
Override(new(dtypes.ClientBlockstore), modules.ClientBlockstore),
Override(new(dtypes.ClientDAG), modules.ClientDAG),
Override(new(ci.PrivKey), lp2p.PrivKey),
Override(new(ci.PubKey), ci.PrivKey.GetPublic),

84
node/impl/client/client.go
View File

@ -46,7 +46,7 @@ import (
"github.com/filecoin-project/lotus/markets/utils"
"github.com/filecoin-project/lotus/node/impl/full"
"github.com/filecoin-project/lotus/node/impl/paych"
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/repo/importmgr"
)
const dealStartBuffer abi.ChainEpoch = 10000 // TODO: allow setting
@ -64,9 +64,7 @@ type API struct {
Retrieval rm.RetrievalClient
Chain *store.ChainStore
LocalDAG dtypes.ClientDAG
Blockstore dtypes.ClientBlockstore
Filestore dtypes.ClientFilestore `optional:"true"`
Imports *importmgr.Mgr
}
func calcDealExpiration(minDuration uint64, md *miner.DeadlineInfo, startEpoch abi.ChainEpoch) abi.ChainEpoch {
@ -195,7 +193,7 @@ func (a *API) ClientGetDealInfo(ctx context.Context, d cid.Cid) (*api.DealInfo,
func (a *API) ClientHasLocal(ctx context.Context, root cid.Cid) (bool, error) {
// TODO: check if we have the ENTIRE dag
offExch := merkledag.NewDAGService(blockservice.New(a.Blockstore, offline.Exchange(a.Blockstore)))
offExch := merkledag.NewDAGService(blockservice.New(a.Imports.Bs, offline.Exchange(a.Imports.Bs)))
_, err := offExch.Get(ctx, root)
if err == ipld.ErrNotFound {
return false, nil
@ -260,9 +258,15 @@ func (a *API) makeRetrievalQuery(ctx context.Context, rp rm.RetrievalPeer, paylo
}
func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (cid.Cid, error) {
id, st, err := a.Imports.NewStore()
if err != nil {
return cid.Cid{}, err
}
if err := a.Imports.AddLabel(id, "source", "import"); err != nil {
return cid.Cid{}, err
}
bufferedDS := ipld.NewBufferedDAG(ctx, a.LocalDAG)
nd, err := a.clientImport(ref, bufferedDS)
nd, err := a.clientImport(ctx, ref, st)
if err != nil {
return cid.Undef, err
@ -274,12 +278,20 @@ func (a *API) ClientImport(ctx context.Context, ref api.FileRef) (cid.Cid, error
func (a *API) ClientImportLocal(ctx context.Context, f io.Reader) (cid.Cid, error) {
file := files.NewReaderFile(f)
bufferedDS := ipld.NewBufferedDAG(ctx, a.LocalDAG)
id, st, err := a.Imports.NewStore()
if err != nil {
return cid.Cid{}, err
}
if err := a.Imports.AddLabel(id, "source", "import-local"); err != nil {
return cid.Cid{}, err
}
bufferedDS := ipld.NewBufferedDAG(ctx, st.DAG)
params := ihelper.DagBuilderParams{
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: nil,
CidBuilder: cid.V1Builder{},
Dagserv: bufferedDS,
}
@ -348,13 +360,21 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
return err
}
order.MinerPeerID = peer.ID(mi.PeerId)
order.MinerPeerID = mi.PeerId
}
if order.Size == 0 {
return xerrors.Errorf("cannot make retrieval deal for zero bytes")
}
id, st, err := a.Imports.NewStore()
if err != nil {
return err
}
if err := a.Imports.AddLabel(id, "source", "retrieval"); err != nil {
return err
}
retrievalResult := make(chan error, 1)
unsubscribe := a.Retrieval.SubscribeToEvents(func(event rm.ClientEvent, state rm.ClientDealState) {
@ -392,14 +412,14 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
ppb := types.BigDiv(order.Total, types.NewInt(order.Size))
_, err := a.Retrieval.Retrieve(
_, err = a.Retrieval.Retrieve(
ctx,
order.Root,
rm.NewParamsV0(ppb, order.PaymentInterval, order.PaymentIntervalIncrease),
order.Total,
order.MinerPeerID,
order.Client,
order.Miner)
order.Miner) // TODO: pass the store here somehow
if err != nil {
return xerrors.Errorf("Retrieve failed: %w", err)
}
@ -424,18 +444,18 @@ func (a *API) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, ref
if err != nil {
return err
}
err = car.WriteCar(ctx, a.LocalDAG, []cid.Cid{order.Root}, f)
err = car.WriteCar(ctx, st.DAG, []cid.Cid{order.Root}, f)
if err != nil {
return err
}
return f.Close()
}
nd, err := a.LocalDAG.Get(ctx, order.Root)
nd, err := st.DAG.Get(ctx, order.Root)
if err != nil {
return xerrors.Errorf("ClientRetrieve: %w", err)
}
file, err := unixfile.NewUnixfsFile(ctx, a.LocalDAG, nd)
file, err := unixfile.NewUnixfsFile(ctx, st.DAG, nd)
if err != nil {
return xerrors.Errorf("ClientRetrieve: %w", err)
}
@ -485,14 +505,22 @@ func (a *API) ClientCalcCommP(ctx context.Context, inpath string, miner address.
}
func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath string) error {
id, st, err := a.Imports.NewStore()
if err != nil {
return err
}
if err := a.Imports.AddLabel(id, "source", "gen-car"); err != nil {
return err
}
bufferedDS := ipld.NewBufferedDAG(ctx, a.LocalDAG)
c, err := a.clientImport(ref, bufferedDS)
bufferedDS := ipld.NewBufferedDAG(ctx, st.DAG)
c, err := a.clientImport(ctx, ref, st)
if err != nil {
return err
}
// TODO: does that defer mean to remove the whole blockstore?
defer bufferedDS.Remove(ctx, c) //nolint:errcheck
ssb := builder.NewSelectorSpecBuilder(basicnode.Style.Any)
@ -505,7 +533,7 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
return err
}
sc := car.NewSelectiveCar(ctx, a.Blockstore, []car.Dag{{Root: c, Selector: allSelector}})
sc := car.NewSelectiveCar(ctx, st.Bstore, []car.Dag{{Root: c, Selector: allSelector}})
if err = sc.Write(f); err != nil {
return err
}
@ -513,7 +541,7 @@ func (a *API) ClientGenCar(ctx context.Context, ref api.FileRef, outputPath stri
return f.Close()
}
func (a *API) clientImport(ref api.FileRef, bufferedDS *ipld.BufferedDAG) (cid.Cid, error) {
func (a *API) clientImport(ctx context.Context, ref api.FileRef, store *importmgr.Store) (cid.Cid, error) {
f, err := os.Open(ref.Path)
if err != nil {
return cid.Undef, err
@ -530,13 +558,13 @@ func (a *API) clientImport(ref api.FileRef, bufferedDS *ipld.BufferedDAG) (cid.C
}
if ref.IsCAR {
var store car.Store
if a.Filestore == nil {
store = a.Blockstore
var st car.Store
if store.Fstore == nil {
st = store.Bstore
} else {
store = (*filestore.Filestore)(a.Filestore)
st = store.Fstore
}
result, err := car.LoadCar(store, file)
result, err := car.LoadCar(st, file)
if err != nil {
return cid.Undef, err
}
@ -548,11 +576,13 @@ func (a *API) clientImport(ref api.FileRef, bufferedDS *ipld.BufferedDAG) (cid.C
return result.Roots[0], nil
}
bufDs := ipld.NewBufferedDAG(ctx, store.DAG)
params := ihelper.DagBuilderParams{
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: nil,
Dagserv: bufferedDS,
CidBuilder: cid.V1Builder{},
Dagserv: bufDs,
NoCopy: true,
}
@ -565,7 +595,7 @@ func (a *API) clientImport(ref api.FileRef, bufferedDS *ipld.BufferedDAG) (cid.C
return cid.Undef, err
}
if err := bufferedDS.Commit(); err != nil {
if err := bufDs.Commit(); err != nil {
return cid.Undef, err
}

16
node/modules/client.go
View File

@ -34,9 +34,25 @@ import (
"github.com/filecoin-project/lotus/node/modules/dtypes"
"github.com/filecoin-project/lotus/node/modules/helpers"
"github.com/filecoin-project/lotus/node/repo"
"github.com/filecoin-project/lotus/node/repo/importmgr"
"github.com/filecoin-project/lotus/paychmgr"
)
func ClientMultiDatastore(lc fx.Lifecycle, r repo.LockedRepo) (dtypes.ClientMultiDstore, error) {
mds, err := importmgr.NewMultiDstore(r, "/client")
if err != nil {
return nil, err
}
lc.Append(fx.Hook{
OnStop: func(ctx context.Context) error {
return mds.Close()
},
})
return mds, nil
}
func ClientFstore(r repo.LockedRepo) (dtypes.ClientFilestore, error) {
clientds, err := r.Datastore("/client")
if err != nil {

6
node/modules/dtypes/storage.go
View File

@ -1,7 +1,6 @@
package dtypes
import (
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
bserv "github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-filestore"
@ -10,9 +9,12 @@ import (
exchange "github.com/ipfs/go-ipfs-exchange-interface"
format "github.com/ipfs/go-ipld-format"
"github.com/filecoin-project/go-fil-markets/storagemarket/impl/requestvalidation"
datatransfer "github.com/filecoin-project/go-data-transfer"
"github.com/filecoin-project/go-fil-markets/piecestore"
"github.com/filecoin-project/go-statestore"
"github.com/filecoin-project/lotus/node/repo/importmgr"
)
// MetadataDS stores metadata
@ -26,9 +28,9 @@ type ChainGCBlockstore blockstore.GCBlockstore
type ChainExchange exchange.Interface
type ChainBlockService bserv.BlockService
type ClientMultiDstore *importmgr.MultiStore
type ClientFilestore *filestore.Filestore
type ClientBlockstore blockstore.Blockstore
type ClientDAG format.DAGService
type ClientDealStore *statestore.StateStore
type ClientRequestValidator *requestvalidation.UnifiedRequestValidator
type ClientDatastore datastore.Batching

39
node/modules/testing/storage.go
View File

@ -1,39 +0,0 @@
package testing
import (
"context"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
dsync "github.com/ipfs/go-datastore/sync"
blockstore "github.com/ipfs/go-ipfs-blockstore"
offline "github.com/ipfs/go-ipfs-exchange-offline"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-merkledag"
"go.uber.org/fx"
)
func MapBlockstore() blockstore.Blockstore {
// TODO: proper datastore
bds := dsync.MutexWrap(datastore.NewMapDatastore())
bs := blockstore.NewBlockstore(bds)
return blockstore.NewIdStore(bs)
}
func MapDatastore() datastore.Batching {
return dsync.MutexWrap(datastore.NewMapDatastore())
}
func MemoryClientDag(lc fx.Lifecycle) ipld.DAGService {
ibs := blockstore.NewBlockstore(datastore.NewMapDatastore())
bsvc := blockservice.New(ibs, offline.Exchange(ibs))
dag := merkledag.NewDAGService(bsvc)
lc.Append(fx.Hook{
OnStop: func(_ context.Context) error {
return bsvc.Close()
},
})
return dag
}

23
node/repo/fsrepo_ds.go
View File

@ -178,6 +178,29 @@ func (fsr *fsLockedRepo) Datastore(ns string) (datastore.Batching, error) {
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()

1
node/repo/importmgr/mbstore.go Normal file
View File

@ -0,0 +1 @@
package importmgr

74
node/repo/importmgr/mgr.go Normal file
View File

@ -0,0 +1,74 @@
package importmgr
import (
"encoding/json"
"fmt"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"golang.org/x/xerrors"
)
type Mgr struct {
mds *MultiStore
Bs blockstore.Blockstore
ds datastore.Batching
}
func New(mds *MultiStore, ds datastore.Batching) *Mgr {
return &Mgr{
mds: mds,
bs: &multiReadBs{
mds: mds,
},
ds: namespace.Wrap(ds, datastore.NewKey("/stores")),
}
}
type storeMeta struct {
Labels map[string]string
}
func (m *Mgr) NewStore() (int64, *Store, error) {
id := m.mds.Next()
st, err := m.mds.Get(id)
if err != nil {
return 0, nil, err
}
meta, err := json.Marshal(&storeMeta{Labels: map[string]string{
"source": "unknown",
}})
if err != nil {
return 0, nil, xerrors.Errorf("marshaling empty store metadata: %w", err)
}
err = m.ds.Put(datastore.NewKey(fmt.Sprintf("%d", id)), meta)
return id, st, err
}
func (m *Mgr) AddLabel(id int64, 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)
}
var sm storeMeta
if err := json.Unmarshal(meta, &sm); err != nil {
return xerrors.Errorf("unmarshaling store meta: %w", err)
}
sm.Labels[key] = value
meta, err = json.Marshal(&storeMeta{})
if err != nil {
return xerrors.Errorf("marshaling store meta: %w", err)
}
return m.ds.Put(datastore.NewKey(fmt.Sprintf("%d", id)), meta)
}
// m.List
// m.Info
// m.Delete

198
node/repo/importmgr/multistore.go Normal file
View File

@ -0,0 +1,198 @@
package importmgr
import (
"context"
"fmt"
"path"
"sync"
"sync/atomic"
"github.com/hashicorp/go-multierror"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"golang.org/x/xerrors"
"github.com/ipfs/go-datastore"
)
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
open map[int64]*Store
next int64
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)
}
mds := &MultiStore{
provider: provider,
namespace: namespace,
}
for _, i := range ids {
if i > mds.next {
mds.next = i
}
_, err := mds.Get(i)
if err != nil {
return nil, xerrors.Errorf("open store %d: %w", i, err)
}
}
return mds, nil
}
func (mds *MultiStore) path(i int64) string {
return path.Join("/", mds.namespace, fmt.Sprintf("%d", i))
}
func (mds *MultiStore) Next() int64 {
return atomic.AddInt64(&mds.next, 1)
}
func (mds *MultiStore) Get(i int64) (*Store, error) {
mds.lk.Lock()
defer mds.lk.Unlock()
store, ok := mds.open[i]
if ok {
return store, nil
}
ds, err := mds.provider.Datastore(mds.path(i))
if err != nil {
return nil, err
}
mds.open[i], err = openStore(ds)
return mds.open[i], err
}
func (mds *MultiStore) Delete(i int64) 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)
}
return mds.provider.DeleteDatastore(mds.path(i))
}
func (mds *MultiStore) Close() error {
mds.lk.Lock()
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))
}
}
return err
}
type multiReadBs struct {
// TODO: some caching
mds *MultiStore
}
func (m *multiReadBs) Has(cid cid.Cid) (bool, error) {
m.mds.lk.RLock()
defer m.mds.lk.RUnlock()
var merr error
for i, store := range m.mds.open {
has, err := store.Bstore.Has(cid)
if err != nil {
merr = multierror.Append(merr, xerrors.Errorf("has (ds %d): %w", i, err))
continue
}
if !has {
continue
}
return true, nil
}
return false, merr
}
func (m *multiReadBs) Get(cid cid.Cid) (blocks.Block, error) {
m.mds.lk.RLock()
defer m.mds.lk.RUnlock()
var merr error
for i, store := range m.mds.open {
has, err := store.Bstore.Has(cid)
if err != nil {
merr = multierror.Append(merr, xerrors.Errorf("has (ds %d): %w", i, err))
continue
}
if !has {
continue
}
val, err := store.Bstore.Get(cid)
if err != nil {
merr = multierror.Append(merr, xerrors.Errorf("get (ds %d): %w", i, err))
continue
}
return val, nil
}
if merr == nil {
return nil, datastore.ErrNotFound
}
return nil, merr
}
func (m *multiReadBs) DeleteBlock(cid cid.Cid) error {
return xerrors.Errorf("operation not supported")
}
func (m *multiReadBs) GetSize(cid cid.Cid) (int, error) {
return 0, xerrors.Errorf("operation not supported")
}
func (m *multiReadBs) Put(block blocks.Block) error {
return xerrors.Errorf("operation not supported")
}
func (m *multiReadBs) PutMany(blocks []blocks.Block) error {
return xerrors.Errorf("operation not supported")
}
func (m *multiReadBs) AllKeysChan(ctx context.Context) (<-chan cid.Cid, error) {
return nil, xerrors.Errorf("operation not supported")
}
func (m *multiReadBs) HashOnRead(enabled bool) {
return
}
var _ blockstore.Blockstore = &multiReadBs{}

54
node/repo/importmgr/store.go Normal file
View File

@ -0,0 +1,54 @@
package importmgr
import (
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"github.com/ipfs/go-filestore"
blockstore "github.com/ipfs/go-ipfs-blockstore"
offline "github.com/ipfs/go-ipfs-exchange-offline"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-merkledag"
)
type Store struct {
ds datastore.Batching
fm *filestore.FileManager
Fstore *filestore.Filestore
Bstore blockstore.Blockstore
bsvc blockservice.BlockService
DAG ipld.DAGService
}
func openStore(ds datastore.Batching) (*Store, error) {
blocks := namespace.Wrap(ds, datastore.NewKey("blocks"))
bs := blockstore.NewBlockstore(blocks)
fm := filestore.NewFileManager(ds, "/")
fm.AllowFiles = true
fstore := filestore.NewFilestore(bs, fm)
ibs := blockstore.NewIdStore(fstore)
bsvc := blockservice.New(ibs, offline.Exchange(ibs))
dag := merkledag.NewDAGService(bsvc)
return &Store{
ds: ds,
fm: fm,
Fstore: fstore,
Bstore: ibs,
bsvc: bsvc,
DAG: dag,
}, nil
}
func (s *Store) Close() error {
return s.bsvc.Close()
}

1
node/repo/interface.go
View File

@ -35,6 +35,7 @@ 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

4
node/repo/memrepo.go
View File

@ -233,6 +233,10 @@ func (lmem *lockedMemRepo) Datastore(ns string) (datastore.Batching, error) {
return namespace.Wrap(lmem.mem.datastore, datastore.NewKey(ns)), nil
}
func (lmem *lockedMemRepo) ListDatastores(ns string) ([]int64, error) {
return nil, nil
}
func (lmem *lockedMemRepo) DeleteDatastore(ns string) error {
/** poof **/
return nil