lotus/storage/sectorstorage/stores/index.go

323 lines
7.1 KiB
Go
Raw Normal View History

2020-03-13 11:59:19 +00:00
package stores
import (
"context"
"net/url"
gopath "path"
2020-03-19 15:10:19 +00:00
"sort"
2020-03-13 11:59:19 +00:00
"sync"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-sectorbuilder"
"github.com/filecoin-project/specs-actors/actors/abi"
2020-03-19 15:10:19 +00:00
"github.com/filecoin-project/specs-actors/actors/abi/big"
2020-03-13 11:59:19 +00:00
2020-03-23 11:40:02 +00:00
"github.com/filecoin-project/lotus/storage/sectorstorage/sectorutil"
2020-03-13 11:59:19 +00:00
)
// ID identifies sector storage by UUID. One sector storage should map to one
// filesystem, local or networked / shared by multiple machines
type ID string
type StorageInfo struct {
2020-03-19 15:10:19 +00:00
ID ID
URLs []string // TODO: Support non-http transports
Weight uint64
2020-03-13 11:59:19 +00:00
CanSeal bool
CanStore bool
}
type SectorIndex interface { // part of storage-miner api
2020-03-19 15:10:19 +00:00
StorageAttach(context.Context, StorageInfo, FsStat) error
StorageInfo(context.Context, ID) (StorageInfo, error)
// TODO: StorageUpdateStats(FsStat)
2020-03-13 11:59:19 +00:00
StorageDeclareSector(ctx context.Context, storageId ID, s abi.SectorID, ft sectorbuilder.SectorFileType) error
StorageDropSector(ctx context.Context, storageId ID, s abi.SectorID, ft sectorbuilder.SectorFileType) error
StorageFindSector(ctx context.Context, sector abi.SectorID, ft sectorbuilder.SectorFileType, allowFetch bool) ([]StorageInfo, error)
2020-03-19 15:10:19 +00:00
StorageBestAlloc(ctx context.Context, allocate sectorbuilder.SectorFileType, sealing bool) ([]StorageInfo, error)
2020-03-13 11:59:19 +00:00
}
2020-03-16 17:50:07 +00:00
type Decl struct {
2020-03-13 11:59:19 +00:00
abi.SectorID
sectorbuilder.SectorFileType
}
2020-03-19 15:10:19 +00:00
type storageEntry struct {
info *StorageInfo
fsi FsStat
}
2020-03-13 11:59:19 +00:00
type Index struct {
2020-03-19 15:10:19 +00:00
lk sync.RWMutex
2020-03-13 11:59:19 +00:00
2020-03-16 17:50:07 +00:00
sectors map[Decl][]ID
2020-03-19 15:10:19 +00:00
stores map[ID]*storageEntry
2020-03-13 11:59:19 +00:00
}
func NewIndex() *Index {
return &Index{
2020-03-16 17:50:07 +00:00
sectors: map[Decl][]ID{},
2020-03-19 15:10:19 +00:00
stores: map[ID]*storageEntry{},
2020-03-13 11:59:19 +00:00
}
}
2020-03-16 17:50:07 +00:00
func (i *Index) StorageList(ctx context.Context) (map[ID][]Decl, error) {
byID := map[ID]map[abi.SectorID]sectorbuilder.SectorFileType{}
for id := range i.stores {
byID[id] = map[abi.SectorID]sectorbuilder.SectorFileType{}
}
for decl, ids := range i.sectors {
for _, id := range ids {
byID[id][decl.SectorID] |= decl.SectorFileType
}
}
out := map[ID][]Decl{}
for id, m := range byID {
2020-03-18 04:40:25 +00:00
out[id] = []Decl{}
2020-03-16 17:50:07 +00:00
for sectorID, fileType := range m {
out[id] = append(out[id], Decl{
SectorID: sectorID,
SectorFileType: fileType,
})
}
}
return out, nil
}
2020-03-19 15:10:19 +00:00
func (i *Index) StorageAttach(ctx context.Context, si StorageInfo, st FsStat) error {
2020-03-13 11:59:19 +00:00
i.lk.Lock()
defer i.lk.Unlock()
2020-03-18 04:40:25 +00:00
log.Infof("New sector storage: %s", si.ID)
2020-03-13 11:59:19 +00:00
if _, ok := i.stores[si.ID]; ok {
for _, u := range si.URLs {
if _, err := url.Parse(u); err != nil {
return xerrors.Errorf("failed to parse url %s: %w", si.URLs, err)
}
}
uloop:
for _, u := range si.URLs {
for _, l := range i.stores[si.ID].info.URLs {
if u == l {
continue uloop
}
}
i.stores[si.ID].info.URLs = append(i.stores[si.ID].info.URLs, u)
}
2020-03-13 11:59:19 +00:00
return nil
}
2020-03-19 15:10:19 +00:00
i.stores[si.ID] = &storageEntry{
info: &si,
fsi: st,
}
2020-03-13 11:59:19 +00:00
return nil
}
func (i *Index) StorageDeclareSector(ctx context.Context, storageId ID, s abi.SectorID, ft sectorbuilder.SectorFileType) error {
i.lk.Lock()
defer i.lk.Unlock()
for _, fileType := range pathTypes {
if fileType&ft == 0 {
continue
}
2020-03-16 17:50:07 +00:00
d := Decl{s, fileType}
2020-03-13 11:59:19 +00:00
for _, sid := range i.sectors[d] {
if sid == storageId {
log.Warnf("sector %v redeclared in %s", s, storageId)
return nil
}
2020-03-13 11:59:19 +00:00
}
i.sectors[d] = append(i.sectors[d], storageId)
2020-03-13 11:59:19 +00:00
}
return nil
}
func (i *Index) StorageDropSector(ctx context.Context, storageId ID, s abi.SectorID, ft sectorbuilder.SectorFileType) error {
i.lk.Lock()
defer i.lk.Unlock()
for _, fileType := range pathTypes {
if fileType&ft == 0 {
continue
}
d := Decl{s, fileType}
if len(i.sectors[d]) == 0 {
return nil
}
rewritten := make([]ID, 0, len(i.sectors[d])-1)
for _, sid := range i.sectors[d] {
if sid == storageId {
continue
}
rewritten = append(rewritten, sid)
}
if len(rewritten) == 0 {
delete(i.sectors, d)
return nil
}
i.sectors[d] = rewritten
}
return nil
}
func (i *Index) StorageFindSector(ctx context.Context, s abi.SectorID, ft sectorbuilder.SectorFileType, allowFetch bool) ([]StorageInfo, error) {
2020-03-19 15:10:19 +00:00
i.lk.RLock()
defer i.lk.RUnlock()
2020-03-13 11:59:19 +00:00
storageIDs := map[ID]uint64{}
for _, pathType := range pathTypes {
if ft&pathType == 0 {
continue
}
for _, id := range i.sectors[Decl{s, pathType}] {
storageIDs[id]++
}
}
2020-03-13 11:59:19 +00:00
out := make([]StorageInfo, 0, len(storageIDs))
for id, n := range storageIDs {
2020-03-13 11:59:19 +00:00
st, ok := i.stores[id]
if !ok {
log.Warnf("storage %s is not present in sector index (referenced by sector %v)", id, s)
continue
}
2020-03-19 15:10:19 +00:00
urls := make([]string, len(st.info.URLs))
for k, u := range st.info.URLs {
2020-03-13 11:59:19 +00:00
rl, err := url.Parse(u)
if err != nil {
return nil, xerrors.Errorf("failed to parse url: %w", err)
}
rl.Path = gopath.Join(rl.Path, ft.String(), sectorutil.SectorName(s))
urls[k] = rl.String()
}
out = append(out, StorageInfo{
2020-03-13 11:59:19 +00:00
ID: id,
URLs: urls,
Weight: st.info.Weight * n, // storage with more sector types is better
2020-03-19 15:10:19 +00:00
CanSeal: st.info.CanSeal,
CanStore: st.info.CanStore,
})
2020-03-13 11:59:19 +00:00
}
if allowFetch {
for id, st := range i.stores {
if _, ok := storageIDs[id]; ok {
continue
}
urls := make([]string, len(st.info.URLs))
for k, u := range st.info.URLs {
rl, err := url.Parse(u)
if err != nil {
return nil, xerrors.Errorf("failed to parse url: %w", err)
}
rl.Path = gopath.Join(rl.Path, ft.String(), sectorutil.SectorName(s))
urls[k] = rl.String()
}
out = append(out, StorageInfo{
ID: id,
URLs: urls,
Weight: st.info.Weight * 0, // TODO: something better than just '0'
CanSeal: st.info.CanSeal,
CanStore: st.info.CanStore,
})
}
}
2020-03-13 11:59:19 +00:00
return out, nil
}
2020-03-16 17:50:07 +00:00
func (i *Index) StorageInfo(ctx context.Context, id ID) (StorageInfo, error) {
2020-03-19 15:10:19 +00:00
i.lk.RLock()
defer i.lk.RUnlock()
2020-03-16 17:50:07 +00:00
si, found := i.stores[id]
if !found {
return StorageInfo{}, xerrors.Errorf("sector store not found")
}
2020-03-19 15:10:19 +00:00
return *si.info, nil
2020-03-16 17:50:07 +00:00
}
2020-03-19 15:10:19 +00:00
func (i *Index) StorageBestAlloc(ctx context.Context, allocate sectorbuilder.SectorFileType, sealing bool) ([]StorageInfo, error) {
i.lk.RLock()
defer i.lk.RUnlock()
var candidates []storageEntry
for _, p := range i.stores {
if sealing && !p.info.CanSeal {
log.Debugf("alloc: not considering %s; can't seal", p.info.ID)
2020-03-16 17:50:07 +00:00
continue
}
2020-03-19 15:10:19 +00:00
if !sealing && !p.info.CanStore {
log.Debugf("alloc: not considering %s; can't store", p.info.ID)
continue
2020-03-16 17:50:07 +00:00
}
2020-03-19 15:10:19 +00:00
// TODO: filter out of space
candidates = append(candidates, *p)
2020-03-16 17:50:07 +00:00
}
2020-03-19 15:10:19 +00:00
if len(candidates) == 0 {
return nil, xerrors.New("no good path found")
}
sort.Slice(candidates, func(i, j int) bool {
iw := big.Mul(big.NewInt(int64(candidates[i].fsi.Free)), big.NewInt(int64(candidates[i].info.Weight)))
jw := big.Mul(big.NewInt(int64(candidates[j].fsi.Free)), big.NewInt(int64(candidates[j].info.Weight)))
return iw.GreaterThan(jw)
})
out := make([]StorageInfo, len(candidates))
for i, candidate := range candidates {
out[i] = *candidate.info
}
return out, nil
}
func (i *Index) FindSector(id abi.SectorID, typ sectorbuilder.SectorFileType) ([]ID, error) {
i.lk.RLock()
defer i.lk.RUnlock()
return i.sectors[Decl{
SectorID: id,
SectorFileType: typ,
}], nil
2020-03-16 17:50:07 +00:00
}
2020-03-13 11:59:19 +00:00
var _ SectorIndex = &Index{}