Remove chunks from cache if no slices need chunk anymore

Author: Philipp Heckel

cmd/fsdup/main.go

@@ -142,6 +142,7 @@ func mapCommand(args []string) {
 	flags := flag.NewFlagSet("map", flag.ExitOnError)
 	debugFlag := flags.Bool("debug", fsdup.Debug, "Enable debug mode")
 	storeFlag := flags.String("store", "index", "Location of the chunk store")
+	cacheFlag := flags.String("cache", "cache", "Location of the chunk cache")
 	targetFlag := flags.String("target", "", "Target device or file used for local caching and live migration")
 
 	flags.Parse(args)
@@ -161,9 +162,10 @@ func mapCommand(args []string) {
 		exit(2, "Invalid syntax: " + string(err.Error()))
 	}
 
+	cache := fsdup.NewFileChunkStore(*cacheFlag)
 	targetFile := *targetFlag
 
-	if err := fsdup.Map(manifestFile, store, targetFile); err != nil {
+	if err := fsdup.Map(manifestFile, store, cache, targetFile); err != nil {
 		exit(2, "Cannot map drive file: " + string(err.Error()))
 	}
 }

log.go

@@ -2,6 +2,7 @@ package fsdup
 
 import (
 	"fmt"
+	"log"
 	"strings"
 )
 
@@ -18,7 +19,7 @@ func debugf(format string, args ...interface{}) {
 	}
 
 	if Debug {
-		fmt.Printf(format, args...)
+		log.Printf(format, args...)
 	}
 }
 

manifest.go

@@ -66,8 +66,8 @@ func NewManifestFromFile(file string) (*manifest, error) {
 // Breakpoints returns a sorted list of breakpoints, useful for sequential disk traversal
 func (m *manifest) Offsets() []int64 {
 	offsets := make([]int64, 0, len(m.diskMap))
-	for breakpoint, _ := range m.diskMap {
-		offsets = append(offsets, breakpoint)
+	for offset, _ := range m.diskMap {
+		offsets = append(offsets, offset)
 	}
 
 	sort.Slice(offsets, func(i, j int) bool {

map.go

@@ -23,10 +23,10 @@ type manifestImage struct {
 	offsets    []int64
 	chunks     map[string]*chunk
 	written    map[int64]bool
-	//chunkCount map[string]int64
+	sliceCount map[string]int64
 }
 
-func Map(manifestFile string, store ChunkStore, targetFile string) error {
+func Map(manifestFile string, store ChunkStore, cache ChunkStore, targetFile string) error {
 	manifest, err := NewManifestFromFile(manifestFile)
 	if err != nil {
 		return err
@@ -51,7 +51,7 @@ func Map(manifestFile string, store ChunkStore, targetFile string) error {
 
 	debugf("Creating device %s ...\n", deviceName)
 
-	image := NewManifestImage(manifest, store, target)
+	image := NewManifestImage(manifest, store, cache, target)
 	device, err := buse.CreateDevice(deviceName, uint(manifest.Size()), image)
 	if err != nil {
 		return err
@@ -61,7 +61,7 @@ func Map(manifestFile string, store ChunkStore, targetFile string) error {
 	signal.Notify(sig, os.Interrupt)
 	go func() {
 		if err := device.Connect(); err != nil {
-			log.Printf("Buse device stopped with error: %s", err)
+			debugf("Buse device stopped with error: %s", err)
 		} else {
 			log.Println("Buse device stopped gracefully.")
 		}
@@ -76,16 +76,31 @@ func Map(manifestFile string, store ChunkStore, targetFile string) error {
 	return nil
 }
 
-func NewManifestImage(manifest *manifest, store ChunkStore, target *os.File) *manifestImage {
+func NewManifestImage(manifest *manifest, store ChunkStore, cache ChunkStore, target *os.File) *manifestImage {
+	sliceCount := make(map[string]int64, 0)
+
+	for _, sliceOffset := range manifest.Offsets() {
+		slice := manifest.Get(sliceOffset)
+		if slice.checksum != nil {
+			checksumStr := fmt.Sprintf("%x", slice.checksum)
+
+			if _, ok := sliceCount[checksumStr]; ok {
+				sliceCount[checksumStr]++
+			} else {
+				sliceCount[checksumStr] = 1
+			}
+		}
+	}
+
 	return &manifestImage{
-		manifest: manifest,
-		store:    store,
-		target:   target,
-		cache:    NewFileChunkStore("cache"),
-		offsets:  manifest.Offsets(), // cache !
-		chunks:   manifest.Chunks(),  // cache !
-		written:  make(map[int64]bool, 0),
-		// chunkCount: make(map[string]int64, 0),
+		manifest:   manifest,
+		store:      store,
+		target:     target,
+		cache:      cache,
+		offsets:    manifest.Offsets(), // cache !
+		chunks:     manifest.Chunks(),  // cache !
+		written:    make(map[int64]bool, 0),
+		sliceCount: sliceCount,
 	}
 }
 
@@ -116,7 +131,7 @@ func (d *manifestImage) WriteAt(p []byte, off uint) error {
 		return err
 	}
 
-	log.Printf("WRITE offset:%d len:%d\n", off, len(p))
+	debugf("WRITE offset:%d len:%d\n", off, len(p))
 
 	written, err := d.target.WriteAt(p, int64(off))
 	if err != nil {
@@ -183,13 +198,14 @@ func (d *manifestImage) syncSlice(offset int64, slice *chunkSlice) error {
 	debugf("Syncing diskoff %d - %d (len %d) -> checksum %x, %d to %d\n",
 		offset, offset + length, length, slice.checksum, slice.from, slice.to)
 
+	checksumStr := fmt.Sprintf("%x", slice.checksum)
 	buffer := make([]byte, chunkSizeMaxBytes) // FIXME: Make this a buffer pool
 	read, err := d.cache.ReadAt(slice.checksum, buffer[:length], slice.from)
 	if err != nil {
 		debugf("Chunk %x not in cache. Retrieving full chunk ...\n", slice.checksum)
 
 		// Read entire chunk, store to cache
-		chunk := d.chunks[fmt.Sprintf("%x", slice.checksum)]
+		chunk := d.chunks[checksumStr]
 
 		// FIXME: This will fill up the local cache will all chunks and never delete it
 		read, err = d.store.ReadAt(slice.checksum, buffer[:chunk.size], 0)
@@ -213,7 +229,16 @@ func (d *manifestImage) syncSlice(offset int64, slice *chunkSlice) error {
 		return err
 	}
 
+	// Accounting
 	d.written[offset] = true
+	d.sliceCount[checksumStr]--
+
+	// Remove from cache if it will never be requested again
+	if d.sliceCount[checksumStr] == 0 {
+		if err := d.cache.Remove(slice.checksum); err != nil {
+			return err
+		}
+	}
 
 	return nil
 }

store.go

@@ -1,6 +1,7 @@
 package fsdup
 
 type ChunkStore interface {
-	Write(checksum []byte, buffer []byte) error
 	ReadAt(checksum []byte, buffer []byte, offset int64) (int, error)
+	Write(checksum []byte, buffer []byte) error
+	Remove(checksum []byte) error
 }

store_ceph.go

@@ -21,6 +21,22 @@ func NewCephStore(configFile string, pool string) *cephChunkStore {
 	}
 }
 
+func (idx *cephChunkStore) ReadAt(checksum []byte, buffer []byte, offset int64) (int, error) {
+	if err := idx.openPool(); err != nil {
+		return 0, err
+	}
+
+	checksumStr := fmt.Sprintf("%x", checksum)
+	read, err := idx.ctx.Read(checksumStr, buffer, uint64(offset))
+	if err != nil {
+		return 0, err
+	} else if read != len(buffer) {
+		return 0, errors.New("cannot read full section")
+	}
+
+	return read, nil
+}
+
 func (idx *cephChunkStore) Write(checksum []byte, buffer []byte) error {
 	if err := idx.openPool(); err != nil {
 		return err
@@ -41,20 +57,18 @@ func (idx *cephChunkStore) Write(checksum []byte, buffer []byte) error {
 	return nil
 }
 
-func (idx *cephChunkStore) ReadAt(checksum []byte, buffer []byte, offset int64) (int, error) {
+func (idx *cephChunkStore) Remove(checksum []byte) error {
 	if err := idx.openPool(); err != nil {
-		return 0, err
+		return err
 	}
 
 	checksumStr := fmt.Sprintf("%x", checksum)
-	read, err := idx.ctx.Read(checksumStr, buffer, uint64(offset))
+	err := idx.ctx.Delete(checksumStr)
 	if err != nil {
-		return 0, err
-	} else if read != len(buffer) {
-		return 0, errors.New("cannot read full section")
+		return err
 	}
 
-	return read, nil
+	return nil
 }
 
 func (idx *cephChunkStore) openPool() error {

store_dummy.go

@@ -16,4 +16,8 @@ func (idx *dummyChunkStore) Write(checksum []byte, buffer []byte) error {
 
 func (idx *dummyChunkStore) ReadAt(checksum []byte, buffer []byte, offset int64) (int, error) {
 	return 0, errors.New("cannot read from a dummy store, dummy!")
+}
+
+func (idx *dummyChunkStore) Remove(checksum []byte) error {
+	return nil
 }

store_file.go

@@ -66,3 +66,19 @@ func (idx *fileChunkStore) ReadAt(checksum []byte, buffer []byte, offset int64)
 
 	return read, nil
 }
+
+func (idx *fileChunkStore) Remove(checksum []byte) error {
+	checksumStr := fmt.Sprintf("%x", checksum)
+	dir1 := fmt.Sprintf("%s/%s", idx.root, checksumStr[0:3])
+	dir2 := fmt.Sprintf("%s/%s/%s", idx.root, checksumStr[0:3], checksumStr[3:6])
+	file := fmt.Sprintf("%s/%s", dir2, checksumStr)
+
+	if err := os.Remove(file); err != nil {
+		return err
+	}
+
+	os.Remove(dir2)
+	os.Remove(dir1)
+
+	return nil
+}