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

@@ -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
-			}
 		}
 	}
 

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)

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)
 }
 

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 {

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)
 }

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)
 		}
 	}