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

Merge pull request #162 from filecoin-project/feat/retrieval

Browse Source 
Retrieval
This commit is contained in:
Łukasz Magiera
2019-08-31 00:29:00 +02:00
committed by GitHub
parent b0d0af603c 640ea80eef
commit 57d0f7c722
48 changed files with 1777 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
node/builder.go
+12
View File
@@ -35,8 +35,11 @@ import (
"github.com/filecoin-project/go-lotus/node/modules/testing"
"github.com/filecoin-project/go-lotus/node/repo"
"github.com/filecoin-project/go-lotus/paych"
"github.com/filecoin-project/go-lotus/retrieval"
"github.com/filecoin-project/go-lotus/retrieval/discovery"
"github.com/filecoin-project/go-lotus/storage"
"github.com/filecoin-project/go-lotus/storage/sector"
"github.com/filecoin-project/go-lotus/storage/sectorblocks"
)
// special is a type used to give keys to modules which
@@ -79,6 +82,7 @@ const (
// storage miner
HandleDealsKey
HandleRetrievalKey
RunSectorServiceKey
RegisterMinerKey
@@ -219,6 +223,10 @@ func Online() Option {
Override(RunBlockSyncKey, modules.RunBlockSync),
Override(HandleIncomingBlocksKey, modules.HandleIncomingBlocks),
Override(new(*discovery.Local), discovery.NewLocal),
Override(new(discovery.PeerResolver), modules.RetrievalResolver),
Override(new(*retrieval.Client), retrieval.NewClient),
Override(new(*deals.Client), deals.NewClient),
Override(RunDealClientKey, modules.RunDealClient),
@@ -232,11 +240,14 @@ func Online() Option {
ApplyIf(func(s *Settings) bool { return s.nodeType == nodeStorageMiner },
Override(new(*sectorbuilder.SectorBuilder), sectorbuilder.New),
Override(new(*sector.Store), sector.NewStore),
Override(new(*sectorblocks.SectorBlocks), sectorblocks.NewSectorBlocks),
Override(new(*storage.Miner), modules.StorageMiner),
Override(new(dtypes.StagingDAG), modules.StagingDAG),
Override(new(*retrieval.Miner), retrieval.NewMiner),
Override(new(*deals.Handler), deals.NewHandler),
Override(HandleRetrievalKey, modules.HandleRetrieval),
Override(HandleDealsKey, modules.HandleDeals),
Override(RunSectorServiceKey, modules.RunSectorService),
Override(RegisterMinerKey, modules.RegisterMiner),
@@ -300,6 +311,7 @@ func Repo(r repo.Repo) Option {
Override(new(dtypes.ChainBlockstore), modules.ChainBlockstore),
Override(new(dtypes.ClientFilestore), modules.ClientFstore),
Override(new(dtypes.ClientBlockstore), modules.ClientBlockstore),
Override(new(dtypes.ClientDAG), modules.ClientDAG),
Override(new(ci.PrivKey), pk),
node/client/import.go
-1
View File
@@ -1 +0,0 @@
package client
node/impl/full/client.go
+57 -5
View File
@@ -3,6 +3,12 @@ package full
import (
"context"
"errors"
"github.com/filecoin-project/go-lotus/build"
"github.com/filecoin-project/go-lotus/retrieval"
"github.com/filecoin-project/go-lotus/retrieval/discovery"
"github.com/ipfs/go-blockservice"
offline "github.com/ipfs/go-ipfs-exchange-offline"
"github.com/ipfs/go-merkledag"
"os"
"github.com/filecoin-project/go-lotus/api"
@@ -31,10 +37,13 @@ type ClientAPI struct {
WalletAPI
PaychAPI
DealClient *deals.Client
DealClient *deals.Client
RetDiscovery discovery.PeerResolver
Retrieval *retrieval.Client
LocalDAG dtypes.ClientDAG
Filestore dtypes.ClientFilestore `optional:"true"`
LocalDAG dtypes.ClientDAG
Blockstore dtypes.ClientBlockstore
Filestore dtypes.ClientFilestore `optional:"true"`
}
func (a *ClientAPI) ClientStartDeal(ctx context.Context, data cid.Cid, miner address.Address, price types.BigInt, blocksDuration uint64) (*cid.Cid, error) {
@@ -116,6 +125,34 @@ func (a *ClientAPI) ClientStartDeal(ctx context.Context, data cid.Cid, miner add
return &c, err
}
func (a *ClientAPI) 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)))
_, err := offExch.Get(ctx, root)
if err == ipld.ErrNotFound {
return false, nil
}
if err != nil {
return false, err
}
return true, nil
}
func (a *ClientAPI) ClientFindData(ctx context.Context, root cid.Cid) ([]api.QueryOffer, error) {
peers, err := a.RetDiscovery.GetPeers(root)
if err != nil {
return nil, err
}
out := make([]api.QueryOffer, len(peers))
for k, p := range peers {
out[k] = a.Retrieval.Query(ctx, p, root)
}
return out, nil
}
func (a *ClientAPI) ClientImport(ctx context.Context, path string) (cid.Cid, error) {
f, err := os.Open(path)
if err != nil {
@@ -134,14 +171,14 @@ func (a *ClientAPI) ClientImport(ctx context.Context, path string) (cid.Cid, err
bufferedDS := ipld.NewBufferedDAG(ctx, a.LocalDAG)
params := ihelper.DagBuilderParams{
Maxlinks: ihelper.DefaultLinksPerBlock,
Maxlinks: build.UnixfsLinksPerLevel,
RawLeaves: true,
CidBuilder: nil,
Dagserv: bufferedDS,
NoCopy: true,
}
db, err := params.New(chunker.DefaultSplitter(file))
db, err := params.New(chunker.NewSizeSplitter(file, int64(build.UnixfsChunkSize)))
if err != nil {
return cid.Undef, err
}
@@ -181,3 +218,18 @@ func (a *ClientAPI) ClientListImports(ctx context.Context) ([]api.Import, error)
})
}
}
func (a *ClientAPI) ClientRetrieve(ctx context.Context, order api.RetrievalOrder, path string) error {
outFile, err := os.OpenFile(path, os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0777)
if err != nil {
return err
}
err = a.Retrieval.RetrieveUnixfs(ctx, order.Root, order.Size, order.MinerPeerID, order.Miner, outFile)
if err != nil {
_ = outFile.Close()
return err
}
return outFile.Close()
}
node/impl/storminer.go
+21 -2
View File
@@ -7,10 +7,12 @@ import (
"math/rand"
"github.com/filecoin-project/go-lotus/api"
"github.com/filecoin-project/go-lotus/build"
"github.com/filecoin-project/go-lotus/chain/address"
"github.com/filecoin-project/go-lotus/lib/sectorbuilder"
"github.com/filecoin-project/go-lotus/storage"
"github.com/filecoin-project/go-lotus/storage/sector"
"github.com/filecoin-project/go-lotus/storage/sectorblocks"
)
type StorageMinerAPI struct {
@@ -19,6 +21,7 @@ type StorageMinerAPI struct {
SectorBuilderConfig *sectorbuilder.SectorBuilderConfig
SectorBuilder *sectorbuilder.SectorBuilder
Sectors *sector.Store
SectorBlocks *sectorblocks.SectorBlocks
Miner *storage.Miner
}
@@ -28,10 +31,10 @@ func (sm *StorageMinerAPI) ActorAddresses(context.Context) ([]address.Address, e
}
func (sm *StorageMinerAPI) StoreGarbageData(ctx context.Context) (uint64, error) {
size := uint64(1016) // this is the most data we can fit in a 1024 byte sector
size := sectorbuilder.UserBytesForSectorSize(build.SectorSize)
name := fmt.Sprintf("fake-file-%d", rand.Intn(100000000))
sectorId, err := sm.Sectors.AddPiece(name, size, io.LimitReader(rand.New(rand.NewSource(42)), 1016), 0)
sectorId, err := sm.Sectors.AddPiece(name, size, io.LimitReader(rand.New(rand.NewSource(42)), 1016))
if err != nil {
return 0, err
}
@@ -53,4 +56,20 @@ func (sm *StorageMinerAPI) SectorsStagedSeal(context.Context) error {
return sm.SectorBuilder.SealAllStagedSectors()
}
func (sm *StorageMinerAPI) SectorsRefs(context.Context) (map[string][]api.SealedRef, error) {
// json can't handle cids as map keys
out := map[string][]api.SealedRef{}
refs, err := sm.SectorBlocks.List()
if err != nil {
return nil, err
}
for k, v := range refs {
out[k.String()] = v
}
return out, nil
}
var _ api.StorageMiner = &StorageMinerAPI{}
node/modules/client.go
+4 -2
View File
@@ -36,9 +36,11 @@ func ClientFstore(r repo.LockedRepo) (dtypes.ClientFilestore, error) {
return filestore.NewFilestore(bs, fm), nil
}
func ClientDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, fstore dtypes.ClientFilestore, rt routing.Routing, h host.Host) dtypes.ClientDAG {
ibs := blockstore.NewIdStore((*filestore.Filestore)(fstore))
func ClientBlockstore(fstore dtypes.ClientFilestore) dtypes.ClientBlockstore {
return blockstore.NewIdStore((*filestore.Filestore)(fstore))
}
func ClientDAG(mctx helpers.MetricsCtx, lc fx.Lifecycle, ibs dtypes.ClientBlockstore, rt routing.Routing, h host.Host) dtypes.ClientDAG {
bitswapNetwork := network.NewFromIpfsHost(h, rt)
exch := bitswap.New(helpers.LifecycleCtx(mctx, lc), bitswapNetwork, ibs)
node/modules/dtypes/storage.go
+1
View File
@@ -21,6 +21,7 @@ type ChainExchange exchange.Interface
type ChainBlockService bserv.BlockService
type ClientFilestore *filestore.Filestore
type ClientBlockstore blockstore.Blockstore
type ClientDAG ipld.DAGService
type StagingDAG ipld.DAGService
node/modules/services.go
+6 -1
View File
@@ -2,7 +2,6 @@ package modules
import (
"context"
"github.com/filecoin-project/go-lotus/storage/sector"
"github.com/libp2p/go-libp2p-core/host"
inet "github.com/libp2p/go-libp2p-core/network"
@@ -14,6 +13,8 @@ import (
"github.com/filecoin-project/go-lotus/chain/sub"
"github.com/filecoin-project/go-lotus/node/hello"
"github.com/filecoin-project/go-lotus/node/modules/helpers"
"github.com/filecoin-project/go-lotus/retrieval/discovery"
"github.com/filecoin-project/go-lotus/storage/sector"
)
func RunHello(mctx helpers.MetricsCtx, lc fx.Lifecycle, h host.Host, svc *hello.Service) {
@@ -83,3 +84,7 @@ func RunSectorService(lc fx.Lifecycle, secst *sector.Store) {
},
})
}
func RetrievalResolver(l *discovery.Local) discovery.PeerResolver {
return discovery.Multi(l)
}
node/modules/storageminer.go
+15 -4
View File
@@ -2,29 +2,30 @@ package modules
import (
"context"
"github.com/filecoin-project/go-lotus/storage/sector"
"path/filepath"
"github.com/ipfs/go-bitswap"
"github.com/ipfs/go-bitswap/network"
"github.com/libp2p/go-libp2p-core/routing"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-datastore"
blockstore "github.com/ipfs/go-ipfs-blockstore"
"github.com/ipfs/go-merkledag"
"github.com/libp2p/go-libp2p-core/host"
"github.com/libp2p/go-libp2p-core/routing"
"github.com/mitchellh/go-homedir"
"go.uber.org/fx"
"github.com/filecoin-project/go-lotus/api"
"github.com/filecoin-project/go-lotus/build"
"github.com/filecoin-project/go-lotus/chain/address"
"github.com/filecoin-project/go-lotus/chain/deals"
"github.com/filecoin-project/go-lotus/lib/sectorbuilder"
"github.com/filecoin-project/go-lotus/node/modules/dtypes"
"github.com/filecoin-project/go-lotus/node/modules/helpers"
"github.com/filecoin-project/go-lotus/node/repo"
"github.com/filecoin-project/go-lotus/retrieval"
"github.com/filecoin-project/go-lotus/storage"
"github.com/filecoin-project/go-lotus/storage/sector"
)
func minerAddrFromDS(ds dtypes.MetadataDS) (address.Address, error) {
@@ -54,7 +55,7 @@ func SectorBuilderConfig(storagePath string) func(dtypes.MetadataDS) (*sectorbui
sb := &sectorbuilder.SectorBuilderConfig{
Miner: minerAddr,
SectorSize: 1024,
SectorSize: build.SectorSize,
MetadataDir: metadata,
SealedDir: sealed,
StagedDir: staging,
@@ -86,6 +87,16 @@ func StorageMiner(mctx helpers.MetricsCtx, lc fx.Lifecycle, api api.FullNode, h
return sm, nil
}
func HandleRetrieval(host host.Host, lc fx.Lifecycle, m *retrieval.Miner) {
lc.Append(fx.Hook{
OnStart: func(context.Context) error {
host.SetStreamHandler(retrieval.QueryProtocolID, m.HandleQueryStream)
host.SetStreamHandler(retrieval.ProtocolID, m.HandleDealStream)
return nil
},
})
}
func HandleDeals(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, h *deals.Handler) {
ctx := helpers.LifecycleCtx(mctx, lc)