diff --git a/gql/resolver.go b/gql/resolver.go index dd21dda01..331a1ee0a 100644 --- a/gql/resolver.go +++ b/gql/resolver.go @@ -20,6 +20,7 @@ import ( "github.com/filecoin-project/boost/node/modules/dtypes" "github.com/filecoin-project/boost/piecedirectory" "github.com/filecoin-project/boost/retrievalmarket/rtvllog" + "github.com/filecoin-project/boost/sectorstatemgr" "github.com/filecoin-project/boost/storagemanager" "github.com/filecoin-project/boost/storagemarket" "github.com/filecoin-project/boost/storagemarket/sealingpipeline" @@ -62,6 +63,7 @@ type resolver struct { legacyProv gfm_storagemarket.StorageProvider legacyDT dtypes.ProviderDataTransfer ps piecestore.PieceStore + ssm *sectorstatemgr.SectorStateMgr sa retrievalmarket.SectorAccessor piecedirectory *piecedirectory.PieceDirectory publisher *storageadapter.DealPublisher @@ -69,7 +71,7 @@ type resolver struct { fullNode v1api.FullNode } -func NewResolver(ctx context.Context, cfg *config.Boost, r lotus_repo.LockedRepo, h host.Host, dealsDB *db.DealsDB, logsDB *db.LogsDB, retDB *rtvllog.RetrievalLogDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, spApi sealingpipeline.API, provider *storagemarket.Provider, legacyProv gfm_storagemarket.StorageProvider, legacyDT dtypes.ProviderDataTransfer, ps piecestore.PieceStore, sa retrievalmarket.SectorAccessor, piecedirectory *piecedirectory.PieceDirectory, publisher *storageadapter.DealPublisher, fullNode v1api.FullNode) *resolver { +func NewResolver(ctx context.Context, cfg *config.Boost, r lotus_repo.LockedRepo, h host.Host, dealsDB *db.DealsDB, logsDB *db.LogsDB, retDB *rtvllog.RetrievalLogDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, spApi sealingpipeline.API, provider *storagemarket.Provider, legacyProv gfm_storagemarket.StorageProvider, legacyDT dtypes.ProviderDataTransfer, ps piecestore.PieceStore, sa retrievalmarket.SectorAccessor, piecedirectory *piecedirectory.PieceDirectory, publisher *storageadapter.DealPublisher, fullNode v1api.FullNode, ssm *sectorstatemgr.SectorStateMgr) *resolver { return &resolver{ ctx: ctx, cfg: cfg, @@ -91,6 +93,7 @@ func NewResolver(ctx context.Context, cfg *config.Boost, r lotus_repo.LockedRepo publisher: publisher, spApi: spApi, fullNode: fullNode, + ssm: ssm, } } diff --git a/gql/resolver_lid.go b/gql/resolver_lid.go new file mode 100644 index 000000000..380d41294 --- /dev/null +++ b/gql/resolver_lid.go @@ -0,0 +1,91 @@ +package gql + +import ( + "context" + "time" + + "github.com/filecoin-project/boost/db" + gqltypes "github.com/filecoin-project/boost/gql/types" + "github.com/filecoin-project/boost/sectorstatemgr" +) + +type dealData struct { + Indexed gqltypes.Uint64 + FlaggedUnsealed gqltypes.Uint64 + FlaggedSealed gqltypes.Uint64 +} + +type pieces struct { + Indexed int32 + FlaggedUnsealed int32 + FlaggedSealed int32 +} + +type sectorUnsealedCopies struct { + Unsealed int32 + Sealed int32 +} + +type sectorProvingState struct { + Active int32 + Inactive int32 +} + +type lidState struct { + DealData dealData + Pieces pieces + SectorUnsealedCopies sectorUnsealedCopies + SectorProvingState sectorProvingState + FlaggedPieces int32 +} + +// query: lid: [LID] +func (r *resolver) LID(ctx context.Context) (*lidState, error) { + var lu *sectorstatemgr.SectorStateUpdates + for lu == nil { + r.ssm.LatestUpdateMu.Lock() + lu = r.ssm.LatestUpdate + r.ssm.LatestUpdateMu.Unlock() + if lu == nil { + time.Sleep(2 * time.Second) + } + } + + var sealed, unsealed int32 + for _, s := range lu.SectorStates { // TODO: consider adding this data directly in SSM + if s == db.SealStateUnsealed { + unsealed++ + } else if s == db.SealStateSealed { + sealed++ + } + } + + fp, err := r.piecedirectory.FlaggedPiecesCount(ctx) + if err != nil { + return nil, err + } + + ls := &lidState{ + FlaggedPieces: int32(fp), + DealData: dealData{ + Indexed: gqltypes.Uint64(12094627905536), + FlaggedUnsealed: gqltypes.Uint64(1094627905536), + FlaggedSealed: gqltypes.Uint64(18094627905536), + }, + Pieces: pieces{ + Indexed: 360, + FlaggedUnsealed: 33, + FlaggedSealed: 480, + }, + SectorUnsealedCopies: sectorUnsealedCopies{ + Sealed: sealed, + Unsealed: unsealed, + }, + SectorProvingState: sectorProvingState{ + Active: int32(len(lu.ActiveSectors)), + Inactive: int32(len(lu.SectorStates) - len(lu.ActiveSectors)), // TODO: add an explicit InactiveSectors in ssm + }, + } + + return ls, nil +} diff --git a/gql/schema.graphql b/gql/schema.graphql index 5748c5ec9..9b23cf61a 100644 --- a/gql/schema.graphql +++ b/gql/schema.graphql @@ -317,6 +317,36 @@ type SealingPipeline { Workers: [Worker]! } +type DealData { + Indexed: Uint64! + FlaggedUnsealed: Uint64! + FlaggedSealed: Uint64! +} + +type Pieces { + Indexed: Int! + FlaggedUnsealed: Int! + FlaggedSealed: Int! +} + +type SectorUnsealedCopies { + Unsealed: Int! + Sealed: Int! +} + +type SectorProvingState { + Active: Int! + Inactive: Int! +} + +type LID { + DealData: DealData! + Pieces: Pieces! + SectorUnsealedCopies: SectorUnsealedCopies! + SectorProvingState: SectorProvingState! + FlaggedPieces: Int! +} + type FundsEscrow { Available: BigInt! Locked: BigInt! @@ -476,6 +506,9 @@ type RootQuery { """Get sealing pipeline state""" sealingpipeline: SealingPipeline! + """Get LID state""" + lid: LID! + """Get funds available""" funds: Funds! diff --git a/node/modules/storageminer.go b/node/modules/storageminer.go index a1fc873cb..ee74ae2bf 100644 --- a/node/modules/storageminer.go +++ b/node/modules/storageminer.go @@ -32,6 +32,7 @@ import ( "github.com/filecoin-project/boost/piecedirectory" brm "github.com/filecoin-project/boost/retrievalmarket/lib" "github.com/filecoin-project/boost/retrievalmarket/rtvllog" + "github.com/filecoin-project/boost/sectorstatemgr" "github.com/filecoin-project/boost/storagemanager" "github.com/filecoin-project/boost/storagemarket" "github.com/filecoin-project/boost/storagemarket/logs" @@ -628,14 +629,14 @@ func NewStorageMarketProvider(provAddr address.Address, cfg *config.Boost) func( } } -func NewGraphqlServer(cfg *config.Boost) func(lc fx.Lifecycle, r repo.LockedRepo, h host.Host, prov *storagemarket.Provider, dealsDB *db.DealsDB, logsDB *db.LogsDB, retDB *rtvllog.RetrievalLogDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, publisher *storageadapter.DealPublisher, spApi sealingpipeline.API, legacyProv gfm_storagemarket.StorageProvider, legacyDT dtypes.ProviderDataTransfer, ps dtypes.ProviderPieceStore, sa retrievalmarket.SectorAccessor, piecedirectory *piecedirectory.PieceDirectory, fullNode v1api.FullNode, bg gql.BlockGetter) *gql.Server { +func NewGraphqlServer(cfg *config.Boost) func(lc fx.Lifecycle, r repo.LockedRepo, h host.Host, prov *storagemarket.Provider, dealsDB *db.DealsDB, logsDB *db.LogsDB, retDB *rtvllog.RetrievalLogDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, publisher *storageadapter.DealPublisher, spApi sealingpipeline.API, legacyProv gfm_storagemarket.StorageProvider, legacyDT dtypes.ProviderDataTransfer, ps dtypes.ProviderPieceStore, sa retrievalmarket.SectorAccessor, piecedirectory *piecedirectory.PieceDirectory, fullNode v1api.FullNode, bg gql.BlockGetter, ssm *sectorstatemgr.SectorStateMgr) *gql.Server { return func(lc fx.Lifecycle, r repo.LockedRepo, h host.Host, prov *storagemarket.Provider, dealsDB *db.DealsDB, logsDB *db.LogsDB, retDB *rtvllog.RetrievalLogDB, plDB *db.ProposalLogsDB, fundsDB *db.FundsDB, fundMgr *fundmanager.FundManager, storageMgr *storagemanager.StorageManager, publisher *storageadapter.DealPublisher, spApi sealingpipeline.API, legacyProv gfm_storagemarket.StorageProvider, legacyDT dtypes.ProviderDataTransfer, - ps dtypes.ProviderPieceStore, sa retrievalmarket.SectorAccessor, piecedirectory *piecedirectory.PieceDirectory, fullNode v1api.FullNode, bg gql.BlockGetter) *gql.Server { + ps dtypes.ProviderPieceStore, sa retrievalmarket.SectorAccessor, piecedirectory *piecedirectory.PieceDirectory, fullNode v1api.FullNode, bg gql.BlockGetter, ssm *sectorstatemgr.SectorStateMgr) *gql.Server { resolverCtx, cancel := context.WithCancel(context.Background()) - resolver := gql.NewResolver(resolverCtx, cfg, r, h, dealsDB, logsDB, retDB, plDB, fundsDB, fundMgr, storageMgr, spApi, prov, legacyProv, legacyDT, ps, sa, piecedirectory, publisher, fullNode) + resolver := gql.NewResolver(resolverCtx, cfg, r, h, dealsDB, logsDB, retDB, plDB, fundsDB, fundMgr, storageMgr, spApi, prov, legacyProv, legacyDT, ps, sa, piecedirectory, publisher, fullNode, ssm) server := gql.NewServer(resolver, bg) lc.Append(fx.Hook{ diff --git a/piecedirectory/doctor.go b/piecedirectory/doctor.go index 3e0caaa37..cb4d4fcdb 100644 --- a/piecedirectory/doctor.go +++ b/piecedirectory/doctor.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "math/rand" - "sync" "time" "github.com/filecoin-project/boost/db" @@ -27,9 +26,6 @@ var doclog = logging.Logger("piecedoc") type Doctor struct { store *bdclient.Store ssm *sectorstatemgr.SectorStateMgr - - latestUpdateMu sync.Mutex - latestUpdate *sectorstatemgr.SectorStateUpdates } func NewDoctor(store *bdclient.Store, ssm *sectorstatemgr.SectorStateMgr) *Doctor { @@ -42,28 +38,6 @@ const avgCheckInterval = 30 * time.Second func (d *Doctor) Run(ctx context.Context) { doclog.Info("piece doctor: running") - go func() { - sub := d.ssm.PubSub.Subscribe() - - for { - select { - case u, ok := <-sub: - if !ok { - log.Debugw("state updates subscription closed") - return - } - log.Debugw("got state updates from SectorStateMgr", "len(u.updates)", len(u.Updates), "len(u.active)", len(u.ActiveSectors), "u.updatedAt", u.UpdatedAt) - - d.latestUpdateMu.Lock() - d.latestUpdate = u - d.latestUpdateMu.Unlock() - - case <-ctx.Done(): - return - } - } - }() - timer := time.NewTimer(0) defer timer.Stop() @@ -76,9 +50,9 @@ func (d *Doctor) Run(ctx context.Context) { err := func() error { var lu *sectorstatemgr.SectorStateUpdates - d.latestUpdateMu.Lock() - lu = d.latestUpdate - d.latestUpdateMu.Unlock() + d.ssm.LatestUpdateMu.Lock() + lu = d.ssm.LatestUpdate + d.ssm.LatestUpdateMu.Unlock() if lu == nil { doclog.Warn("sector state manager not yet updated") return nil diff --git a/react/src/LID.js b/react/src/LID.js index d0e20e299..001f05de1 100644 --- a/react/src/LID.js +++ b/react/src/LID.js @@ -1,6 +1,7 @@ /* global BigInt */ import {useMutation, useQuery} from "@apollo/react-hooks"; import { + LIDQuery, FlaggedPiecesQuery, PieceBuildIndexMutation, PieceStatusQuery, PiecesWithPayloadCidQuery, PiecesWithRootPayloadCidQuery } from "./gql"; @@ -39,59 +40,7 @@ export function LIDPage(props) { } function LIDContent() { - return
- - - - - - - - -
- - - - - -
-
-} - -function BlockStatsSection() { - return
-

Block Stats

- - - - - - - - - - - - -
Total blocks:{addCommas(32129310123)}
Avg Block size:{humanFileSize(256*1024)}
-
-} - -function mockDealData() { - return { - data: { - storage: { - Indexed: BigInt(12345432343223), - Flagged: BigInt(1234512345423), - Sealed: BigInt(1634512345423), - } - } - } -} - -function DealDataSection() { - // const {loading, error, data} = useQuery(StorageQuery, { pollInterval: 10000 }) - const {loading, error, data} = mockDealData() + const {loading, error, data} = useQuery(LIDQuery, { pollInterval: 30000 }) if (loading) { return
Loading...
@@ -100,184 +49,136 @@ function DealDataSection() { return
Error: {error.message}
} - var storage = data.storage + const d = data.lid - const bars = [{ + const dealDataBars = [{ name: 'Indexed', className: 'indexed', - amount: storage.Indexed, + amount: d.DealData.Indexed, description: '' }, { - name: 'Flagged', + name: 'Flagged (unsealed)', className: 'flagged', - amount: storage.Flagged, + amount: d.DealData.FlaggedUnsealed, description: '' }, { - name: 'Sealed Only', + name: 'Flagged (sealed only)', className: 'sealed', - amount: storage.Sealed, + amount: d.DealData.FlaggedSealed, description: '' }] - return
-

Deal Data

- -
- - -
-
-} - -function mockPiecesData() { - return { - data: { - storage: { - Indexed: BigInt(2132), - Flagged: BigInt(123), - Sealed: BigInt(213), - } - } - } -} - -function PiecesSection() { - // const {loading, error, data} = useQuery(StorageQuery, { pollInterval: 10000 }) - const {loading, error, data} = mockPiecesData() - - if (loading) { - return
Loading...
- } - if (error) { - return
Error: {error.message}
- } - - var storage = data.storage - - const bars = [{ + const piecesBars = [{ name: 'Indexed', className: 'indexed', - amount: storage.Indexed, + amount: d.Pieces.Indexed, description: '' }, { - name: 'Flagged', + name: 'Flagged (unsealed)', className: 'flagged', - amount: storage.Flagged, + amount: d.Pieces.FlaggedUnsealed, description: '' }, { - name: 'Sealed Only', + name: 'Flagged (sealed only)', className: 'sealed', - amount: storage.Sealed, + amount: d.Pieces.FlaggedSealed, description: '' }] - return
-

Pieces

- -
- - -
- -
-

Flagged Pieces

-

- {addCommas(storage.Flagged)} Flagged Pieces - View Flagged Pieces -

-
-
-} - -function mockSectorUnsealedData() { - return { - data: { - sectors: { - Unsealed: BigInt(1334), - Sealed: BigInt(321), - } - } - } -} - -function SectorUnsealedSection() { - // const {loading, error, data} = useQuery(StorageQuery, { pollInterval: 10000 }) - const {loading, error, data} = mockSectorUnsealedData() - - if (loading) { - return
Loading...
- } - if (error) { - return
Error: {error.message}
- } - - var sectors = data.sectors - - const bars = [{ - name: 'Unsealed Copy', + const barsSuc = [{ + name: 'Unsealed', className: 'unsealed', - amount: sectors.Unsealed, + amount: d.SectorUnsealedCopies.Unsealed, description: '' }, { name: 'Sealed Only', className: 'sealed', - amount: sectors.Sealed, + amount: d.SectorUnsealedCopies.Sealed, description: '' }] - return
-

Sector Unsealed Copies

- -
- - -
-
-} - -function mockSectorProvingState() { - return { - data: { - sectors: { - Active: BigInt(1274), - Inactive: BigInt(381), - } - } - } -} - -function SectorProvingState() { - // const {loading, error, data} = useQuery(StorageQuery, { pollInterval: 10000 }) - const {loading, error, data} = mockSectorProvingState() - - if (loading) { - return
Loading...
- } - if (error) { - return
Error: {error.message}
- } - - var sectors = data.sectors - - const bars = [{ + const barsSps = [{ name: 'Active', className: 'active', - amount: sectors.Active, + amount: d.SectorProvingState.Active, description: '' }, { name: 'Inactive', className: 'inactive', - amount: sectors.Inactive, + amount: d.SectorProvingState.Inactive, description: '' }] + return
+ + + + + + + + +
+
+

Deal Data

+
+ + +
+
+
+

Pieces

+ +
+ + +
+ +
+

Flagged Pieces

+

+ {addCommas(d.FlaggedPieces)} Flagged Pieces + View Flagged Pieces +

+
+
+ 		+
+

Sectors Copies

+ +
+ + +
+
+
+

Sectors Proving State

+ +
+ + +
+
+
+
+} + +function BlockStatsSection() { return
-

Sector Proving State

+

Block Stats

-
- - -
+ + + + + + + + + + + +
Total blocks:{addCommas(32129310123)}
Avg Block size:{humanFileSize(256*1024)}
} diff --git a/react/src/gql.js b/react/src/gql.js index 55be46ae8..f78eb3403 100644 --- a/react/src/gql.js +++ b/react/src/gql.js @@ -471,6 +471,32 @@ const PieceStatusQuery = gql` } `; +const LIDQuery = gql` + query AppLIDQuery { + lid { + DealData { + Indexed + FlaggedUnsealed + FlaggedSealed + } + Pieces { + Indexed + FlaggedUnsealed + FlaggedSealed + } + SectorUnsealedCopies { + Unsealed + Sealed + } + SectorProvingState { + Active + Inactive + } + FlaggedPieces + } + } +`; + const StorageQuery = gql` query AppStorageQuery { storage { @@ -712,6 +738,7 @@ export { PieceBuildIndexMutation, PieceStatusQuery, FlaggedPiecesQuery, + LIDQuery, StorageQuery, LegacyStorageQuery, FundsQuery, diff --git a/sectorstatemgr/sectorstatemgr.go b/sectorstatemgr/sectorstatemgr.go index de2fbac8d..2156fc202 100644 --- a/sectorstatemgr/sectorstatemgr.go +++ b/sectorstatemgr/sectorstatemgr.go @@ -44,6 +44,9 @@ type SectorStateMgr struct { PubSub *PubSub + LatestUpdateMu sync.Mutex + LatestUpdate *SectorStateUpdates + sdb *db.SectorStateDB } @@ -77,10 +80,36 @@ func NewSectorStateMgr(cfg *config.Boost) func(lc fx.Lifecycle, sdb *db.SectorSt } } +func (m *SectorStateMgr) UpdateLatest(ctx context.Context) { + go func() { + sub := m.PubSub.Subscribe() + + for { + select { + case u, ok := <-sub: + if !ok { + log.Debugw("state updates subscription closed") + return + } + log.Debugw("got state updates from SectorStateMgr", "len(u.updates)", len(u.Updates), "len(u.active)", len(u.ActiveSectors), "u.updatedAt", u.UpdatedAt) + + m.LatestUpdateMu.Lock() + m.LatestUpdate = u + m.LatestUpdateMu.Unlock() + + case <-ctx.Done(): + return + } + } + }() +} + func (m *SectorStateMgr) Run(ctx context.Context) { duration := time.Duration(m.cfg.StorageListRefreshDuration) log.Infof("starting sector state manager running on interval %s", duration.String()) + m.UpdateLatest(ctx) + // Check immediately err := m.checkForUpdates(ctx) if err != nil {