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

Merge remote-tracking branch 'origin/next' into feat/cid-builder

Browse Source
This commit is contained in:
Łukasz Magiera
2020-07-28 16:06:06 +02:00
parent bb54dc97b6 8c95821e9c
commit 09b90773d8
52 changed files with 851 additions and 672 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/repo/importmgr/mbstore.go
-96
View File
@@ -1,96 +0,0 @@
package importmgr
import (
"context"
"github.com/hashicorp/go-multierror"
"golang.org/x/xerrors"
blocks "github.com/ipfs/go-block-format"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/lotus/lib/blockstore"
)
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, blockstore.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{}
node/repo/importmgr/mgr.go
+10 -12
View File
@@ -8,12 +8,12 @@ import (
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"github.com/filecoin-project/lotus/lib/blockstore"
"github.com/filecoin-project/go-multistore"
)
type Mgr struct {
mds *MultiStore
mds *multistore.MultiStore
ds datastore.Batching
Blockstore blockstore.Blockstore
@@ -28,12 +28,10 @@ const (
LMTime = "mtime" // File modification timestamp
)
func New(mds *MultiStore, ds datastore.Batching) *Mgr {
func New(mds *multistore.MultiStore, ds datastore.Batching) *Mgr {
return &Mgr{
mds: mds,
Blockstore: &multiReadBs{
mds: mds,
},
mds: mds,
Blockstore: mds.MultiReadBlockstore(),
ds: datastore.NewLogDatastore(namespace.Wrap(ds, datastore.NewKey("/stores")), "storess"),
}
@@ -43,7 +41,7 @@ type StoreMeta struct {
Labels map[string]string
}
func (m *Mgr) NewStore() (int, *Store, error) {
func (m *Mgr) NewStore() (multistore.StoreID, *multistore.Store, error) {
id := m.mds.Next()
st, err := m.mds.Get(id)
if err != nil {
@@ -61,7 +59,7 @@ func (m *Mgr) NewStore() (int, *Store, error) {
return id, st, err
}
func (m *Mgr) AddLabel(id int, key, value string) error { // source, file path, data CID..
func (m *Mgr) AddLabel(id multistore.StoreID, 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)
@@ -82,11 +80,11 @@ func (m *Mgr) AddLabel(id int, key, value string) error { // source, file path,
return m.ds.Put(datastore.NewKey(fmt.Sprintf("%d", id)), meta)
}
func (m *Mgr) List() []int {
func (m *Mgr) List() []multistore.StoreID {
return m.mds.List()
}
func (m *Mgr) Info(id int) (*StoreMeta, error) {
func (m *Mgr) Info(id multistore.StoreID) (*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)
@@ -100,7 +98,7 @@ func (m *Mgr) Info(id int) (*StoreMeta, error) {
return &sm, nil
}
func (m *Mgr) Remove(id int) error {
func (m *Mgr) Remove(id multistore.StoreID) error {
if err := m.mds.Delete(id); err != nil {
return xerrors.Errorf("removing import: %w", err)
}
node/repo/importmgr/multistore.go
-188
View File
@@ -1,188 +0,0 @@
package importmgr
import (
"encoding/json"
"fmt"
"sort"
"sync"
"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 MultiStore struct {
ds datastore.Batching
open map[int]*Store
next int
lk sync.RWMutex
}
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{
ds: ds,
open: map[int]*Store{},
}
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) Next() int {
mds.lk.Lock()
defer mds.lk.Unlock()
mds.next++
return mds.next
}
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 int) (*Store, error) {
mds.lk.Lock()
defer mds.lk.Unlock()
store, ok := mds.open[i]
if ok {
return store, nil
}
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, xerrors.Errorf("could not open new store: %w", err)
}
err = mds.updateStores()
if err != nil {
return nil, xerrors.Errorf("updating stores: %w", err)
}
return mds.open[i], nil
}
func (mds *MultiStore) List() []int {
mds.lk.RLock()
defer mds.lk.RUnlock()
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 int) error {
mds.lk.Lock()
defer mds.lk.Unlock()
store, ok := mds.open[i]
if !ok {
return nil
}
delete(mds.open, i)
err := store.Close()
if err != nil {
return xerrors.Errorf("closing store: %w", err)
}
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 {
mds.lk.Lock()
defer mds.lk.Unlock()
var err error
for _, s := range mds.open {
err = multierr.Append(err, s.Close())
}
mds.open = make(map[int]*Store)
return err
}
node/repo/importmgr/store.go
-55
View File
@@ -1,55 +0,0 @@
package importmgr
import (
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
"github.com/ipfs/go-filestore"
offline "github.com/ipfs/go-ipfs-exchange-offline"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-merkledag"
"github.com/filecoin-project/lotus/lib/blockstore"
)
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.WrapIDStore(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()
}