Add filepool files with backing content

Filepool files with backing content allows the use of read only
datasources as initial content for a filepool backed file. It is
implemented with efficient copy on write semantics which will copy any
sector from the backing store upon writes.

Author: meroton-benjamin

pkg/builder/file_pool_stats_build_executor.go

@@ -13,6 +13,8 @@ import (
 	"github.com/buildbarn/bb-storage/pkg/filesystem"
 	"github.com/buildbarn/bb-storage/pkg/util"
 
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/anypb"
 )
 
@@ -77,6 +79,10 @@ func (fp *statsCollectingFilePool) NewFile() (filesystem.FileReadWriter, error)
 	}, nil
 }
 
+func (fp *statsCollectingFilePool) NewFileWithReadLayer(readLayer filepool.ReadLayer, sizeBytes uint64) (filesystem.FileReadWriter, error) {
+	return nil, status.Error(codes.Internal, "Not implemented")
+}
+
 // statsCollectingFileReadWriter is a decorator for
 // filesystem.FileReadWriter that measures the number of file operations
 // performed.

pkg/filesystem/filepool/BUILD.bazel

@@ -9,6 +9,7 @@ go_library(
         "file_pool.go",
         "metrics_file_pool.go",
         "quota_enforcing_file_pool.go",
+        "read_layer.go",
     ],
     importpath = "github.com/buildbarn/bb-remote-execution/pkg/filesystem/filepool",
     visibility = ["//visibility:public"],
@@ -31,6 +32,7 @@ go_test(
         "block_device_backed_file_pool_test.go",
         "empty_file_pool_test.go",
         "quota_enforcing_file_pool_test.go",
+        "read_layer_test.go",
     ],
     deps = [
         ":filepool",

pkg/filesystem/filepool/block_device_backed_file_pool.go

@@ -33,14 +33,35 @@ func NewBlockDeviceBackedFilePool(blockDevice blockdevice.BlockDevice, sectorAll
 	}
 }
 
+type zeroReaderAt struct{}
+
+func (zeroReaderAt) ReadAt(p []byte, off int64) (int, error) {
+	for i := range p {
+		p[i] = 0
+	}
+	return len(p), nil
+}
+
 func (fp *blockDeviceBackedFilePool) NewFile() (filesystem.FileReadWriter, error) {
-	return &blockDeviceBackedFile{
+	readLayer := NewReadLayer(&zeroReaderAt{}, 0)
+	return fp.NewFileWithReadLayer(readLayer, 0)
+}
+
+func (fp *blockDeviceBackedFilePool) NewFileWithReadLayer(readLayer ReadLayer, sizeBytes uint64) (filesystem.FileReadWriter, error) {
+	file := &blockDeviceBackedFile{
 		fp: fp,
-	}, nil
+		rl: readLayer,
+	}
+	err := file.Truncate(int64(sizeBytes))
+	if err != nil {
+		return nil, err
+	}
+	return file, nil
 }
 
 type blockDeviceBackedFile struct {
 	fp        *blockDeviceBackedFilePool
+	rl        ReadLayer
 	sizeBytes uint64
 	sectors   []uint32
 }
@@ -172,23 +193,17 @@ func (f *blockDeviceBackedFile) GetNextRegionOffset(off int64, regionType filesy
 func (f *blockDeviceBackedFile) readFromSectors(p []byte, sectorIndex, lastSectorIndex, offsetWithinSector int) (int, error) {
 	if sectorIndex >= len(f.sectors) {
 		// Attempted to read from a hole located at the
-		// end of the file. Fill up all of the remaining
-		// space with zero bytes.
-		for i := 0; i < len(p); i++ {
-			p[i] = 0
-		}
-		return len(p), nil
+		// end of the file. Delegate to ReadLayer.
+		offset := f.fp.sectorSizeBytes*sectorIndex + offsetWithinSector
+		return f.rl.ReadAt(p, int64(offset))
 	}
 
 	sector, sectorsToRead := f.getSectorsContiguous(sectorIndex, lastSectorIndex)
 	p = f.limitBufferToSectorBoundary(p, sectorsToRead, offsetWithinSector)
 	if sector == 0 {
 		// Attempted to read from a sparse region of the file.
-		// Fill in zero bytes.
-		for i := 0; i < len(p); i++ {
-			p[i] = 0
-		}
-		return len(p), nil
+		offset := f.fp.sectorSizeBytes*sectorIndex + offsetWithinSector
+		return f.rl.ReadAt(p, int64(offset))
 	}
 
 	// Attempted to read from a region of the file that contains
@@ -267,6 +282,9 @@ func (f *blockDeviceBackedFile) Truncate(size int64) error {
 	if size < 0 {
 		return status.Errorf(codes.InvalidArgument, "Negative truncation size: %d", size)
 	}
+	if err := f.rl.Truncate(size); err != nil {
+		return fmt.Errorf("read layer truncate failed: %w", err)
+	}
 
 	sectorIndex := int(size / int64(f.fp.sectorSizeBytes))
 	offsetWithinSector := int(size % int64(f.fp.sectorSizeBytes))
@@ -299,7 +317,7 @@ func (f *blockDeviceBackedFile) Truncate(size int64) error {
 // writeToNewSectors is used to write data into new sectors. This
 // function is called when holes in a sparse file are filled up or when
 // data is appended to the end of a file.
-func (f *blockDeviceBackedFile) writeToNewSectors(p []byte, offsetWithinSector int) (int, uint32, int, error) {
+func (f *blockDeviceBackedFile) writeToNewSectors(p []byte, fromSector int, offsetWithinSector int) (int, uint32, int, error) {
 	// Allocate space to store the data.
 	sectorsToAllocate := int((uint64(offsetWithinSector) + uint64(len(p)) + uint64(f.fp.sectorSizeBytes) - 1) / uint64(f.fp.sectorSizeBytes))
 	firstSector, sectorsAllocated, err := f.fp.sectorAllocator.AllocateContiguous(sectorsToAllocate)
@@ -314,10 +332,15 @@ func (f *blockDeviceBackedFile) writeToNewSectors(p []byte, offsetWithinSector i
 	nWritten := len(p)
 
 	// Write the first sector separately when we need to introduce
-	// leading zero padding.
+	// leading read layer padding.
 	sector := firstSector
 	if offsetWithinSector > 0 {
 		buf := make([]byte, f.fp.sectorSizeBytes)
+		logicalOffset := fromSector * f.fp.sectorSizeBytes
+		if _, err := f.rl.ReadAt(buf[:offsetWithinSector], int64(logicalOffset)); err != nil {
+			f.fp.sectorAllocator.FreeContiguous(firstSector, sectorsAllocated)
+			return 0, 0, 0, err
+		}
 		nWritten := copy(buf[offsetWithinSector:], p)
 		if _, err := f.fp.blockDevice.WriteAt(buf, f.toDeviceOffset(sector, 0)); err != nil {
 			f.fp.sectorAllocator.FreeContiguous(firstSector, sectorsAllocated)
@@ -340,9 +363,14 @@ func (f *blockDeviceBackedFile) writeToNewSectors(p []byte, offsetWithinSector i
 	}
 
 	// Write the last sector separately when we need to introduce
-	// trailing zero padding.
+	// trailing read layer padding.
 	if len(p) > 0 {
 		buf := make([]byte, f.fp.sectorSizeBytes)
+		logicalOffset := uint32(len(p)) + (sector-firstSector)*uint32(f.fp.sectorSizeBytes)
+		if _, err := f.rl.ReadAt(buf[len(p):], int64(logicalOffset)); err != nil {
+			f.fp.sectorAllocator.FreeContiguous(firstSector, sectorsAllocated)
+			return 0, 0, 0, err
+		}
 		copy(buf, p)
 		if _, err := f.fp.blockDevice.WriteAt(buf, f.toDeviceOffset(sector, 0)); err != nil {
 			f.fp.sectorAllocator.FreeContiguous(firstSector, sectorsAllocated)
@@ -375,7 +403,7 @@ func (f *blockDeviceBackedFile) writeToSectors(p []byte, sectorIndex, lastSector
 		// Attempted to write past the end-of-file or within a
 		// hole located at the end of a sparse file. Allocate
 		// space and grow the file.
-		bytesWritten, firstSector, sectorsAllocated, err := f.writeToNewSectors(p, offsetWithinSector)
+		bytesWritten, firstSector, sectorsAllocated, err := f.writeToNewSectors(p, sectorIndex, offsetWithinSector)
 		if err != nil {
 			return 0, err
 		}
@@ -389,7 +417,7 @@ func (f *blockDeviceBackedFile) writeToSectors(p []byte, sectorIndex, lastSector
 	if sector == 0 {
 		// Attempted to write to a hole within a sparse file.
 		// Allocate space and insert sectors into the file.
-		bytesWritten, firstSector, sectorsAllocated, err := f.writeToNewSectors(p, offsetWithinSector)
+		bytesWritten, firstSector, sectorsAllocated, err := f.writeToNewSectors(p, sectorIndex, offsetWithinSector)
 		if err != nil {
 			return 0, err
 		}

pkg/filesystem/filepool/block_device_backed_file_pool_test.go

@@ -3,6 +3,7 @@ package filepool_test
 import (
 	"io"
 	"math"
+	"strings"
 	"testing"
 
 	"github.com/buildbarn/bb-remote-execution/internal/mock"
@@ -319,4 +320,105 @@ func TestBlockDeviceBackedFilePool(t *testing.T) {
 		_, err = f.WriteAt([]byte{0}, -1)
 		testutil.RequireEqualStatus(t, status.Error(codes.InvalidArgument, "Negative write offset: -1"), err)
 	})
+
+	t.Run("ReadFromHoleReturnsBackingData", func(t *testing.T) {
+		rl := filepool.NewReadLayer(strings.NewReader("HelloWorld"), 10)
+		file, err := pool.NewFileWithReadLayer(rl, 10)
+		require.NoError(t, err)
+
+		buf := make([]byte, 10)
+		n, err := file.ReadAt(buf, 0)
+		require.Equal(t, err, io.EOF)
+		require.Equal(t, 10, n)
+		require.Equal(t, []byte("HelloWorld"), buf)
+
+		require.NoError(t, file.Close())
+	})
+
+	t.Run("ReadFromHoleBeyondBackingReturnsZeroes", func(t *testing.T) {
+		rl := filepool.NewReadLayer(strings.NewReader("abc"), 3)
+		file, err := pool.NewFileWithReadLayer(rl, 6)
+		require.NoError(t, err)
+
+		buf := make([]byte, 6)
+		n, err := file.ReadAt(buf, 0)
+		require.Equal(t, err, io.EOF)
+		require.Equal(t, 6, n)
+		require.Equal(t, []byte("abc\x00\x00\x00"), buf)
+
+		require.NoError(t, file.Close())
+	})
+
+	t.Run("TruncatePropagatesToBackingLayer", func(t *testing.T) {
+		rl := filepool.NewReadLayer(strings.NewReader("abcdefghij"), 10)
+		file, err := pool.NewFileWithReadLayer(rl, 10)
+		require.NoError(t, err)
+
+		require.NoError(t, file.Truncate(4))
+
+		buf := make([]byte, 6)
+		n, err := file.ReadAt(buf, 0)
+		require.Equal(t, 4, n)
+		require.Equal(t, err, io.EOF)
+		require.Equal(t, []byte("abcd\x00\x00"), buf)
+
+		require.NoError(t, file.Close())
+	})
+
+	t.Run("TruncateHasNoGhosting", func(t *testing.T) {
+		rl := filepool.NewReadLayer(strings.NewReader("abcdefghij"), 10)
+		file, err := pool.NewFileWithReadLayer(rl, 10)
+		require.NoError(t, err)
+		// shrunk to 4 bytes
+		require.NoError(t, file.Truncate(4))
+		buf := make([]byte, 6)
+		n, err := file.ReadAt(buf, 0)
+		require.Equal(t, 4, n)
+		require.Equal(t, err, io.EOF)
+		require.Equal(t, []byte("abcd\x00\x00"), buf)
+		// grow to 10 bytes
+		require.NoError(t, file.Truncate(10))
+		n, err = file.ReadAt(buf, 0)
+		require.Equal(t, 6, n)
+		require.NoError(t, err)
+		require.Equal(t, []byte("abcd\x00\x00"), buf)
+
+		require.NoError(t, file.Close())
+	})
+
+	t.Run("WriteOverridesBackingLayer", func(t *testing.T) {
+		// Use 4 byte sectors for clarity.
+		pool := filepool.NewBlockDeviceBackedFilePool(blockDevice, sectorAllocator, 4)
+		// Each letter covers a full sector.
+		rl := filepool.NewReadLayer(strings.NewReader("AAAABBBBCCCC"), 12)
+		file, err := pool.NewFileWithReadLayer(rl, 12)
+		require.NoError(t, err)
+
+		// Write ZZZ at offset 2, this spans the first two sectors which
+		// requires us to bring them into our sparse file. We will allocate
+		// sector 10 and 11 of our block device for this (address 36 and 40).
+		sectorAllocator.EXPECT().AllocateContiguous(2).Return(uint32(10), 2, nil)
+		blockDevice.EXPECT().WriteAt([]byte("AAZZ"), int64(36)).Return(4, nil)
+		blockDevice.EXPECT().WriteAt([]byte("ZBBB"), int64(40)).Return(4, nil)
+		n, err := file.WriteAt([]byte("ZZZ"), 2)
+		require.NoError(t, err)
+		require.Equal(t, 3, n)
+
+		// This data would then be expected to be read back to us.
+		blockDevice.EXPECT().ReadAt(gomock.Len(8), int64(36)).DoAndReturn(
+			func(p []byte, offset int64) (int, error) {
+				copy(p, []byte("AAZZZBBB"))
+				return 8, nil
+			},
+		)
+
+		buf := make([]byte, 12)
+		n, err = file.ReadAt(buf, 0)
+		require.Equal(t, err, io.EOF)
+		require.Equal(t, n, 12)
+		require.Equal(t, []byte("AAZZZBBBCCCC"), buf)
+
+		sectorAllocator.EXPECT().FreeList([]uint32{10, 11})
+		require.NoError(t, file.Close())
+	})
 }

pkg/filesystem/filepool/empty_file_pool.go

@@ -13,6 +13,10 @@ func (fp emptyFilePool) NewFile() (filesystem.FileReadWriter, error) {
 	return nil, status.Error(codes.ResourceExhausted, "Cannot create file in empty file pool")
 }
 
+func (fp emptyFilePool) NewFileWithReadLayer(ReadLayer, uint64) (filesystem.FileReadWriter, error) {
+	return nil, status.Error(codes.ResourceExhausted, "Cannot create file in empty file pool")
+}
+
 // EmptyFilePool is a FilePool that does not permit the creation of new
 // files. It is used as the default FilePool for the root of the
 // worker's FUSE file system to disallow the creation of files not bound

pkg/filesystem/filepool/file_pool.go

@@ -10,6 +10,12 @@ import (
 // File handles returned by NewFile() are not thread-safe. Additional
 // locking needs to be done at higher levels to permit safe concurrent
 // access.
+//
+// Files returned by NewFileWithReadLayer allows the creation of
+// temporary files which are backed by a different data source. This
+// datasource is presumed to be immutable and allows efficient creation
+// of writeable files with content.
 type FilePool interface {
 	NewFile() (filesystem.FileReadWriter, error)
+	NewFileWithReadLayer(ReadLayer, uint64) (filesystem.FileReadWriter, error)
 }

pkg/filesystem/filepool/metrics_file_pool.go

@@ -54,6 +54,10 @@ func (fp *metricsFilePool) NewFile() (filesystem.FileReadWriter, error) {
 	}, nil
 }
 
+func (fp *metricsFilePool) NewFileWithReadLayer(readLayer ReadLayer, sizeBytes uint64) (filesystem.FileReadWriter, error) {
+	return fp.base.NewFileWithReadLayer(readLayer, sizeBytes)
+}
+
 type metricsFile struct {
 	filesystem.FileReadWriter
 }

pkg/filesystem/filepool/quota_enforcing_file_pool.go

@@ -68,6 +68,10 @@ func (fp *quotaEnforcingFilePool) NewFile() (filesystem.FileReadWriter, error) {
 	}, nil
 }
 
+func (fp *quotaEnforcingFilePool) NewFileWithReadLayer(readLayer ReadLayer, sizeBytes uint64) (filesystem.FileReadWriter, error) {
+	return nil, status.Error(codes.Internal, "Not implemented")
+}
+
 type quotaEnforcingFile struct {
 	filesystem.FileReadWriter