From 0f63f04c142d8e94ec5ab6af739cb4e8f710fdb0 Mon Sep 17 00:00:00 2001 From: Jarek Kowalski Date: Sat, 21 Apr 2018 12:42:33 -0700 Subject: [PATCH] changed how pack block are staged to first add them as inline, then convert to a block as necessary and not the opposite --- block/block_manager.go | 63 ++++++++++++++++++------------------- block/block_manager_test.go | 2 +- block/pack_index.go | 4 +-- block/pack_index_test.go | 55 ++++++++++++++++++++++---------- block/pack_index_v1.go | 16 ++++++---- repo/repository_test.go | 2 +- 6 files changed, 83 insertions(+), 59 deletions(-) diff --git a/block/block_manager.go b/block/block_manager.go index da4a1b1c8..aba9ee291 100644 --- a/block/block_manager.go +++ b/block/block_manager.go @@ -74,8 +74,8 @@ type Manager struct { blockIDToIndex map[ContentID]packIndex // maps block ID to corresponding index packBlockIDToIndex map[PhysicalBlockID]packIndex // maps pack block ID to corresponding index - currentPackData []byte // data for the current block - currentPackIndex packIndexBuilder // index of a current block + currentPackDataLength int // length of the current block + currentPackIndex packIndexBuilder // index of a current block pendingPackIndexes []packIndex // pending indexes of blocks that have been saved. flushPackIndexesAfter time.Time // time when those indexes should be flushed @@ -118,21 +118,10 @@ func (bm *Manager) DeleteBlock(blockID ContentID) error { func (bm *Manager) addToPackLocked(ctx context.Context, blockID ContentID, data []byte) error { bm.assertLocked() - if len(bm.currentPackData) == 0 && bm.maxPreambleLength > 0 { - preambleLength := rand.Intn(bm.maxPreambleLength-bm.minPreambleLength+1) + bm.minPreambleLength - preamble := make([]byte, preambleLength, preambleLength+len(data)) - if _, err := io.ReadFull(cryptorand.Reader, preamble); err != nil { - return err - } + bm.currentPackDataLength += len(data) + shouldFinish := bm.currentPackDataLength >= bm.maxPackSize - bm.currentPackData = preamble - } - - offset := len(bm.currentPackData) - shouldFinish := offset+len(data) >= bm.maxPackSize - bm.currentPackData = append(bm.currentPackData, data...) - - bm.currentPackIndex.addPackedBlock(blockID, uint32(offset), uint32(len(data))) + bm.currentPackIndex.addInlineBlock(blockID, data) bm.blockIDToIndex[blockID] = bm.currentPackIndex if shouldFinish { @@ -224,7 +213,7 @@ func (bm *Manager) invariantViolated(msg string, args ...interface{}) { func (bm *Manager) startPackIndexLocked() { bm.currentPackIndex = newPackIndexV1(bm.timeNow()) - bm.currentPackData = []byte{} + bm.currentPackDataLength = 0 } func (bm *Manager) flushPackIndexesLocked(ctx context.Context) error { @@ -265,32 +254,46 @@ func (bm *Manager) writePackIndexes(ctx context.Context, ndx []packIndex, isComp func (bm *Manager) finishPackLocked(ctx context.Context) error { if !isIndexEmpty(bm.currentPackIndex) { log.Debug().Msg("finishing pack") - if len(bm.currentPackData) < bm.maxInlineContentLength { - bm.currentPackIndex.packedToInline(bm.currentPackData) - bm.currentPackData = nil - } - if bm.currentPackData != nil { + + if bm.currentPackDataLength > 0 && bm.currentPackDataLength > bm.maxInlineContentLength { + // repack inline blocks as a stored block + var blockData []byte + + preambleLength := rand.Intn(bm.maxPreambleLength-bm.minPreambleLength+1) + bm.minPreambleLength + preamble := make([]byte, preambleLength) + if _, err := io.ReadFull(cryptorand.Reader, preamble); err != nil { + return err + } + + blockData = append(blockData, preamble...) + + items := bm.currentPackIndex.clearInlineBlocks() + for blockID, data := range items { + offset := uint32(len(blockData)) + size := uint32(len(data)) + blockData = append(blockData, data...) + bm.currentPackIndex.addPackedBlock(blockID, offset, size) + } + if bm.paddingUnit > 0 { - if missing := bm.paddingUnit - (len(bm.currentPackData) % bm.paddingUnit); missing > 0 { + if missing := bm.paddingUnit - (len(blockData) % bm.paddingUnit); missing > 0 { postamble := make([]byte, missing) if _, err := io.ReadFull(cryptorand.Reader, postamble); err != nil { return fmt.Errorf("can't allocate random bytes for postamble: %v", err) } - bm.currentPackData = append(bm.currentPackData, postamble...) + blockData = append(blockData, postamble...) } } - packBlockID, err := bm.writeUnpackedBlockNotLocked(ctx, bm.currentPackData, "", "") + packBlockID, err := bm.writeUnpackedBlockNotLocked(ctx, blockData, "", "") if err != nil { return fmt.Errorf("can't save pack data block %q: %v", packBlockID, err) } - bm.currentPackIndex.finishPack(packBlockID, uint64(len(bm.currentPackData)), bm.writeFormatVersion) + bm.currentPackIndex.finishPack(packBlockID, uint64(len(blockData)), bm.writeFormatVersion) bm.packBlockIDToIndex[packBlockID] = bm.currentPackIndex } bm.pendingPackIndexes = append(bm.pendingPackIndexes, bm.currentPackIndex) - } else { - log.Printf("nothing to write - pack is empty") } bm.startPackIndexLocked() @@ -681,10 +684,6 @@ func (bm *Manager) getPendingBlockLocked(blockID ContentID) ([]byte, error) { if bi.payload != nil { return bi.payload, nil } - - if bm.currentPackData != nil { - return bm.currentPackData[bi.offset : bi.offset+bi.size], nil - } } } diff --git a/block/block_manager_test.go b/block/block_manager_test.go index 627adc551..239f0b50b 100644 --- a/block/block_manager_test.go +++ b/block/block_manager_test.go @@ -50,7 +50,7 @@ func TestBlockZeroBytes1(t *testing.T) { bm := newTestBlockManager(data, keyTime, nil) writeBlockAndVerify(ctx, t, bm, []byte{}) bm.Flush(ctx) - if got, want := len(data), 2; got != want { + if got, want := len(data), 1; got != want { t.Errorf("unexpected number of blocks: %v, wanted %v", got, want) } dumpBlockManagerData(data) diff --git a/block/pack_index.go b/block/pack_index.go index 883833b4d..3ad1866b2 100644 --- a/block/pack_index.go +++ b/block/pack_index.go @@ -27,10 +27,10 @@ type packBlockInfo struct { type packIndexBuilder interface { packIndex + addInlineBlock(blockID ContentID, data []byte) addPackedBlock(blockID ContentID, offset, size uint32) + clearInlineBlocks() map[ContentID][]byte deleteBlock(blockID ContentID) - - packedToInline(data []byte) finishPack(packBlockID PhysicalBlockID, packLength uint64, formatVersion int32) } diff --git a/block/pack_index_test.go b/block/pack_index_test.go index 6f2de211b..69b055b6f 100644 --- a/block/pack_index_test.go +++ b/block/pack_index_test.go @@ -27,14 +27,20 @@ func verifyPackIndex(t *testing.T, ndx packIndexBuilder, ts time.Time) { t.Errorf("unexpected created time: %v, wanted %v", ndx.createTimeNanos(), ts.UnixNano()) } - blockIDs := []ContentID{ + packedBlockIDs := []ContentID{ "xabcdef", "abcdef", "0123456780abcdef0123456780abcdef0123456780abcdef0123456780abcdef", "x0123456780abcdef0123456780abcdef0123456780abcdef0123456780abcdef", } - var data []byte + inlineBlockIDs := []ContentID{ + "xabcdefaa", + "abcdefaa", + "0123456780abcdef0123456780abcdef0123456780abcdef0123456780abcdefaa", + "x0123456780abcdef0123456780abcdef0123456780abcdef0123456780abcdefaa", + } + var offset uint32 var blockData = map[ContentID][]byte{} @@ -42,13 +48,10 @@ func verifyPackIndex(t *testing.T, ndx packIndexBuilder, ts time.Time) { var blockSizes = map[ContentID]uint32{} // add blocks to pack index - for _, blockID := range blockIDs { + for _, blockID := range packedBlockIDs { blkSize := uint32(rand.Intn(100) + 1) - blockContent := make([]byte, blkSize) - blockData[blockID] = blockContent blockSizes[blockID] = blkSize blockOffsets[blockID] = offset - data = append(data, blockContent...) verifyIndexBlockNotFound(t, ndx, blockID) ndx.addPackedBlock(blockID, offset, blkSize) @@ -56,29 +59,47 @@ func verifyPackIndex(t *testing.T, ndx packIndexBuilder, ts time.Time) { offset += uint32(blkSize) } + // add some inline blocks to pack index + for _, blockID := range inlineBlockIDs { + blkSize := uint32(rand.Intn(100) + 1) + blockContent := make([]byte, blkSize) + rand.Read(blockContent) + blockData[blockID] = blockContent + + verifyIndexBlockNotFound(t, ndx, blockID) + ndx.addInlineBlock(blockID, blockContent) + verifyIndexBlockInline(t, ndx, blockID, blockContent) + } + // verify offsets and sizes again - for _, blockID := range blockIDs { + for _, blockID := range packedBlockIDs { verifyIndexBlockFoundPacked(t, ndx, blockID, blockOffsets[blockID], blockSizes[blockID]) } - ndx.packedToInline(data) - for _, blockID := range blockIDs { + for _, blockID := range inlineBlockIDs { verifyIndexBlockInline(t, ndx, blockID, blockData[blockID]) } - ndx.deleteBlock(blockIDs[0]) - verifyIndexBlockNotFound(t, ndx, blockIDs[0]) - verifyIndexBlockDeleted(t, ndx, blockIDs[0]) + ndx.deleteBlock(packedBlockIDs[0]) + verifyIndexBlockNotFound(t, ndx, packedBlockIDs[0]) + verifyIndexBlockDeleted(t, ndx, packedBlockIDs[0]) - ndx.deleteBlock(blockIDs[1]) - verifyIndexBlockNotFound(t, ndx, blockIDs[1]) - verifyIndexBlockDeleted(t, ndx, blockIDs[1]) + inline := ndx.clearInlineBlocks() + if !reflect.DeepEqual(inline, blockData) { + t.Errorf("unexpected result of clearInlineBlocks(): %v, wanted %v", inline, blockData) + } - ndx.finishPack("some-physical-block", uint64(len(data)), 77) + for k := range inline { + verifyIndexBlockNotFound(t, ndx, k) + } + + someLength := uint64(1234) + + ndx.finishPack("some-physical-block", someLength, 77) if got, want := ndx.packBlockID(), PhysicalBlockID("some-physical-block"); got != want { t.Errorf("unexpected pack block ID: %q, wanted %q", got, want) } - if got, want := ndx.packLength(), uint64(len(data)); got != want { + if got, want := ndx.packLength(), someLength; got != want { t.Errorf("unexpected pack length: %v, wanted %v", got, want) } if got, want := ndx.formatVersion(), int32(77); got != want { diff --git a/block/pack_index_v1.go b/block/pack_index_v1.go index 4f101e5d1..adab364fc 100644 --- a/block/pack_index_v1.go +++ b/block/pack_index_v1.go @@ -30,13 +30,13 @@ func (p protoPackIndexV1) finishPack(packBlockID PhysicalBlockID, packLength uin p.ndx.FormatVersion = formatVersion } -func (p protoPackIndexV1) packedToInline(packedData []byte) { - for k, os := range p.ndx.Items { - offset, size := unpackOffsetAndSize(os) - p.ndx.InlineItems[k] = packedData[offset : offset+size] +func (p protoPackIndexV1) clearInlineBlocks() map[ContentID][]byte { + result := map[ContentID][]byte{} + for k, b := range p.ndx.InlineItems { + result[ContentID(k)] = b } - - p.ndx.Items = map[string]uint64{} + p.ndx.InlineItems = map[string][]byte{} + return result } func (p protoPackIndexV1) getBlock(blockID ContentID) (packBlockInfo, bool) { @@ -87,6 +87,10 @@ func (p protoPackIndexV1) packLength() uint64 { return p.ndx.PackLength } +func (p protoPackIndexV1) addInlineBlock(blockID ContentID, data []byte) { + p.ndx.InlineItems[string(blockID)] = append([]byte{}, data...) +} + func (p protoPackIndexV1) addPackedBlock(blockID ContentID, offset, size uint32) { p.ndx.Items[string(blockID)] = packOffsetAndSize(offset, size) } diff --git a/repo/repository_test.go b/repo/repository_test.go index afb34e72b..c0ce37bf2 100644 --- a/repo/repository_test.go +++ b/repo/repository_test.go @@ -93,7 +93,7 @@ func TestWriters(t *testing.T) { if got, want := len(data), 2; got != want { // 1 format block + 1 pack index block (including inline data blocks) - t.Errorf("unexpected data written to the storage (%v), wanted %v: %v", got, want, data) + t.Errorf("unexpected data written to the storage (%v), wanted %v", got, want) dumpBlockManagerData(data) } }