From 4313d5f6abe7aaa059acd7b3d6b2c98183f20dbb Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Sun, 10 Jun 2018 20:35:32 -0700 Subject: [PATCH] added block.Manager.Refresh method --- block/block_manager.go | 33 +++++++++------ block/block_manager_test.go | 1 + block/committed_block_index.go | 5 ++- block/inmemory_committed_block_index.go | 54 ++++++++++++++++++------- block/list_cache.go | 2 +- block/simple_committed_block_index.go | 39 +++++++++++++++--- 6 files changed, 100 insertions(+), 34 deletions(-) diff --git a/block/block_manager.go b/block/block_manager.go index 83d096f05..d522f5b2e 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -76,6 +76,8 @@ type Manager struct { flushPackIndexesAfter time.Time // time when those indexes should be flushed activeBlocksExtraTime time.Duration + closed chan struct{} + writeFormatVersion int32 // format version to write maxPackSize int @@ -403,12 +405,12 @@ func sortBlocksByTime(b []IndexInfo) { }) } -func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, error) { +func (bm *Manager) loadPackIndexesUnlocked(ctx context.Context) ([]IndexInfo, bool, error) { nextSleepTime := 100 * time.Millisecond for i := 0; i < indexLoadAttempts; i++ { if err := ctx.Err(); err != nil { - return nil, err + return nil, false, err } if i > 0 { @@ -420,7 +422,7 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, erro blocks, err := bm.listCache.listIndexBlocks(ctx) if err != nil { - return nil, err + return nil, false, err } err = bm.tryLoadPackIndexBlocksLocked(ctx, blocks) @@ -429,17 +431,19 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, erro for _, b := range blocks { blockIDs = append(blockIDs, b.FileName) } - if err = bm.committedBlocks.use(blockIDs); err != nil { - return nil, err + var updated bool + updated, err = bm.committedBlocks.use(blockIDs) + if err != nil { + return nil, false, err } - return blocks, nil + return blocks, updated, nil } if err != storage.ErrBlockNotFound { - return nil, err + return nil, false, err } } - return nil, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts) + return nil, false, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts) } func (bm *Manager) tryLoadPackIndexBlocksLocked(ctx context.Context, blocks []IndexInfo) error { @@ -507,18 +511,16 @@ func (bm *Manager) unprocessedIndexBlocks(blocks []IndexInfo) (<-chan string, er // Close closes the block manager. func (bm *Manager) Close() { bm.blockCache.close() + close(bm.closed) } // CompactIndexes performs compaction of index blocks ensuring that # of small blocks is between minSmallBlockCount and maxSmallBlockCount func (bm *Manager) CompactIndexes(ctx context.Context, minSmallBlockCount int, maxSmallBlockCount int) error { - bm.lock() - defer bm.unlock() - if maxSmallBlockCount < minSmallBlockCount { return fmt.Errorf("invalid block counts") } - indexBlocks, err := bm.loadPackIndexesLocked(ctx) + indexBlocks, _, err := bm.loadPackIndexesUnlocked(ctx) if err != nil { return fmt.Errorf("error loading indexes: %v", err) } @@ -982,6 +984,12 @@ func (bm *Manager) assertLocked() { } } +// Refresh reloads the committed block indexes. +func (bm *Manager) Refresh(ctx context.Context) (bool, error) { + _, updated, err := bm.loadPackIndexesUnlocked(ctx) + return updated, err +} + type cachedList struct { Timestamp time.Time `json:"timestamp"` Blocks []IndexInfo `json:"blocks"` @@ -1075,6 +1083,7 @@ func newManagerWithOptions(ctx context.Context, st storage.Storage, f Formatting st: st, activeBlocksExtraTime: activeBlocksExtraTime, writeFormatVersion: int32(f.Version), + closed: make(chan struct{}), } if os.Getenv("KOPIA_VERIFY_INVARIANTS") != "" { diff --git a/block/block_manager_test.go b/block/block_manager_test.go index 1a1000592..84530842b 100644 --- a/block/block_manager_test.go +++ b/block/block_manager_test.go @@ -266,6 +266,7 @@ func TestBlockManagerConcurrency(t *testing.T) { preexistingBlock := writeBlockAndVerify(ctx, t, bm, seededRandomData(10, 100)) bm.Flush(ctx) + dumpBlockManagerData(t, data) bm1 := newTestBlockManager(data, keyTime, nil) bm2 := newTestBlockManager(data, keyTime, nil) bm3 := newTestBlockManager(data, keyTime, fakeTimeNowWithAutoAdvance(fakeTime.Add(1), 1*time.Second)) diff --git a/block/committed_block_index.go b/block/committed_block_index.go index 2994e0dd0..133794de5 100644 --- a/block/committed_block_index.go +++ b/block/committed_block_index.go @@ -11,11 +11,12 @@ type committedBlockIndex interface { addBlock(indexBlockID string, indexData []byte, use bool) error hasIndexBlockID(indexBlockID string) (bool, error) - use(indexBlockIDs []string) error + use(indexBlockIDs []string) (bool, error) } func newCommittedBlockIndex() committedBlockIndex { return &inMemoryCommittedBlockIndex{ - physicalBlocks: make(map[string]packindex.Index), + cachedPhysicalBlocks: make(map[string]packindex.Index), + usedPhysicalBlocks: make(map[string]packindex.Index), } } diff --git a/block/inmemory_committed_block_index.go b/block/inmemory_committed_block_index.go index 6821a2e93..664d4aeae 100644 --- a/block/inmemory_committed_block_index.go +++ b/block/inmemory_committed_block_index.go @@ -7,12 +7,14 @@ import ( "github.com/kopia/kopia/internal/packindex" "github.com/kopia/kopia/storage" + "github.com/rs/zerolog/log" ) type inMemoryCommittedBlockIndex struct { - mu sync.Mutex - indexes packindex.Merged - physicalBlocks map[string]packindex.Index + mu sync.Mutex + indexes packindex.Merged + usedPhysicalBlocks map[string]packindex.Index + cachedPhysicalBlocks map[string]packindex.Index } func (b *inMemoryCommittedBlockIndex) getBlock(blockID string) (Info, error) { @@ -36,18 +38,19 @@ func (b *inMemoryCommittedBlockIndex) addBlock(indexBlockID string, data []byte, b.mu.Lock() defer b.mu.Unlock() - if b.physicalBlocks[indexBlockID] == nil { + if b.cachedPhysicalBlocks[indexBlockID] == nil { ndx, err := packindex.Open(bytes.NewReader(data)) if err != nil { return err } - b.physicalBlocks[indexBlockID] = ndx - - if use { - b.indexes = append(b.indexes, ndx) - } + b.cachedPhysicalBlocks[indexBlockID] = ndx } + if use && b.usedPhysicalBlocks[indexBlockID] == nil { + ndx := b.cachedPhysicalBlocks[indexBlockID] + b.usedPhysicalBlocks[indexBlockID] = ndx + b.indexes = append(b.indexes, ndx) + } return nil } @@ -62,27 +65,50 @@ func (b *inMemoryCommittedBlockIndex) hasIndexBlockID(indexBlockID string) (bool b.mu.Lock() defer b.mu.Unlock() - return b.physicalBlocks[indexBlockID] != nil, nil + return b.cachedPhysicalBlocks[indexBlockID] != nil, nil } -func (b *inMemoryCommittedBlockIndex) use(packFiles []string) error { +func (b *inMemoryCommittedBlockIndex) packFilesChanged(packFiles []string) bool { + if len(packFiles) != len(b.usedPhysicalBlocks) { + return true + } + + for _, packFile := range packFiles { + if b.usedPhysicalBlocks[packFile] == nil { + return true + } + } + + return false +} + +func (b *inMemoryCommittedBlockIndex) use(packFiles []string) (bool, error) { b.mu.Lock() defer b.mu.Unlock() + if !b.packFilesChanged(packFiles) { + return false, nil + } + + log.Printf("using %v", packFiles) + var newIndexes packindex.Merged + newUsedBlocks := map[string]packindex.Index{} defer func() { newIndexes.Close() //nolint:errcheck }() for _, e := range packFiles { - ndx := b.physicalBlocks[e] + ndx := b.cachedPhysicalBlocks[e] if ndx == nil { - return fmt.Errorf("unable to open pack index %q", e) + return false, fmt.Errorf("unable to open pack index %q", e) } //log.Printf("opened %v with %v entries", e, ndx.EntryCount()) newIndexes = append(newIndexes, ndx) + newUsedBlocks[e] = ndx } b.indexes = newIndexes + b.usedPhysicalBlocks = newUsedBlocks newIndexes = nil - return nil + return true, nil } diff --git a/block/list_cache.go b/block/list_cache.go index f2968a269..f19ead3a2 100644 --- a/block/list_cache.go +++ b/block/list_cache.go @@ -34,7 +34,6 @@ func (c *listCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) { } } - log.Debug().Msg("listing index blocks from source") blocks, err := listIndexBlocksFromStorage(ctx, c.st) if err == nil { c.saveListToCache(ctx, &cachedList{ @@ -42,6 +41,7 @@ func (c *listCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) { Timestamp: time.Now(), }) } + log.Debug().Msgf("found %v index blocks from source", len(blocks)) return blocks, err } diff --git a/block/simple_committed_block_index.go b/block/simple_committed_block_index.go index e1672f4c9..a94234201 100644 --- a/block/simple_committed_block_index.go +++ b/block/simple_committed_block_index.go @@ -24,6 +24,7 @@ type simpleCommittedBlockIndex struct { dirname string mu sync.Mutex + inUse map[string]packindex.Index merged packindex.Merged } @@ -111,11 +112,15 @@ func (b *simpleCommittedBlockIndex) addBlock(indexBlockID string, data []byte, u b.mu.Lock() defer b.mu.Unlock() + if b.inUse[indexBlockID] != nil { + return nil + } + ndx, err := b.openIndex(b.indexBlockPath(indexBlockID)) if err != nil { return fmt.Errorf("unable to open pack index %q: %v", indexBlockID, err) } - + b.inUse[indexBlockID] = ndx b.merged = append(b.merged, ndx) return nil } @@ -137,11 +142,31 @@ func (b *simpleCommittedBlockIndex) openIndex(fullpath string) (packindex.Index, return packindex.Open(f) } -func (b *simpleCommittedBlockIndex) use(packFiles []string) error { +func (b *simpleCommittedBlockIndex) packFilesChanged(packFiles []string) bool { + if len(packFiles) != len(b.inUse) { + return true + } + + for _, packFile := range packFiles { + if b.inUse[packFile] == nil { + return true + } + } + + return false +} + +func (b *simpleCommittedBlockIndex) use(packFiles []string) (bool, error) { b.mu.Lock() defer b.mu.Unlock() + if !b.packFilesChanged(packFiles) { + return false, nil + } + log.Printf("set of index files has changed (had %v, now %v)", len(b.inUse), len(packFiles)) + var newMerged packindex.Merged + newInUse := map[string]packindex.Index{} defer func() { newMerged.Close() //nolint:errcheck }() @@ -149,7 +174,7 @@ func (b *simpleCommittedBlockIndex) use(packFiles []string) error { remaining := map[string]os.FileInfo{} entries, err := ioutil.ReadDir(b.dirname) if err != nil { - return fmt.Errorf("can't list cache: %v", err) + return false, fmt.Errorf("can't list cache: %v", err) } for _, ent := range entries { @@ -164,14 +189,17 @@ func (b *simpleCommittedBlockIndex) use(packFiles []string) error { fullpath := filepath.Join(b.dirname, fname) ndx, err := b.openIndex(fullpath) if err != nil { - return fmt.Errorf("unable to open pack index %q: %v", fullpath, err) + return false, fmt.Errorf("unable to open pack index %q: %v", fullpath, err) } log.Printf("opened %v with %v entries", fullpath, ndx.EntryCount()) newMerged = append(newMerged, ndx) + newInUse[e] = ndx } b.merged = newMerged + b.inUse = newInUse newMerged = nil + for _, rem := range remaining { if time.Since(rem.ModTime()) > unusedCommittedBlockIndexCleanupTime { log.Printf("removing unused %v %v", rem.Name(), rem.ModTime()) @@ -182,7 +210,7 @@ func (b *simpleCommittedBlockIndex) use(packFiles []string) error { log.Printf("keeping unused %v because it's too new %v", rem.Name(), rem.ModTime()) } } - return nil + return true, nil } func newSimpleCommittedBlockIndex(dirname string) (committedBlockIndex, error) { @@ -190,6 +218,7 @@ func newSimpleCommittedBlockIndex(dirname string) (committedBlockIndex, error) { s := &simpleCommittedBlockIndex{ dirname: dirname, + inUse: map[string]packindex.Index{}, } return s, nil }