exposed block.Manager.ListBlocks() which only returns block IDs instead of full info

This commit is contained in:
Jarek Kowalski committed 2018-05-05 16:16:48 -07:00
1 parent 3a02ef24e0
commit 62ce401cd6
6 files changed
+43 -19

No files matched your search

+28 -2
View File
@@ -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()
+1 -1
View File
@@ -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
}
+1 -1
View File
@@ -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
}
+1 -1
View File
@@ -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
}
+1 -1
View File
@@ -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)
}
}
+11 -13
View File
@@ -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()