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

Merge pull request #90 from filecoin-project/feat/fetch-reserve-space

Browse Source 
remote: Fetch storage reservation
This commit is contained in:
Łukasz Magiera 2020-08-05 23:37:00 +02:00 committed by GitHub
commit b7dca7fb4e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 80 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
fsutil/filesize_unix.go
View File

@ -1,6 +1,7 @@
package fsutil
import (
"os"
"syscall"
"golang.org/x/xerrors"
@ -14,12 +15,15 @@ type SizeInfo struct {
func FileSize(path string) (SizeInfo, error) {
var stat syscall.Stat_t
if err := syscall.Stat(path, &stat); err != nil {
if err == syscall.ENOENT {
return SizeInfo{}, os.ErrNotExist
}
return SizeInfo{}, xerrors.Errorf("stat: %w", err)
}
// NOTE: stat.Blocks is in 512B blocks, NOT in stat.Blksize
// See https://www.gnu.org/software/libc/manual/html_node/Attribute-Meanings.html
return SizeInfo{
int64(stat.Blocks) * 512,
int64(stat.Blocks) * 512, // NOTE: int64 cast is needed on osx
}, nil
}

8
stores/interface.go
View File

@ -9,15 +9,15 @@ import (
type PathType string
const (
PathStorage = "storage"
PathSealing = "sealing"
PathStorage PathType = "storage"
PathSealing PathType = "sealing"
)
type AcquireMode string
const (
AcquireMove = "move"
AcquireCopy = "copy"
AcquireMove AcquireMode = "move"
AcquireCopy AcquireMode = "copy"
)
type Store interface {

21
stores/local.go
View File

@ -50,7 +50,10 @@ type LocalStorage interface {
SetStorage(func(*StorageConfig)) error
Stat(path string) (fsutil.FsStat, error)
DiskUsage(path string) (int64, error) // returns real disk usage for a file/directory
// returns real disk usage for a file/directory
// os.ErrNotExit when file doesn't exist
DiskUsage(path string) (int64, error)
}
const MetaFile = "sectorstore.json"
@ -77,7 +80,7 @@ type path struct {
func (p *path) stat(ls LocalStorage) (fsutil.FsStat, error) {
stat, err := ls.Stat(p.local)
if err != nil {
return fsutil.FsStat{}, err
return fsutil.FsStat{}, xerrors.Errorf("stat %s: %w", p.local, err)
}
stat.Reserved = p.reserved
@ -88,7 +91,17 @@ func (p *path) stat(ls LocalStorage) (fsutil.FsStat, error) {
continue
}
used, err := ls.DiskUsage(p.sectorPath(id, fileType))
sp := p.sectorPath(id, fileType)
used, err := ls.DiskUsage(sp)
if err == os.ErrNotExist {
p, ferr := tempFetchDest(sp, false)
if ferr != nil {
return fsutil.FsStat{}, ferr
}
used, err = ls.DiskUsage(p)
}
if err != nil {
log.Errorf("getting disk usage of '%s': %+v", p.sectorPath(id, fileType), err)
continue
@ -279,7 +292,7 @@ func (st *Local) Reserve(ctx context.Context, sid abi.SectorID, spt abi.Register
stat, err := p.stat(st.localStorage)
if err != nil {
return nil, err
return nil, xerrors.Errorf("getting local storage stat: %w", err)
}
overhead := int64(overheadTab[fileType]) * int64(ssize) / FSOverheadDen

69
stores/remote.go
View File

@ -95,6 +95,33 @@ func (r *Remote) AcquireSector(ctx context.Context, s abi.SectorID, spt abi.Regi
return SectorPaths{}, SectorPaths{}, xerrors.Errorf("local acquire error: %w", err)
}
var toFetch SectorFileType
for _, fileType := range PathTypes {
if fileType&existing == 0 {
continue
}
if PathByType(paths, fileType) == "" {
toFetch |= fileType
}
}
apaths, ids, err := r.local.AcquireSector(ctx, s, spt, FTNone, toFetch, pathType, op)
if err != nil {
return SectorPaths{}, SectorPaths{}, xerrors.Errorf("allocate local sector for fetching: %w", err)
}
odt := FSOverheadSeal
if pathType == PathStorage {
odt = FsOverheadFinalized
}
releaseStorage, err := r.local.Reserve(ctx, s, spt, toFetch, ids, odt)
if err != nil {
return SectorPaths{}, SectorPaths{}, xerrors.Errorf("reserving storage space: %w", err)
}
defer releaseStorage()
for _, fileType := range PathTypes {
if fileType&existing == 0 {
continue
@ -104,15 +131,18 @@ func (r *Remote) AcquireSector(ctx context.Context, s abi.SectorID, spt abi.Regi
continue
}
ap, storageID, url, err := r.acquireFromRemote(ctx, s, spt, fileType, pathType, op)
dest := PathByType(apaths, fileType)
storageID := PathByType(ids, fileType)
url, err := r.acquireFromRemote(ctx, s, fileType, dest)
if err != nil {
return SectorPaths{}, SectorPaths{}, err
}
SetPathByType(&paths, fileType, ap)
SetPathByType(&stores, fileType, string(storageID))
SetPathByType(&paths, fileType, dest)
SetPathByType(&stores, fileType, storageID)
if err := r.index.StorageDeclareSector(ctx, storageID, s, fileType, op == AcquireMove); err != nil {
if err := r.index.StorageDeclareSector(ctx, ID(storageID), s, fileType, op == AcquireMove); err != nil {
log.Warnf("declaring sector %v in %s failed: %+v", s, storageID, err)
continue
}
@ -127,49 +157,44 @@ func (r *Remote) AcquireSector(ctx context.Context, s abi.SectorID, spt abi.Regi
return paths, stores, nil
}
func tempDest(spath string) (string, error) {
func tempFetchDest(spath string, create bool) (string, error) {
st, b := filepath.Split(spath)
tempdir := filepath.Join(st, FetchTempSubdir)
if err := os.MkdirAll(tempdir, 755); err != nil {
if create {
if err := os.MkdirAll(tempdir, 0755); err != nil {
return "", xerrors.Errorf("creating temp fetch dir: %w", err)
}
}
return filepath.Join(tempdir, b), nil
}
func (r *Remote) acquireFromRemote(ctx context.Context, s abi.SectorID, spt abi.RegisteredSealProof, fileType SectorFileType, pathType PathType, op AcquireMode) (string, ID, string, error) {
func (r *Remote) acquireFromRemote(ctx context.Context, s abi.SectorID, fileType SectorFileType, dest string) (string, error) {
si, err := r.index.StorageFindSector(ctx, s, fileType, false)
if err != nil {
return "", "", "", err
return "", err
}
if len(si) == 0 {
return "", "", "", xerrors.Errorf("failed to acquire sector %v from remote(%d): %w", s, fileType, storiface.ErrSectorNotFound)
return "", xerrors.Errorf("failed to acquire sector %v from remote(%d): %w", s, fileType, storiface.ErrSectorNotFound)
}
sort.Slice(si, func(i, j int) bool {
return si[i].Weight < si[j].Weight
})
apaths, ids, err := r.local.AcquireSector(ctx, s, spt, FTNone, fileType, pathType, op)
if err != nil {
return "", "", "", xerrors.Errorf("allocate local sector for fetching: %w", err)
}
dest := PathByType(apaths, fileType)
storageID := PathByType(ids, fileType)
var merr error
for _, info := range si {
// TODO: see what we have local, prefer that
for _, url := range info.URLs {
tempDest, err := tempDest(dest)
tempDest, err := tempFetchDest(dest, true)
if err != nil {
return "", "", "", err
return "", err
}
if err := os.RemoveAll(dest); err != nil {
return "", "", "", xerrors.Errorf("removing dest: %w", err)
return "", xerrors.Errorf("removing dest: %w", err)
}
err = r.fetch(ctx, url, tempDest)
@ -179,17 +204,17 @@ func (r *Remote) acquireFromRemote(ctx context.Context, s abi.SectorID, spt abi.
}
if err := move(tempDest, dest); err != nil {
return "", "", "", xerrors.Errorf("fetch move error (storage %s) %s -> %s: %w", info.ID, tempDest, dest, err)
return "", xerrors.Errorf("fetch move error (storage %s) %s -> %s: %w", info.ID, tempDest, dest, err)
}
if merr != nil {
log.Warnw("acquireFromRemote encountered errors when fetching sector from remote", "errors", merr)
}
return dest, ID(storageID), url, nil
return url, nil
}
}
return "", "", "", xerrors.Errorf("failed to acquire sector %v from remote (tried %v): %w", s, si, merr)
return "", xerrors.Errorf("failed to acquire sector %v from remote (tried %v): %w", s, si, merr)
}
func (r *Remote) fetch(ctx context.Context, url, outname string) error {