pagecache: more proper integration

Author: sahib

catfs/fs.go

@@ -24,6 +24,7 @@ import (
 	"github.com/sahib/brig/catfs/db"
 	ie "github.com/sahib/brig/catfs/errors"
 	"github.com/sahib/brig/catfs/mio"
+	"github.com/sahib/brig/catfs/mio/pagecache"
 	n "github.com/sahib/brig/catfs/nodes"
 	"github.com/sahib/brig/catfs/vcs"
 	"github.com/sahib/brig/repo/hints"
@@ -101,6 +102,10 @@ type FS struct {
 
 	// interface to load stream hints
 	hintManager HintManager
+
+	// cache for storing pages written to catfs.Handle
+	// (may be nil if not used, e.g. for tests)
+	pageCache pagecache.Cache
 }
 
 // ErrReadOnly is returned when a file system was created in read only mode
@@ -354,6 +359,7 @@ func NewFilesystem(
 	readOnly bool,
 	fsCfg *config.Config,
 	hintManager HintManager,
+	pageCache pagecache.Cache,
 ) (*FS, error) {
 	kv, err := db.NewBadgerDatabase(dbPath)
 	if err != nil {
@@ -394,6 +400,7 @@ func NewFilesystem(
 		repinControl:      make(chan string, 1),
 		pinner:            pinCache,
 		hintManager:       hintManager,
+		pageCache:         pageCache,
 	}
 
 	// Start the garbage collection background task.

catfs/fs_test.go

@@ -50,7 +50,7 @@ func withDummyFSReadOnly(t *testing.T, readOnly bool, fn func(fs *FS)) {
 
 	fsCfg := cfg.Section("fs")
 
-	fs, err := NewFilesystem(backend, dbPath, owner, readOnly, fsCfg, nil)
+	fs, err := NewFilesystem(backend, dbPath, owner, readOnly, fsCfg, nil, nil)
 	if err != nil {
 		t.Fatalf("Failed to create filesystem: %v", err)
 	}

catfs/handle.go

@@ -7,7 +7,7 @@ import (
 	"sync"
 
 	"github.com/sahib/brig/catfs/mio"
-	"github.com/sahib/brig/catfs/mio/overlay"
+	"github.com/sahib/brig/catfs/mio/pagecache"
 	n "github.com/sahib/brig/catfs/nodes"
 )
 
@@ -24,7 +24,7 @@ type Handle struct {
 	fs          *FS
 	file        *n.File
 	lock        sync.Mutex
-	layer       *overlay.Layer
+	layer       *pagecache.Layer
 	stream      mio.Stream
 	wasModified bool
 	isClosed    bool
@@ -40,6 +40,10 @@ func newHandle(fs *FS, file *n.File, readOnly bool) *Handle {
 }
 
 func (hdl *Handle) initStreamIfNeeded() error {
+	if hdl.fs.pageCache == nil {
+		return errors.New("no page cache was initialized")
+	}
+
 	if hdl.stream != nil {
 		return nil
 	}
@@ -60,18 +64,20 @@ func (hdl *Handle) initStreamIfNeeded() error {
 		return err
 	}
 
-	hdl.layer = overlay.NewLayer(hdl.stream)
-	hdl.layer.Truncate(int64(hdl.file.Size()))
-	hdl.layer.SetSize(int64(hdl.file.Size()))
-	return nil
+	hdl.layer, err = pagecache.NewLayer(
+		hdl.stream,
+		hdl.fs.pageCache,
+		int64(hdl.file.Inode()),
+		int64(hdl.file.Size()),
+	)
+
+	return err
 }
 
 // Read will try to fill `buf` as much as possible.
 // The seek pointer will be advanced by the number of bytes written.
 // Take care, `buf` might still have contents, even if io.EOF was returned.
 func (hdl *Handle) Read(buf []byte) (int, error) {
-	var err error
-
 	hdl.lock.Lock()
 	defer hdl.lock.Unlock()
 
@@ -83,19 +89,23 @@ func (hdl *Handle) Read(buf []byte) (int, error) {
 		return 0, err
 	}
 
-	// TODO: not sure if that makes sense...
-	// we should just read whatever the underlying stream thinks it has.
-	n, err := io.ReadFull(hdl.layer, buf)
-	isEOF := (err == io.ErrUnexpectedEOF || err == io.EOF)
-	if err != nil && !isEOF {
-		return 0, err
+	return hdl.layer.Read(buf)
+}
+
+// ReadAt reads from the overlay at `off` into `buf`.
+func (hdl *Handle) ReadAt(buf []byte, off int64) (int, error) {
+	hdl.lock.Lock()
+	defer hdl.lock.Unlock()
+
+	if hdl.isClosed {
+		return 0, ErrIsClosed
 	}
 
-	if isEOF {
-		return n, io.EOF
+	if err := hdl.initStreamIfNeeded(); err != nil {
+		return 0, err
 	}
 
-	return n, nil
+	return hdl.layer.ReadAt(buf, off)
 }
 
 // Write will write the contents of `buf` to the current position.
@@ -112,42 +122,8 @@ func (hdl *Handle) Write(buf []byte) (int, error) {
 		return 0, ErrIsClosed
 	}
 
-	if err := hdl.initStreamIfNeeded(); err != nil {
-		return 0, err
-	}
-
-	// Currently, we do not check if the file was actually modified
-	// (i.e. data changed compared to before)
 	hdl.wasModified = true
-
-	n, err := hdl.layer.Write(buf)
-	if err != nil {
-		return n, err
-	}
-
-	// Advance the write pointer when writing things to the buffer.
-	if _, err := hdl.stream.Seek(int64(n), io.SeekCurrent); err != nil && err != io.EOF {
-		return n, err
-	}
-
-	minSize := uint64(hdl.layer.MinSize())
-	if hdl.file.Size() < minSize {
-		hdl.fs.mu.Lock()
-		hdl.file.SetSize(minSize)
-
-		// Make sure to save the size change:
-		if err := hdl.fs.lkr.StageNode(hdl.file); err != nil {
-			hdl.fs.mu.Unlock()
-			return 0, err
-		}
-
-		hdl.fs.mu.Unlock()
-
-		// Also auto-truncate on every write.
-		hdl.layer.Truncate(int64(minSize))
-	}
-
-	return n, nil
+	return hdl.layer.Write(buf)
 }
 
 // WriteAt writes data from `buf` at offset `off` counted from the start (0 offset).

catfs/mio/pagecache/cache.go

@@ -1,4 +1,4 @@
-package overlay
+package pagecache
 
 import (
 	"github.com/sahib/brig/catfs/mio/pagecache/page"

catfs/mio/pagecache/doc.go

@@ -1,4 +1,4 @@
-// Package overlay implements a io.ReaderAt and io.WriterAt that is similar in
+// Package pagecache implements a io.ReaderAt and io.WriterAt that is similar in
 // function to the OverlayFS of Linux. It overlays a read-only stream and
 // enables write support. The writes will take priority on the data in stream
 // and will therefore be visible when calling ReadAt() of the overlay.
@@ -28,4 +28,4 @@
 //
 // NOTE: Whenever uint32 is used in this code, it refers to per-page offsets or
 // size. When int64 is used the content is an offset of the underlying offset.
-package overlay
+package pagecache

catfs/mio/pagecache/mdcache/l1.go

@@ -18,7 +18,7 @@ import (
 // 2. We must be able to set a max memory bound.
 // 3. We must avoid copying of pages due to performance reasons.
 //
-// The most popular libraries fail always one of the criterias:
+// The most popular libraries fail always one of the criteria:
 //
 // - fastcache: fails 1 and 3.
 // - ristretto: fails 1.
@@ -109,16 +109,14 @@ func (c *l1cache) Get(pk pageKey) (*page.Page, error) {
 	return item.Page, nil
 }
 
-func (c *l1cache) Del(pks []pageKey) error {
+func (c *l1cache) Del(pks []pageKey) {
 	for _, pk := range pks {
 		delItem, ok := c.m[pk]
 		if ok {
 			c.k.Remove(delItem.Link)
 			delete(c.m, pk)
 		}
 	}
-
-	return nil
 }
 
 func (c *l1cache) Close() error {

catfs/mio/pagecache/mdcache/l1_test.go

@@ -37,7 +37,7 @@ func TestL1GetSetDel(t *testing.T) {
 		require.Equal(t, pset.Data, pgot.Data)
 		require.Equal(t, pset.Extents, pgot.Extents)
 
-		require.NoError(t, l1.Del([]pageKey{pk}))
+		l1.Del([]pageKey{pk})
 		_, err = l1.Get(pk)
 		require.Error(t, page.ErrCacheMiss)
 	})

catfs/mio/pagecache/mdcache/l2.go

@@ -58,9 +58,9 @@ func (c *l2cache) Get(pk pageKey) (*page.Page, error) {
 	return page.FromBytes(pdata)
 }
 
-func (c *l2cache) Del(pks []pageKey) error {
+func (c *l2cache) Del(pks []pageKey) {
 	if c == nil {
-		return nil
+		return
 	}
 
 	for _, pk := range pks {
@@ -70,8 +70,6 @@ func (c *l2cache) Del(pks []pageKey) error {
 			log.Warnf("page l2: failed to delete %s", path)
 		}
 	}
-
-	return nil
 }
 
 func (c *l2cache) Close() error {

catfs/mio/pagecache/mdcache/l2_test.go

@@ -46,7 +46,7 @@ func TestL2GetSetDel(t *testing.T) {
 		require.Equal(t, pset.Data, pgot.Data)
 		require.Equal(t, pset.Extents, pgot.Extents)
 
-		require.NoError(t, l2.Del([]pageKey{pk}))
+		l2.Del([]pageKey{pk})
 		_, err = l2.Get(pk)
 		require.Error(t, page.ErrCacheMiss)
 	})
@@ -61,5 +61,5 @@ func TestL2Nil(t *testing.T) {
 	require.Error(t, page.ErrCacheMiss)
 
 	require.NoError(t, l2.Set(pageKey{0, 1}, dummyPage(0, 1024)))
-	require.NoError(t, l2.Del([]pageKey{{0, 1}}))
+	l2.Del([]pageKey{{0, 1}})
 }

catfs/mio/pagecache/mdcache/mdcache.go

@@ -1,3 +1,4 @@
+// Package mdcache implements a leveled memory/disk cache combination.
 package mdcache
 
 import (
@@ -9,9 +10,21 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+// Options give room for finetuning the behavior of Memory/Disk cache.
 type Options struct {
+	// MaxMemoryUsage of L1 in bytes
 	MaxMemoryUsage int64
-	SwapDirectory  string
+
+	// SwapDirectory specifies where L2 pages are stored.
+	// If empty, no l2 cache is used. Instead another l1 cache
+	// is used in its place, rendering MaxMemoryUsage useless.
+	// You have to set both for an effect.
+	SwapDirectory string
+
+	// L1CacheMissRefill will propagate
+	// data from L2 to L1 if it could be found
+	// successfully.
+	L1CacheMissRefill bool
 
 	// TODO: Those need to be still implemented.
 
@@ -23,17 +36,12 @@ type Options struct {
 	// on load. Reduces storage, but increases CPU usage if you're swapping.
 	// Since swapping is slow anyways this is recommended.
 	L2Compress bool
-
-	// L1CacheMissRefill will propagate
-	// data from L2 to L1 if it could be found
-	// successfully.
-	L1CacheMissRefill bool
 }
 
 type cacheLayer interface {
 	Get(pk pageKey) (*page.Page, error)
 	Set(pk pageKey, p *page.Page) error
-	Del(pks []pageKey) error
+	Del(pks []pageKey)
 	Close() error
 }
 
@@ -62,7 +70,8 @@ func (pk pageKey) String() string {
 	return filepath.Join(string(s[:2]), string(s[2:]))
 }
 
-func NewDirCache(opts Options) (*MDCache, error) {
+// New returns a new Memory/Disk cache
+func New(opts Options) (*MDCache, error) {
 	l2, err := newL2Cache(opts.SwapDirectory)
 	if err != nil {
 		return nil, err
@@ -89,6 +98,7 @@ func NewDirCache(opts Options) (*MDCache, error) {
 	}, nil
 }
 
+// Lookup implements pagecache.Cache
 func (dc *MDCache) Lookup(inode int64, pageIdx uint32) (*page.Page, error) {
 	dc.mu.Lock()
 	defer dc.mu.Unlock()
@@ -120,6 +130,7 @@ func (dc *MDCache) get(pk pageKey) (*page.Page, error) {
 	}
 }
 
+// Merge implements pagecache.Cache
 func (dc *MDCache) Merge(inode int64, pageIdx, off uint32, write []byte) error {
 	dc.mu.Lock()
 	defer dc.mu.Unlock()
@@ -149,6 +160,7 @@ func (dc *MDCache) Merge(inode int64, pageIdx, off uint32, write []byte) error {
 	return dc.l1.Set(pk, p)
 }
 
+// Evict implements pagecache.Cache
 func (dc *MDCache) Evict(inode, size int64) error {
 	dc.mu.Lock()
 	defer dc.mu.Unlock()
@@ -164,18 +176,12 @@ func (dc *MDCache) Evict(inode, size int64) error {
 		pks = append(pks, pageKey{inode: inode, pageIdx: pageIdx})
 	}
 
-	if err := dc.l1.Del(pks); err != nil {
-		log.WithError(err).Warnf("l1 delete failed for %v", pks)
-	}
-
-	// TODO: This will spam logs in case of no page:
-	if err := dc.l2.Del(pks); err != nil {
-		log.WithError(err).Warnf("l2 delete failed for %v", pks)
-	}
-
+	dc.l1.Del(pks)
+	dc.l2.Del(pks)
 	return nil
 }
 
+// Close closes the cache contents and cleans up resources.
 func (dc *MDCache) Close() error {
 	dc.mu.Lock()
 	defer dc.mu.Unlock()

catfs/mio/pagecache/mdcache/mdcache_test.go

@@ -15,7 +15,7 @@ func withMDCache(t *testing.T, fn func(mdc *MDCache)) {
 	require.NoError(t, err)
 	defer os.RemoveAll(tmpDir)
 
-	md, err := NewDirCache(Options{
+	md, err := New(Options{
 		MaxMemoryUsage:    4 * page.Size,
 		SwapDirectory:     tmpDir,
 		L1CacheMissRefill: true,