diff --git a/block/block_cache.go b/block/block_cache.go index 3edb43104..8cec9324b 100644 --- a/block/block_cache.go +++ b/block/block_cache.go @@ -6,15 +6,16 @@ import ( "path/filepath" "time" - "github.com/kopia/kopia/storage/filesystem" - "github.com/kopia/kopia/storage" + "github.com/kopia/kopia/storage/filesystem" ) type blockCache interface { getBlock(ctx context.Context, cacheKey string, physicalBlockID PhysicalBlockID, offset, length int64) ([]byte, error) putBlock(ctx context.Context, blockID PhysicalBlockID, data []byte) error - listIndexBlocks(ctx context.Context, full bool, extraTime time.Duration) ([]IndexInfo, error) + listIndexBlocks(ctx context.Context) ([]IndexInfo, error) + deleteListCache(ctx context.Context) + deleteBlock(ctx context.Context, blockID PhysicalBlockID) error close() error } diff --git a/block/block_manager.go b/block/block_manager.go index 462c0d165..e4bd21940 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -8,7 +8,6 @@ import ( "encoding/hex" "fmt" "io" - "math" "math/rand" "os" "path/filepath" @@ -25,19 +24,22 @@ import ( ) const ( - parallelFetches = 5 // number of parallel reads goroutines - flushPackIndexTimeout = 10 * time.Minute // time after which all pending indexes are flushes - newIndexBlockPrefix = "n" - compactedBlockSuffix = "-z" - defaultMinPreambleLength = 32 - defaultMaxPreambleLength = 32 - defaultPaddingUnit = 4096 - autoCompactionBlockCount = 16 + parallelFetches = 5 // number of parallel reads goroutines + flushPackIndexTimeout = 10 * time.Minute // time after which all pending indexes are flushes + newIndexBlockPrefix = "n" + defaultMinPreambleLength = 32 + defaultMaxPreambleLength = 32 + defaultPaddingUnit = 4096 + autoCompactionMinBlockCount = 4 * parallelFetches + autoCompactionMaxBlockCount = 64 + defaultActiveBlocksExtraTime = 10 * time.Minute currentWriteVersion = 1 minSupportedReadVersion = 0 maxSupportedReadVersion = currentWriteVersion + + indexLoadAttempts = 10 ) // Info is an information about a single block managed by Manager. @@ -244,12 +246,12 @@ func (bm *Manager) flushPackIndexesLocked(ctx context.Context) error { data := buf.Bytes() dataCopy := append([]byte(nil), data...) - indexBlockID, err := bm.writePackIndexesNew(ctx, data, false) + indexBlockID, err := bm.writePackIndexesNew(ctx, data) if err != nil { return err } - if err := bm.committedBlocks.addBlock(indexBlockID, dataCopy); err != nil { + if err := bm.committedBlocks.addBlock(indexBlockID, dataCopy, true); err != nil { return fmt.Errorf("unable to add committed block: %v", err) } bm.packIndexBuilder = packindex.NewBuilder() @@ -259,14 +261,8 @@ func (bm *Manager) flushPackIndexesLocked(ctx context.Context) error { return nil } -func (bm *Manager) writePackIndexesNew(ctx context.Context, data []byte, isCompaction bool) (PhysicalBlockID, error) { - var suffix string - if isCompaction { - suffix = compactedBlockSuffix - } - - inverseTimePrefix := fmt.Sprintf("%016x", math.MaxInt64-bm.timeNow().UnixNano()) - return bm.encryptAndWriteBlockNotLocked(ctx, data, newIndexBlockPrefix+inverseTimePrefix, suffix) +func (bm *Manager) writePackIndexesNew(ctx context.Context, data []byte) (PhysicalBlockID, error) { + return bm.encryptAndWriteBlockNotLocked(ctx, data, newIndexBlockPrefix) } func (bm *Manager) finishPackLocked(ctx context.Context) error { @@ -370,29 +366,12 @@ func appendRandomBytes(b []byte, count int) ([]byte, error) { return append(b, rnd...), nil } -// ListIndexBlocks returns the list of all index blocks, including inactive, sorted by time. -func (bm *Manager) ListIndexBlocks(ctx context.Context) ([]IndexInfo, error) { - blocks, err := bm.cache.listIndexBlocks(ctx, true, 0) - if err != nil { - return nil, fmt.Errorf("error listing index blocks: %v", err) - } - - sortBlocksByTime(blocks) - return blocks, nil -} - -// ActiveIndexBlocks returns the list of active index blocks, sorted by time. -func (bm *Manager) ActiveIndexBlocks(ctx context.Context) ([]IndexInfo, error) { - blocks, err := bm.cache.listIndexBlocks(ctx, false, bm.activeBlocksExtraTime) +// IndexBlocks returns the list of active index blocks, sorted by time. +func (bm *Manager) IndexBlocks(ctx context.Context) ([]IndexInfo, error) { + blocks, err := bm.cache.listIndexBlocks(ctx) if err != nil { return nil, err } - // for i, b := range blocks { - // log.Printf("found block #%v %v %v %v", i, b.BlockID, b.Length, b.Timestamp) - // } - if len(blocks) == 0 { - return nil, nil - } sortBlocksByTime(blocks) return blocks, nil @@ -404,19 +383,52 @@ func sortBlocksByTime(b []IndexInfo) { }) } -func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]PhysicalBlockID, error) { - blocks, err := bm.ActiveIndexBlocks(ctx) - if err != nil { - return nil, err +func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]IndexInfo, error) { + nextSleepTime := 100 * time.Millisecond + + for i := 0; i < indexLoadAttempts; i++ { + if err := ctx.Err(); err != nil { + return nil, err + } + + if i > 0 { + bm.cache.deleteListCache(ctx) + log.Printf("encountered NOT_FOUND when loading, sleeping %v before retrying #%v", nextSleepTime, i) + time.Sleep(nextSleepTime) + nextSleepTime *= 2 + } + + blocks, err := bm.cache.listIndexBlocks(ctx) + if err != nil { + return nil, err + } + + err = bm.tryLoadPackIndexBlocksLocked(ctx, blocks) + if err == nil { + var blockIDs []PhysicalBlockID + for _, b := range blocks { + blockIDs = append(blockIDs, b.BlockID) + } + if err = bm.committedBlocks.use(blockIDs); err != nil { + return nil, err + } + return blocks, nil + } + if err != storage.ErrBlockNotFound { + return nil, err + } } - indexBlockIDs := indexBlockIDs(blocks) - ch, err := bm.unprocessedIndexBlocks(indexBlockIDs) + return nil, fmt.Errorf("unable to load pack indexes despite %v retries", indexLoadAttempts) +} + +func (bm *Manager) tryLoadPackIndexBlocksLocked(ctx context.Context, blocks []IndexInfo) error { + ch, err := bm.unprocessedIndexBlocks(blocks) if err != nil { - return nil, err + return err } if len(ch) == 0 { - return indexBlockIDs, nil + return nil } var wg sync.WaitGroup @@ -435,8 +447,8 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]PhysicalBlockID return } - if err := bm.loadPackIndexes(indexBlockID, data); err != nil { - errors <- err + if err := bm.committedBlocks.addBlock(indexBlockID, data, false); err != nil { + errors <- fmt.Errorf("unable to add to committed block cache: %v", err) return } } @@ -448,66 +460,89 @@ func (bm *Manager) loadPackIndexesLocked(ctx context.Context) ([]PhysicalBlockID // Propagate async errors, if any. for err := range errors { - return nil, err + return err } - return indexBlockIDs, nil + return nil } // unprocessedIndexBlocks returns a closed channel filled with block IDs that are not in committedBlocks cache. -func (bm *Manager) unprocessedIndexBlocks(blocks []PhysicalBlockID) (<-chan PhysicalBlockID, error) { +func (bm *Manager) unprocessedIndexBlocks(blocks []IndexInfo) (<-chan PhysicalBlockID, error) { ch := make(chan PhysicalBlockID, len(blocks)) - for _, blockID := range blocks { - has, err := bm.committedBlocks.hasIndexBlockID(blockID) + for _, block := range blocks { + has, err := bm.committedBlocks.hasIndexBlockID(block.BlockID) if err != nil { return nil, err } if has { - log.Printf("index block %q already in cache, skipping", blockID) + log.Printf("index block %q already in cache, skipping", block.BlockID) continue } - ch <- blockID + ch <- block.BlockID } close(ch) return ch, nil } -func indexBlockIDs(blocks []IndexInfo) []PhysicalBlockID { - var indexBlockIDs []PhysicalBlockID - for _, b := range blocks { - indexBlockIDs = append(indexBlockIDs, b.BlockID) - } - return indexBlockIDs -} - -func (bm *Manager) loadPackIndexes(indexBlockID PhysicalBlockID, data []byte) error { - if err := bm.committedBlocks.addBlock(indexBlockID, data); err != nil { - return fmt.Errorf("unable to add to committed block cache: %v", err) - } - - return nil -} - -func (bm *Manager) fetchCommittedIndexAndMaybeCompact(ctx context.Context, compactThresold int) error { +// 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() - indexBlocks, err := bm.loadPackIndexesLocked(ctx) - if err != nil { - return err + if maxSmallBlockCount < minSmallBlockCount { + return fmt.Errorf("invalid block counts") } - if len(indexBlocks) > compactThresold { - log.Info().Msgf("compacting block indexes (block count %v exceeds threshold of %v)", len(indexBlocks), compactThresold) - return bm.compactIndexes(ctx, indexBlocks) + indexBlocks, err := bm.loadPackIndexesLocked(ctx) + if err != nil { + return fmt.Errorf("error loading indexes: %v", err) + } + + blocksToCompact := bm.getBlocksToCompact(indexBlocks, minSmallBlockCount, maxSmallBlockCount) + + if err := bm.compactAndDeleteIndexBlocks(ctx, blocksToCompact); err != nil { + log.Warn().Msgf("error performing quick compaction: %v", err) } return nil } -// CompactIndexes performs compaction of index blocks. -func (bm *Manager) CompactIndexes(ctx context.Context) error { - return bm.fetchCommittedIndexAndMaybeCompact(ctx, 1) +func (bm *Manager) getBlocksToCompact(indexBlocks []IndexInfo, minSmallBlockCount int, maxSmallBlockCount int) []IndexInfo { + var nonCompactedBlocks []IndexInfo + var totalSizeNonCompactedBlocks int64 + + var verySmallBlocks []IndexInfo + var totalSizeVerySmallBlocks int64 + + var mediumSizedBlocks []IndexInfo + var totalSizeMediumSizedBlocks int64 + + for _, b := range indexBlocks { + if b.Length > int64(bm.maxPackSize) { + continue + } + + nonCompactedBlocks = append(nonCompactedBlocks, b) + if b.Length < int64(bm.maxPackSize/20) { + verySmallBlocks = append(verySmallBlocks, b) + totalSizeVerySmallBlocks += b.Length + } else { + mediumSizedBlocks = append(mediumSizedBlocks, b) + totalSizeMediumSizedBlocks += b.Length + } + totalSizeNonCompactedBlocks += b.Length + } + + if len(nonCompactedBlocks) < minSmallBlockCount { + // current count is below min allowed - nothing to do + return nil + } + + if len(verySmallBlocks) > len(nonCompactedBlocks)/2 && len(mediumSizedBlocks)+1 < minSmallBlockCount { + return verySmallBlocks + } + + return nonCompactedBlocks } // ListBlocks returns IDs of blocks matching given prefix. @@ -563,16 +598,16 @@ func (bm *Manager) ListBlockInfos(prefix string) ([]Info, error) { return result, nil } -func (bm *Manager) compactIndexes(ctx context.Context, indexBlocks []PhysicalBlockID) error { +func (bm *Manager) compactAndDeleteIndexBlocks(ctx context.Context, indexBlocks []IndexInfo) error { if len(indexBlocks) <= 1 { - log.Printf("skipping index compaction - already compacted") return nil } + log.Debug().Msgf("compacting %v blocks", len(indexBlocks)) t0 := time.Now() bld := packindex.NewBuilder() for _, indexBlock := range indexBlocks { - data, err := bm.getPhysicalBlockInternal(ctx, indexBlock) + data, err := bm.getPhysicalBlockInternal(ctx, indexBlock.BlockID) if err != nil { return err } @@ -592,9 +627,25 @@ func (bm *Manager) compactIndexes(ctx context.Context, indexBlocks []PhysicalBlo if err := bld.Build(&buf); err != nil { return fmt.Errorf("unable to build an index: %v", err) } - compactedIndexBlock, err := bm.writePackIndexesNew(ctx, buf.Bytes(), true) - log.Info().Msgf("wrote compacted index to %v (%v bytes) in %v", compactedIndexBlock, buf.Len(), time.Since(t0)) - return err + + compactedIndexBlock, err := bm.writePackIndexesNew(ctx, buf.Bytes()) + if err != nil { + return fmt.Errorf("unable to write compacted indexes: %v", err) + } + + log.Debug().Msgf("wrote compacted index (%v bytes) in %v", compactedIndexBlock, time.Since(t0)) + + for _, indexBlock := range indexBlocks { + if indexBlock.BlockID == compactedIndexBlock { + continue + } + + if err := bm.cache.deleteBlock(ctx, indexBlock.BlockID); err != nil { + log.Warn().Msgf("unable to delete compacted block %q: %v", indexBlock.BlockID, err) + } + } + + return nil } // Flush completes writing any pending packs and writes pack indexes to the underlyign storage. @@ -653,7 +704,7 @@ func validatePrefix(prefix string) error { func (bm *Manager) writePackDataNotLocked(ctx context.Context, data []byte) (PhysicalBlockID, error) { if bm.writeFormatVersion == 0 { // 0 blocks are encrypted together - return bm.encryptAndWriteBlockNotLocked(ctx, data, "", "") + return bm.encryptAndWriteBlockNotLocked(ctx, data, "") } suffix := make([]byte, 16) @@ -672,9 +723,9 @@ func (bm *Manager) writePackDataNotLocked(ctx context.Context, data []byte) (Phy return physicalBlockID, nil } -func (bm *Manager) encryptAndWriteBlockNotLocked(ctx context.Context, data []byte, prefix string, suffix string) (PhysicalBlockID, error) { +func (bm *Manager) encryptAndWriteBlockNotLocked(ctx context.Context, data []byte, prefix string) (PhysicalBlockID, error) { hash := bm.hashData(data) - physicalBlockID := PhysicalBlockID(prefix + hex.EncodeToString(hash) + suffix) + physicalBlockID := PhysicalBlockID(prefix + hex.EncodeToString(hash)) // Encrypt the block in-place. atomic.AddInt64(&bm.stats.EncryptedBytes, int64(len(data))) @@ -898,26 +949,14 @@ type cachedList struct { // If 'full' is set to true, this function lists and returns all blocks, // if 'full' is false, the function returns only blocks from the last 2 compactions. // The list of blocks is not guaranteed to be sorted. -func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage, full bool, extraTime time.Duration) ([]IndexInfo, error) { - maxCompactions := 1 - if full { - maxCompactions = math.MaxInt32 - } - +func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage) ([]IndexInfo, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() ch := st.ListBlocks(ctx, newIndexBlockPrefix) var results []IndexInfo - numCompactions := 0 - - var timestampCutoff time.Time for it := range ch { - if !timestampCutoff.IsZero() && it.TimeStamp.Before(timestampCutoff) { - break - } - if it.Error != nil { return nil, it.Error } @@ -928,16 +967,14 @@ func listIndexBlocksFromStorage(ctx context.Context, st storage.Storage, full bo Length: it.Length, } results = append(results, ii) - - if strings.Contains(string(ii.BlockID), compactedBlockSuffix) { - numCompactions++ - if numCompactions == maxCompactions { - timestampCutoff = it.TimeStamp.Add(-extraTime) - } - } } - return results, nil + select { + case <-ctx.Done(): + return nil, ctx.Err() + default: + return results, nil + } } // NewManager creates new block manager with given packing options and a formatter. @@ -997,8 +1034,8 @@ func newManagerWithOptions(ctx context.Context, st storage.Storage, f Formatting m.startPackIndexLocked() - if err := m.fetchCommittedIndexAndMaybeCompact(ctx, autoCompactionBlockCount); err != nil { - return nil, fmt.Errorf("unable initialize indexes: %v", err) + if err := m.CompactIndexes(ctx, autoCompactionMinBlockCount, autoCompactionMaxBlockCount); err != nil { + return nil, fmt.Errorf("error initializing block manager: %v", err) } return m, nil diff --git a/block/block_manager_test.go b/block/block_manager_test.go index 025e3e37e..5359b0e23 100644 --- a/block/block_manager_test.go +++ b/block/block_manager_test.go @@ -29,7 +29,7 @@ const ( var fakeTime = time.Date(2017, 1, 1, 0, 0, 0, 0, time.UTC) func init() { - //zerolog.SetGlobalLevel(zerolog.InfoLevel) + zerolog.SetGlobalLevel(zerolog.DebugLevel) log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr}) } @@ -54,7 +54,7 @@ func TestBlockZeroBytes1(t *testing.T) { if got, want := len(data), 2; got != want { t.Errorf("unexpected number of blocks: %v, wanted %v", got, want) } - //dumpBlockManagerData(t, data) + dumpBlockManagerData(t, data) bm = newTestBlockManager(data, keyTime, nil) verifyBlock(ctx, t, bm, blockID, []byte{}) } @@ -67,7 +67,6 @@ func TestBlockZeroBytes2(t *testing.T) { writeBlockAndVerify(ctx, t, bm, seededRandomData(10, 10)) writeBlockAndVerify(ctx, t, bm, []byte{}) bm.Flush(ctx) - //dumpBlockManagerData(t, data) if got, want := len(data), 2; got != want { t.Errorf("unexpected number of blocks: %v, wanted %v", got, want) dumpBlockManagerData(t, data) @@ -166,7 +165,7 @@ func TestBlockManagerEmpty(t *testing.T) { func verifyActiveIndexBlockCount(ctx context.Context, t *testing.T, bm *Manager, expected int) { t.Helper() - blks, err := bm.ActiveIndexBlocks(ctx) + blks, err := bm.IndexBlocks(ctx) if err != nil { t.Errorf("error listing active index blocks: %v", err) return @@ -318,11 +317,11 @@ func TestBlockManagerConcurrency(t *testing.T) { t.Errorf("unexpected index count before compaction: %v, wanted %v", got, want) } - if err := bm4.CompactIndexes(ctx); err != nil { + if err := bm4.CompactIndexes(ctx, 1, 1); err != nil { t.Errorf("compaction error: %v", err) } - if got, want := getIndexCount(data), 5; got != want { - t.Errorf("unexpected index count after partial compaction: %v, wanted %v", got, want) + if got, want := getIndexCount(data), 1; got != want { + t.Errorf("unexpected index count after compaction: %v, wanted %v", got, want) } // new block manager at this point can see all data. @@ -332,7 +331,7 @@ func TestBlockManagerConcurrency(t *testing.T) { verifyBlock(ctx, t, bm5, bm1block, seededRandomData(31, 100)) verifyBlock(ctx, t, bm5, bm2block, seededRandomData(32, 100)) verifyBlock(ctx, t, bm5, bm3block, seededRandomData(33, 100)) - if err := bm5.CompactIndexes(ctx); err != nil { + if err := bm5.CompactIndexes(ctx, 1, 1); err != nil { t.Errorf("compaction error: %v", err) } } @@ -471,7 +470,7 @@ func verifyVersionCompat(t *testing.T, writeVersion int) { // make sure we can read everything verifyBlockManagerDataSet(ctx, t, mgr, dataSet) - if err := mgr.CompactIndexes(ctx); err != nil { + if err := mgr.CompactIndexes(ctx, 1, 1); err != nil { t.Fatalf("unable to compact indexes: %v", err) } if err := mgr.Flush(ctx); err != nil { @@ -508,11 +507,10 @@ func newTestBlockManager(data map[string][]byte, keyTime map[string]time.Time, t BlockFormat: "TESTONLY_MD5", MaxPackSize: maxPackSize, }, CachingOptions{}, timeFunc, 0) - bm.checkInvariantsOnUnlock = true - if err != nil { panic("can't create block manager: " + err.Error()) } + bm.checkInvariantsOnUnlock = true return bm } diff --git a/block/committed_block_index.go b/block/committed_block_index.go index ec8668d01..2832512b7 100644 --- a/block/committed_block_index.go +++ b/block/committed_block_index.go @@ -1,15 +1,21 @@ package block +import ( + "github.com/kopia/kopia/internal/packindex" +) + type committedBlockIndex interface { getBlock(blockID string) (Info, error) listBlocks(prefix string, cb func(i Info) error) error - addBlock(indexBlockID PhysicalBlockID, indexData []byte) error + addBlock(indexBlockID PhysicalBlockID, indexData []byte, use bool) error hasIndexBlockID(indexBlockID PhysicalBlockID) (bool, error) + + use(indexBlockIDs []PhysicalBlockID) error } func newCommittedBlockIndex() committedBlockIndex { return &inMemoryCommittedBlockIndex{ - physicalBlocks: make(map[PhysicalBlockID]bool), + physicalBlocks: make(map[PhysicalBlockID]packindex.Index), } } diff --git a/block/inmemory_committed_block_index.go b/block/inmemory_committed_block_index.go index 1b04a640e..06f067fd2 100644 --- a/block/inmemory_committed_block_index.go +++ b/block/inmemory_committed_block_index.go @@ -2,6 +2,7 @@ package block import ( "bytes" + "fmt" "sync" "github.com/kopia/kopia/internal/packindex" @@ -11,7 +12,7 @@ import ( type inMemoryCommittedBlockIndex struct { mu sync.Mutex indexes packindex.Merged - physicalBlocks map[PhysicalBlockID]bool + physicalBlocks map[PhysicalBlockID]packindex.Index } func (b *inMemoryCommittedBlockIndex) getBlock(blockID string) (Info, error) { @@ -31,18 +32,20 @@ func (b *inMemoryCommittedBlockIndex) getBlock(blockID string) (Info, error) { } -func (b *inMemoryCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte) error { +func (b *inMemoryCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte, use bool) error { b.mu.Lock() defer b.mu.Unlock() - ndx, err := packindex.Open(bytes.NewReader(data)) - if err != nil { - return err - } + if b.physicalBlocks[indexBlockID] == nil { + ndx, err := packindex.Open(bytes.NewReader(data)) + if err != nil { + return err + } + b.physicalBlocks[indexBlockID] = ndx - if !b.physicalBlocks[indexBlockID] { - b.physicalBlocks[indexBlockID] = true - b.indexes = append(b.indexes, ndx) + if use { + b.indexes = append(b.indexes, ndx) + } } return nil @@ -59,5 +62,27 @@ func (b *inMemoryCommittedBlockIndex) hasIndexBlockID(indexBlockID PhysicalBlock b.mu.Lock() defer b.mu.Unlock() - return b.physicalBlocks[indexBlockID], nil + return b.physicalBlocks[indexBlockID] != nil, nil +} + +func (b *inMemoryCommittedBlockIndex) use(packBlockIDs []PhysicalBlockID) error { + b.mu.Lock() + defer b.mu.Unlock() + + var newIndexes packindex.Merged + defer func() { + newIndexes.Close() //nolint:errcheck + }() + for _, e := range packBlockIDs { + ndx := b.physicalBlocks[e] + if ndx == nil { + return fmt.Errorf("unable to open pack index %q", e) + } + + //log.Printf("opened %v with %v entries", e, ndx.EntryCount()) + newIndexes = append(newIndexes, ndx) + } + b.indexes = newIndexes + newIndexes = nil + return nil } diff --git a/block/local_storage_cache.go b/block/local_storage_cache.go index 5046c7497..1ff6fe701 100644 --- a/block/local_storage_cache.go +++ b/block/local_storage_cache.go @@ -20,8 +20,7 @@ import ( const ( sweepCacheFrequency = 1 * time.Minute - fullListCacheItem = "list-full" - activeListCacheItem = "list-active" + fullListCacheItem = "list" ) type localStorageCache struct { @@ -80,30 +79,27 @@ func (c *localStorageCache) putBlock(ctx context.Context, blockID PhysicalBlockI return c.st.PutBlock(ctx, string(blockID), bytes.NewReader(data)) } -func (c *localStorageCache) listIndexBlocks(ctx context.Context, full bool, extraTime time.Duration) ([]IndexInfo, error) { - var cachedListBlockID string +func (c *localStorageCache) deleteBlock(ctx context.Context, blockID PhysicalBlockID) error { + c.deleteListCache(ctx) + return c.st.DeleteBlock(ctx, string(blockID)) +} - if full { - cachedListBlockID = fullListCacheItem - } else { - cachedListBlockID = activeListCacheItem - } - - ci, err := c.readBlocksFromCacheBlock(ctx, cachedListBlockID) +func (c *localStorageCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) { + ci, err := c.readBlocksFromCacheBlock(ctx, fullListCacheItem) if err == nil { expirationTime := ci.Timestamp.Add(c.listCacheDuration) if time.Now().Before(expirationTime) { - log.Debug().Bool("full", full).Str("blockID", cachedListBlockID).Msg("retrieved index blocks from cache") + log.Debug().Str("blockID", fullListCacheItem).Msg("retrieved index blocks from cache") return ci.Blocks, nil } } else if err != storage.ErrBlockNotFound { - log.Warn().Str("blockID", cachedListBlockID).Err(err).Msgf("unable to open cache file") + log.Warn().Err(err).Msgf("unable to open cache file") } - log.Debug().Bool("full", full).Msg("listing index blocks from source") - blocks, err := listIndexBlocksFromStorage(ctx, c.st, full, extraTime) + log.Debug().Msg("listing index blocks from source") + blocks, err := listIndexBlocksFromStorage(ctx, c.st) if err == nil { - c.saveListToCache(ctx, cachedListBlockID, &cachedList{ + c.saveListToCache(ctx, fullListCacheItem, &cachedList{ Blocks: blocks, Timestamp: time.Now(), }) @@ -229,9 +225,7 @@ func (c *localStorageCache) sweepDirectory(ctx context.Context) (err error) { } func (c *localStorageCache) deleteListCache(ctx context.Context) { - for _, it := range []string{fullListCacheItem, activeListCacheItem} { - if err := c.cacheStorage.DeleteBlock(ctx, it); err != nil && err != storage.ErrBlockNotFound { - log.Warn().Err(err).Msg("unable to delete cache item") - } + if err := c.cacheStorage.DeleteBlock(ctx, fullListCacheItem); err != nil && err != storage.ErrBlockNotFound { + log.Warn().Err(err).Msg("unable to delete cache item") } } diff --git a/block/null_block_cache.go b/block/null_block_cache.go index edf7f5dd0..d6aa8d30a 100644 --- a/block/null_block_cache.go +++ b/block/null_block_cache.go @@ -3,7 +3,6 @@ package block import ( "bytes" "context" - "time" "github.com/kopia/kopia/storage" ) @@ -20,8 +19,15 @@ func (c nullBlockCache) putBlock(ctx context.Context, blockID PhysicalBlockID, d return c.st.PutBlock(ctx, string(blockID), bytes.NewReader(data)) } -func (c nullBlockCache) listIndexBlocks(ctx context.Context, full bool, extraTime time.Duration) ([]IndexInfo, error) { - return listIndexBlocksFromStorage(ctx, c.st, full, extraTime) +func (c nullBlockCache) listIndexBlocks(ctx context.Context) ([]IndexInfo, error) { + return listIndexBlocksFromStorage(ctx, c.st) +} + +func (c nullBlockCache) deleteListCache(ctx context.Context) { +} + +func (c nullBlockCache) deleteBlock(ctx context.Context, blockID PhysicalBlockID) error { + return c.st.DeleteBlock(ctx, string(blockID)) } func (c nullBlockCache) close() error { diff --git a/block/simple_committed_block_index.go b/block/simple_committed_block_index.go index dd7eb3d9b..211fcdd28 100644 --- a/block/simple_committed_block_index.go +++ b/block/simple_committed_block_index.go @@ -5,7 +5,6 @@ import ( "io/ioutil" "os" "path/filepath" - "strings" "sync" "github.com/kopia/kopia/internal/packindex" @@ -17,13 +16,17 @@ import ( const simpleIndexSuffix = ".sndx" type simpleCommittedBlockIndex struct { - dirname string - indexesMutex sync.Mutex - indexBlocks map[PhysicalBlockID]bool - merged packindex.Merged + dirname string + + mu sync.Mutex + indexBlocks map[PhysicalBlockID]bool + merged packindex.Merged } func (b *simpleCommittedBlockIndex) getBlock(blockID string) (Info, error) { + b.mu.Lock() + defer b.mu.Unlock() + info, err := b.merged.GetInfo(blockID) if info != nil { return *info, nil @@ -35,75 +38,44 @@ func (b *simpleCommittedBlockIndex) getBlock(blockID string) (Info, error) { } func (b *simpleCommittedBlockIndex) hasIndexBlockID(indexBlockID PhysicalBlockID) (bool, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.indexBlocks[indexBlockID], nil } -func (b *simpleCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte) error { +func (b *simpleCommittedBlockIndex) addBlock(indexBlockID PhysicalBlockID, data []byte, use bool) error { fullPath := filepath.Join(b.dirname, string(indexBlockID+simpleIndexSuffix)) if err := ioutil.WriteFile(fullPath, data, 0600); err != nil { return err } - b.indexesMutex.Lock() - defer b.indexesMutex.Unlock() + b.mu.Lock() + defer b.mu.Unlock() - if !b.indexBlocks[indexBlockID] { - ndx, err := b.openIndex(fullPath) - if err != nil { - return fmt.Errorf("unable to open pack index %q: %v", fullPath, err) - } + b.indexBlocks[indexBlockID] = true - b.indexBlocks[indexBlockID] = true - b.merged = append(b.merged, ndx) + if !use { + return nil } + ndx, err := b.openIndex(fullPath) + if err != nil { + return fmt.Errorf("unable to open pack index %q: %v", fullPath, err) + } + + b.merged = append(b.merged, ndx) + return nil } func (b *simpleCommittedBlockIndex) listBlocks(prefix string, cb func(i Info) error) error { - return b.merged.Iterate(prefix, cb) -} + b.mu.Lock() + m := b.merged + b.mu.Unlock() -func (b *simpleCommittedBlockIndex) loadIndexes() error { - b.indexesMutex.Lock() - defer b.indexesMutex.Unlock() - - entries, err := ioutil.ReadDir(b.dirname) - if err != nil { - return err - } - - newIndexes := map[PhysicalBlockID]bool{} - var newMerged packindex.Merged - defer func() { - newMerged.Close() //nolint:errcheck - }() - for _, e := range entries { - if !strings.HasSuffix(e.Name(), simpleIndexSuffix) { - continue - } - - fullpath := filepath.Join(b.dirname, e.Name()) - ndx, err := b.openIndex(fullpath) - if err != nil { - return fmt.Errorf("unable to open pack index %q: %v", fullpath, err) - } - - log.Printf("opened %v with %v entries", fullpath, ndx.EntryCount()) - - // ndx.Iterate("", func(i Info) error { - // log.Info().Msgf("i: %v blk:%v off:%v len:%v", i.BlockID, i.PackBlockID, i.PackOffset, i.Length) - // return nil - // }) - - newIndexes[PhysicalBlockID(strings.TrimSuffix(e.Name(), simpleIndexSuffix))] = true - newMerged = append(newMerged, ndx) - } - b.indexBlocks = newIndexes - b.merged = newMerged - newMerged = nil - return nil + return m.Iterate(prefix, cb) } func (b *simpleCommittedBlockIndex) openIndex(fullpath string) (packindex.Index, error) { @@ -116,12 +88,38 @@ func (b *simpleCommittedBlockIndex) openIndex(fullpath string) (packindex.Index, return packindex.Open(f) } +func (b *simpleCommittedBlockIndex) use(packBlockIDs []PhysicalBlockID) error { + b.mu.Lock() + defer b.mu.Unlock() + + newIndexes := map[PhysicalBlockID]bool{} + var newMerged packindex.Merged + defer func() { + newMerged.Close() //nolint:errcheck + }() + for _, e := range packBlockIDs { + fullpath := filepath.Join(b.dirname, string(e)+simpleIndexSuffix) + ndx, err := b.openIndex(fullpath) + if err != nil { + return fmt.Errorf("unable to open pack index %q: %v", fullpath, err) + } + + log.Printf("opened %v with %v entries", fullpath, ndx.EntryCount()) + newIndexes[e] = true + newMerged = append(newMerged, ndx) + } + b.indexBlocks = newIndexes + b.merged = newMerged + newMerged = nil + return nil +} + func newSimpleCommittedBlockIndex(dirname string) (committedBlockIndex, error) { _ = os.MkdirAll(dirname, 0700) - s := &simpleCommittedBlockIndex{dirname: dirname} - if err := s.loadIndexes(); err != nil { - return nil, err + s := &simpleCommittedBlockIndex{ + dirname: dirname, + indexBlocks: map[PhysicalBlockID]bool{}, } return s, nil } diff --git a/cli/command_block_index_list.go b/cli/command_block_index_list.go index 32c0700b0..c474d479b 100644 --- a/cli/command_block_index_list.go +++ b/cli/command_block_index_list.go @@ -4,25 +4,16 @@ import ( "context" "fmt" - "github.com/kopia/kopia/block" "github.com/kopia/kopia/repo" ) var ( blockIndexListCommand = blockIndexCommands.Command("list", "List block indexes").Alias("ls").Default() - blockIndexListAll = blockIndexListCommand.Flag("all", "List all blocks, not just active ones").Short('a').Bool() blockIndexListSummary = blockIndexListCommand.Flag("summary", "Display block summary").Bool() ) func runListBlockIndexesAction(ctx context.Context, rep *repo.Repository) error { - var blks []block.IndexInfo - var err error - - if !*blockIndexListAll { - blks, err = rep.Blocks.ActiveIndexBlocks(ctx) - } else { - blks, err = rep.Blocks.ListIndexBlocks(ctx) - } + blks, err := rep.Blocks.IndexBlocks(ctx) if err != nil { return err diff --git a/cli/command_block_index_optimize.go b/cli/command_block_index_optimize.go index f11a456cd..f314fd9dc 100644 --- a/cli/command_block_index_optimize.go +++ b/cli/command_block_index_optimize.go @@ -7,11 +7,13 @@ import ( ) var ( - optimizeCommand = blockIndexCommands.Command("optimize", "Optimize block indexes.") + optimizeCommand = blockIndexCommands.Command("optimize", "Optimize block indexes.") + optimizeMinSmallBlocks = blockIndexCommands.Flag("min-small-blocks", "Minimum number of small blocks that can be left after compaction.").Default("1").Int() + optimizeMaxSmallBlocks = blockIndexCommands.Flag("max-small-blocks", "Maximum number of small blocks that can be left after compaction.").Default("1").Int() ) func runOptimizeCommand(ctx context.Context, rep *repo.Repository) error { - return rep.Blocks.CompactIndexes(ctx) + return rep.Blocks.CompactIndexes(ctx, *optimizeMinSmallBlocks, *optimizeMaxSmallBlocks) } func init() { diff --git a/cli/command_block_index_show.go b/cli/command_block_index_show.go index 5e5689ffe..221098ec5 100644 --- a/cli/command_block_index_show.go +++ b/cli/command_block_index_show.go @@ -26,7 +26,7 @@ func getIndexBlocksToShow(ctx context.Context, rep *repo.Repository) ([]block.Ph } if len(blockIDs) == 1 && blockIDs[0] == "active" { - b, err := rep.Blocks.ActiveIndexBlocks(ctx) + b, err := rep.Blocks.IndexBlocks(ctx) if err != nil { return nil, err } diff --git a/internal/packindex/packindex_test.go b/internal/packindex/packindex_test.go index 7f3b8ea69..4ac2a0bc9 100644 --- a/internal/packindex/packindex_test.go +++ b/internal/packindex/packindex_test.go @@ -3,6 +3,7 @@ package packindex_test import ( "bytes" "crypto/sha1" + "encoding/hex" "fmt" "math/rand" "reflect" @@ -13,8 +14,6 @@ import ( ) func TestPackIndex(t *testing.T) { - b := packindex.NewBuilder() - blockNumber := 0 deterministicBlockID := func(prefix string, id int) string { @@ -96,19 +95,41 @@ func TestPackIndex(t *testing.T) { } infoMap := map[string]packindex.Info{} + b1 := packindex.NewBuilder() + b2 := packindex.NewBuilder() + b3 := packindex.NewBuilder() for _, info := range infos { infoMap[info.BlockID] = info - b.Add(info) + b1.Add(info) + b2.Add(info) + b3.Add(info) } - var buf bytes.Buffer - if err := b.Build(&buf); err != nil { + var buf1 bytes.Buffer + var buf2 bytes.Buffer + var buf3 bytes.Buffer + if err := b1.Build(&buf1); err != nil { t.Errorf("unable to build: %v", err) } + if err := b1.Build(&buf2); err != nil { + t.Errorf("unable to build: %v", err) + } + if err := b1.Build(&buf3); err != nil { + t.Errorf("unable to build: %v", err) + } + data1 := buf1.Bytes() + data2 := buf2.Bytes() + data3 := buf3.Bytes() - data := buf.Bytes() - ndx, err := packindex.Open(bytes.NewReader(data)) + if !reflect.DeepEqual(data1, data2) { + t.Errorf("builder output not stable: %x vs %x", hex.Dump(data1), hex.Dump(data2)) + } + if !reflect.DeepEqual(data2, data3) { + t.Errorf("builder output not stable: %x vs %x", hex.Dump(data2), hex.Dump(data3)) + } + + ndx, err := packindex.Open(bytes.NewReader(data1)) if err != nil { t.Fatalf("can't open index: %v", err) } diff --git a/repo/repository_test.go b/repo/repository_test.go index fa03431d3..b770b2952 100644 --- a/repo/repository_test.go +++ b/repo/repository_test.go @@ -186,7 +186,7 @@ func TestPackingSimple(t *testing.T) { verify(ctx, t, repo, oid2a, []byte(content2), "packed-object-2") verify(ctx, t, repo, oid3a, []byte(content3), "packed-object-3") - if err := repo.Blocks.CompactIndexes(ctx); err != nil { + if err := repo.Blocks.CompactIndexes(ctx, 1, 1); err != nil { t.Errorf("optimize error: %v", err) } data, _, repo = setupTestWithData(t, data, keyTime, func(n *NewRepositoryOptions) { @@ -196,7 +196,7 @@ func TestPackingSimple(t *testing.T) { verify(ctx, t, repo, oid2a, []byte(content2), "packed-object-2") verify(ctx, t, repo, oid3a, []byte(content3), "packed-object-3") - if err := repo.Blocks.CompactIndexes(ctx); err != nil { + if err := repo.Blocks.CompactIndexes(ctx, 1, 1); err != nil { t.Errorf("optimize error: %v", err) } _, _, repo = setupTestWithData(t, data, keyTime, func(n *NewRepositoryOptions) { diff --git a/tests/end_to_end_test/end_to_end_test.go b/tests/end_to_end_test/end_to_end_test.go index c793f170a..a6c6f7da4 100644 --- a/tests/end_to_end_test/end_to_end_test.go +++ b/tests/end_to_end_test/end_to_end_test.go @@ -112,11 +112,10 @@ func TestEndToEnd(t *testing.T) { // expect 5 blocks, each snapshot creation adds one index block. e.runAndVerifyOutputLineCount(t, 5, "blockindex", "ls") e.runAndExpectSuccess(t, "blockindex", "optimize") - e.runAndVerifyOutputLineCount(t, 6, "blockindex", "ls") - e.runAndVerifyOutputLineCount(t, 6, "blockindex", "ls", "--all") + e.runAndVerifyOutputLineCount(t, 1, "blockindex", "ls") e.runAndExpectSuccess(t, "snapshot", "create", ".", dir1, dir2) - e.runAndVerifyOutputLineCount(t, 7, "blockindex", "ls") + e.runAndVerifyOutputLineCount(t, 2, "blockindex", "ls") } func (e *testenv) runAndExpectSuccess(t *testing.T, args ...string) []string { diff --git a/tests/stress_test/stress_test.go b/tests/stress_test/stress_test.go index 35e4f038b..d592e90a5 100644 --- a/tests/stress_test/stress_test.go +++ b/tests/stress_test/stress_test.go @@ -16,7 +16,7 @@ import ( "github.com/kopia/kopia/storage" ) -const goroutineCount = 8 +const goroutineCount = 16 func TestStressBlockManager(t *testing.T) { data := map[string][]byte{}