diff --git a/block/block_manager.go b/block/block_manager.go index d4217edfb..27079cd0a 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -504,8 +504,34 @@ func (bm *Manager) CompactIndexes(ctx context.Context) error { return bm.fetchCommittedIndexAndMaybeCompact(ctx, 1) } -// ListBlocks returns the metadata about blocks with a given prefix and kind. -func (bm *Manager) ListBlocks(prefix string) ([]Info, error) { +// ListBlocks returns IDs of blocks matching given prefix. +func (bm *Manager) ListBlocks(prefix string) ([]string, error) { + bm.lock() + defer bm.unlock() + + var result []string + + appendToResult := func(i Info) error { + if i.Deleted || !strings.HasPrefix(i.BlockID, prefix) { + return nil + } + if bi, ok := bm.packIndexBuilder[i.BlockID]; ok && bi.Deleted { + return nil + } + result = append(result, i.BlockID) + return nil + } + + for _, bi := range bm.packIndexBuilder { + _ = appendToResult(*bi) + } + + _ = bm.committedBlocks.listBlocks(prefix, appendToResult) + return result, nil +} + +// ListBlockInfos returns the metadata about blocks with a given prefix and kind. +func (bm *Manager) ListBlockInfos(prefix string) ([]Info, error) { bm.lock() defer bm.unlock() diff --git a/cli/command_block_list.go b/cli/command_block_list.go index b80aeb34b..c04bd3c66 100644 --- a/cli/command_block_list.go +++ b/cli/command_block_list.go @@ -19,7 +19,7 @@ var ( ) func runListBlocksAction(ctx context.Context, rep *repo.Repository) error { - blocks, err := rep.Blocks.ListBlocks(*blockListPrefix) + blocks, err := rep.Blocks.ListBlockInfos(*blockListPrefix) if err != nil { return err } diff --git a/cli/command_block_stats.go b/cli/command_block_stats.go index e699de0d1..2d9ead5dc 100644 --- a/cli/command_block_stats.go +++ b/cli/command_block_stats.go @@ -17,7 +17,7 @@ var ( ) func runBlockStatsAction(ctx context.Context, rep *repo.Repository) error { - blocks, err := rep.Blocks.ListBlocks("") + blocks, err := rep.Blocks.ListBlockInfos("") if err != nil { return err } diff --git a/cli/command_object_ls.go b/cli/command_object_ls.go index 99b9009ea..6c974016d 100644 --- a/cli/command_object_ls.go +++ b/cli/command_object_ls.go @@ -13,7 +13,7 @@ var ( ) func runListObjectsAction(ctx context.Context, rep *repo.Repository) error { - info, err := rep.Blocks.ListBlocks(*objectListPrefix) + info, err := rep.Blocks.ListBlockInfos(*objectListPrefix) if err != nil { return err } diff --git a/examples/repository/main.go b/examples/repository/main.go index 964ec4a32..3dbc1307e 100644 --- a/examples/repository/main.go +++ b/examples/repository/main.go @@ -45,6 +45,6 @@ func main() { } for _, b := range blks { - log.Printf("found block: %v with length %v", b.BlockID, b.Length) + log.Printf("found block %v", b) } } diff --git a/manifest/manifest_manager.go b/manifest/manifest_manager.go index a81944ab9..e407bc203 100644 --- a/manifest/manifest_manager.go +++ b/manifest/manifest_manager.go @@ -246,16 +246,14 @@ func (m *Manager) load(ctx context.Context) error { return nil } -func (m *Manager) loadManifestBlocks(ctx context.Context, blocks []block.Info) error { +func (m *Manager) loadManifestBlocks(ctx context.Context, blockIDs []string) error { t0 := time.Now() log.Debug().Dur("duration_ms", time.Since(t0)).Msgf("finished loading manifest blocks.") - for _, b := range blocks { - m.blockIDs = append(m.blockIDs, b.BlockID) - } + m.blockIDs = append(m.blockIDs, blockIDs...) - manifests, err := m.loadBlocksInParallel(ctx, blocks) + manifests, err := m.loadBlocksInParallel(ctx, blockIDs) if err != nil { return err } @@ -276,10 +274,10 @@ func (m *Manager) loadManifestBlocks(ctx context.Context, blocks []block.Info) e return nil } -func (m *Manager) loadBlocksInParallel(ctx context.Context, blocks []block.Info) ([]manifest, error) { - errors := make(chan error, len(blocks)) - manifests := make(chan manifest, len(blocks)) - blockIDs := make(chan string, len(blocks)) +func (m *Manager) loadBlocksInParallel(ctx context.Context, blockIDs []string) ([]manifest, error) { + errors := make(chan error, len(blockIDs)) + manifests := make(chan manifest, len(blockIDs)) + ch := make(chan string, len(blockIDs)) var wg sync.WaitGroup for i := 0; i < 8; i++ { @@ -287,7 +285,7 @@ func (m *Manager) loadBlocksInParallel(ctx context.Context, blocks []block.Info) go func(workerID int) { defer wg.Done() - for blk := range blockIDs { + for blk := range ch { t1 := time.Now() man, err := m.loadManifestBlock(ctx, blk) log.Debug().Dur("duration", time.Since(t1)).Str("blk", blk).Int("worker", workerID).Err(err).Msg("manifest block loaded") @@ -301,10 +299,10 @@ func (m *Manager) loadBlocksInParallel(ctx context.Context, blocks []block.Info) } // feed block IDs for goroutines - for _, b := range blocks { - blockIDs <- b.BlockID + for _, b := range blockIDs { + ch <- b } - close(blockIDs) + close(ch) // wait for workers to complete wg.Wait()