diff --git a/storage/caching/cache_entry.go b/storage/caching/cache_entry.go deleted file mode 100644 index 19e231614..000000000 --- a/storage/caching/cache_entry.go +++ /dev/null @@ -1,66 +0,0 @@ -package caching - -import ( - "bytes" - "encoding/binary" - "errors" - "fmt" - "time" -) - -const ( - cacheEntryFormatVersion = 1 - sizeDoesNotExists = 0x4000000000000000 -) - -type blockCacheEntry struct { - accessTime int64 - size int64 -} - -func (e *blockCacheEntry) exists() bool { - return e.size != sizeDoesNotExists -} - -func (e blockCacheEntry) GoString() string { - ts := time.Unix(e.accessTime/1000000000, e.accessTime%1000000000) - switch e.size { - case sizeDoesNotExists: - return fmt.Sprintf("entry[not-found;acc:%v]", ts) - default: - return fmt.Sprintf("entry[size:%v;acc:%v]", e.size, ts) - } -} - -func (e *blockCacheEntry) serialize() []byte { - var buf bytes.Buffer - - buf.WriteByte(cacheEntryFormatVersion) - binary.Write(&buf, binary.BigEndian, e.accessTime) - binary.Write(&buf, binary.BigEndian, e.size) - - return buf.Bytes() -} - -func (e *blockCacheEntry) deserialize(b []byte) error { - r := bytes.NewReader(b) - - var version byte - if err := binary.Read(r, binary.BigEndian, &version); err != nil { - return err - } - - if version != cacheEntryFormatVersion { - return errors.New("invalid format") - } - - if err := binary.Read(r, binary.BigEndian, &e.accessTime); err != nil { - return err - } - - if err := binary.Read(r, binary.BigEndian, &e.size); err != nil { - return err - } - - return nil -} diff --git a/storage/caching/caching_storage.go b/storage/caching/caching_storage.go deleted file mode 100644 index e169a7ce9..000000000 --- a/storage/caching/caching_storage.go +++ /dev/null @@ -1,246 +0,0 @@ -// Package caching implements a caching wrapper around another Storage. -package caching - -import ( - "context" - "fmt" - "log" - "os" - "path/filepath" - "time" - - "github.com/boltdb/bolt" - "github.com/kopia/kopia/storage" - "github.com/kopia/kopia/storage/filesystem" -) - -var ( - dbBucketBlocks = []byte("Blocks") - currentTime func() int64 -) - -const ( - defaultCacheSizeBytes = 100000000 -) - -type cachingStorage struct { - master storage.Storage - cache storage.Storage - db *bolt.DB - sizeBytes int64 - - *lockMap -} - -func defaultGetCurrentTime() int64 { - return time.Now().UnixNano() -} - -var getCurrentTime = defaultGetCurrentTime - -func (c *cachingStorage) getCacheEntry(block string) (result blockCacheEntry, ok bool) { - c.db.Update(func(t *bolt.Tx) error { - b := t.Bucket(dbBucketBlocks) - if b == nil { - return nil - } - - cacheKey := []byte(block) - entryBytes := b.Get(cacheKey) - var e blockCacheEntry - - if entryBytes != nil { - err := e.deserialize(entryBytes) - if err == nil { - //log.Printf("got cache entry: %v %#v", block, e) - result = e - ok = true - - e.accessTime = getCurrentTime() - - return b.Put(cacheKey, e.serialize()) - } - - log.Printf("WARNING: Can't read cache entry: %v", err) - } - - return nil - }) - return -} - -func (c *cachingStorage) setCacheEntrySize(block string, size int64) { - c.db.Update(func(t *bolt.Tx) error { - b, err := t.CreateBucketIfNotExists(dbBucketBlocks) - if err != nil { - return err - } - - e := blockCacheEntry{ - accessTime: getCurrentTime(), - size: size, - } - - // log.Printf("updating cache entry %v with size %x", block, size) - return b.Put([]byte(block), e.serialize()) - }) -} - -func (c *cachingStorage) removeCacheEntry(block string) { - c.db.Update(func(t *bolt.Tx) error { - b, err := t.CreateBucketIfNotExists(dbBucketBlocks) - if err != nil { - return err - } - - return b.Delete([]byte(block)) - }) -} - -func (c *cachingStorage) BlockSize(id string) (int64, error) { - if entry, ok := c.getCacheEntry(id); ok { - if entry.exists() { - return entry.size, nil - } - - return 0, storage.ErrBlockNotFound - } - - c.Lock(id) - defer c.Unlock(id) - - l, err := c.master.BlockSize(id) - if err != nil { - if err == storage.ErrBlockNotFound { - c.setCacheEntrySize(id, sizeDoesNotExists) - } - return 0, err - } - - c.setCacheEntrySize(id, l) - return l, nil -} - -func (c *cachingStorage) DeleteBlock(id string) error { - c.Lock(id) - defer c.Unlock(id) - - // Remove from cache first. - c.cache.DeleteBlock(id) - - if err := c.master.DeleteBlock(id); err != nil { - return err - } - c.setCacheEntrySize(id, sizeDoesNotExists) - - return nil -} - -func (c *cachingStorage) GetBlock(id string, offset, length int64) ([]byte, error) { - c.Lock(id) - defer c.Unlock(id) - - if blockCacheEntry, ok := c.getCacheEntry(id); ok { - if !blockCacheEntry.exists() { - return nil, storage.ErrBlockNotFound - } - - v, err := c.cache.GetBlock(id, offset, length) - if err == nil { - return v, nil - } - } - - // Download from master - b, err := c.master.GetBlock(id, 0, -1) - - if err == nil { - l := int64(len(b)) - c.cache.PutBlock(id, b) - c.setCacheEntrySize(id, l) - } else if err == storage.ErrBlockNotFound { - c.setCacheEntrySize(id, sizeDoesNotExists) - } - - return b, err -} - -func (c *cachingStorage) PutBlock(id string, data []byte) error { - c.Lock(id) - defer c.Unlock(id) - - // Remove from cache first. - c.cache.DeleteBlock(id) - c.removeCacheEntry(id) - - return c.master.PutBlock(id, data) -} - -func (c *cachingStorage) ListBlocks(prefix string) (chan storage.BlockMetadata, storage.CancelFunc) { - return c.master.ListBlocks(prefix) -} - -func (c *cachingStorage) Close() error { - if c.db != nil { - c.db.Close() - c.db = nil - } - - if c.cache != nil { - c.cache.Close() - c.cache = nil - } - - if c.master != nil { - c.master.Close() - c.master = nil - } - return nil -} - -// Options provides options for the caching storage wrapper -type Options struct { - CacheDir string `json:"cacheDir"` - CacheSizeBytes int64 `json:"cacheSizeBytes"` -} - -// NewWrapper creates new caching storage wrapper. -func NewWrapper(ctx context.Context, master storage.Storage, options *Options) (storage.Storage, error) { - if options.CacheDir == "" { - return nil, fmt.Errorf("Cache directory must be specified") - } - cacheDataDir := filepath.Join(options.CacheDir, "data") - cacheDBFile := filepath.Join(options.CacheDir, "cache.db") - - os.MkdirAll(cacheDataDir, 0700) - - db, err := bolt.Open(cacheDBFile, 0600, &bolt.Options{Timeout: 1 * time.Second}) - if err != nil { - return nil, fmt.Errorf("cannot open cache database: %v", err) - } - - cs, err := filesystem.New(ctx, &filesystem.Options{ - Path: cacheDataDir, - }) - - if err != nil { - return nil, fmt.Errorf("cannot open cache directory: %v", err) - } - - sizeBytes := options.CacheSizeBytes - if sizeBytes == 0 { - sizeBytes = defaultCacheSizeBytes - } - - s := &cachingStorage{ - master: master, - cache: cs, - db: db, - sizeBytes: sizeBytes, - lockMap: newLockMap(), - } - - return s, nil -} - -var _ storage.Storage = &cachingStorage{} diff --git a/storage/caching/caching_storage_test.go b/storage/caching/caching_storage_test.go deleted file mode 100644 index d8e7c3e7f..000000000 --- a/storage/caching/caching_storage_test.go +++ /dev/null @@ -1,147 +0,0 @@ -package caching - -import ( - "context" - "fmt" - "io/ioutil" - "log" - "os" - "strings" - - "github.com/kopia/kopia/internal/storagetesting" - "github.com/kopia/kopia/storage/logging" - - "testing" -) - -type tracer []string - -func (tr *tracer) Printf(format string, args ...interface{}) { - *tr = append(*tr, fmt.Sprintf(format, args...)) - log.Printf("base: %v", fmt.Sprintf(format, args...)) -} - -func (tr *tracer) clear() { - *tr = nil -} - -func (tr *tracer) assertActivityAndClear(t *testing.T, expectedLogs ...string) bool { - b := tr.assertActivity(t, expectedLogs...) - tr.clear() - return b -} - -func (tr *tracer) assertActivity(t *testing.T, expectedLogs ...string) bool { - if !tr.matches(expectedLogs...) { - t.Errorf("Unexpected calls. Got %v calls:\n%v\nExpected %v calls:\n%v\n", - len(*tr), - strings.Join(*tr, "\n"), - len(expectedLogs), - strings.Join(expectedLogs, "\n")) - return false - } - - return true -} - -func (tr *tracer) matches(expectedLogs ...string) bool { - if len(expectedLogs) != len(*tr) { - return false - } - - for i, entry := range *tr { - if !strings.HasPrefix(entry, expectedLogs[i]) { - return false - } - } - - return true -} - -func (tr *tracer) assertNoActivity(t *testing.T) bool { - return tr.assertActivity(t) -} - -func TestCache(t *testing.T) { - tmpdir, err := ioutil.TempDir("", "kopia-cache") - if err != nil { - t.Errorf("cannot create temp directory for testing: %v", err) - return - } - defer os.RemoveAll(tmpdir) - - // tmpdir = "/tmp/cache" - // os.RemoveAll(tmpdir) - - masterData := map[string][]byte{} - master := storagetesting.NewMapStorage(masterData, nil) - - var tr tracer - master = logging.NewWrapper(master, logging.Output(tr.Printf)) - - cache, err := NewWrapper(context.Background(), master, &Options{CacheDir: tmpdir}) - defer cache.Close() - if err != nil { - t.Errorf("cannot create cache: %v", err) - return - } - - data1 := []byte("foo-bar") - data2 := []byte("baz-qux") - masterData["x"] = data1 - masterData["z"] = data2 - - tr.assertNoActivity(t) - - storagetesting.AssertGetBlock(t, cache, "x", data1) - tr.assertActivityAndClear(t, "GetBlock") - storagetesting.AssertBlockExists(t, cache, "x", true) - tr.assertNoActivity(t) - - storagetesting.AssertGetBlockNotFound(t, cache, "y") - tr.assertActivityAndClear(t, "GetBlock") - storagetesting.AssertBlockExists(t, cache, "y", false) - tr.assertNoActivity(t) - - storagetesting.AssertGetBlock(t, cache, "z", data2) - tr.assertActivityAndClear(t, "GetBlock") - storagetesting.AssertBlockExists(t, cache, "z", true) - tr.assertNoActivity(t) - - storagetesting.AssertGetBlock(t, cache, "x", data1) - storagetesting.AssertBlockExists(t, cache, "x", true) - tr.assertNoActivity(t) - - storagetesting.AssertGetBlockNotFound(t, cache, "y") - storagetesting.AssertBlockExists(t, cache, "y", false) - tr.assertNoActivity(t) - - storagetesting.AssertGetBlock(t, cache, "z", data2) - storagetesting.AssertBlockExists(t, cache, "z", true) - tr.assertNoActivity(t) - - cache.DeleteBlock("z") - tr.assertActivityAndClear(t, "DeleteBlock") - - storagetesting.AssertBlockExists(t, cache, "z", false) - storagetesting.AssertGetBlockNotFound(t, cache, "z") - tr.assertNoActivity(t) - - cache.PutBlock("z", data1) - tr.assertActivityAndClear(t, "PutBlock") - - storagetesting.AssertBlockExists(t, cache, "z", true) - tr.assertActivityAndClear(t, "BlockSize") - - storagetesting.AssertGetBlock(t, cache, "z", data1) - tr.assertActivityAndClear(t, "GetBlock") - - cache.PutBlock("x2", data1) - tr.assertActivityAndClear(t, "PutBlock") - - storagetesting.AssertListResults(t, cache, "", "x", "x2", "z") - tr.assertActivityAndClear(t, "ListBlocks", "Cancelled ListBlocks") - - cache.Close() - tr.assertActivityAndClear(t, "Close") -} diff --git a/storage/caching/lock_map.go b/storage/caching/lock_map.go deleted file mode 100644 index 6713356af..000000000 --- a/storage/caching/lock_map.go +++ /dev/null @@ -1,40 +0,0 @@ -// Package caching implements a caching wrapper around another Storage. -package caching - -import "sync" - -type lockMap struct { - cond *sync.Cond - locks map[string]bool -} - -func (l *lockMap) getSync(id string) (*sync.Cond, map[string]bool) { - return l.cond, l.locks -} - -func (l *lockMap) Lock(id string) { - cv, locks := l.getSync(id) - - cv.L.Lock() - for locks[id] { - cv.Wait() - } - locks[id] = true - cv.L.Unlock() -} - -func (l *lockMap) Unlock(id string) { - cv, locks := l.getSync(id) - - cv.L.Lock() - delete(locks, id) - cv.Signal() - cv.L.Unlock() -} - -func newLockMap() *lockMap { - return &lockMap{ - cond: &sync.Cond{L: &sync.Mutex{}}, - locks: map[string]bool{}, - } -} diff --git a/storage/caching/lock_map_test.go b/storage/caching/lock_map_test.go deleted file mode 100644 index 6be937544..000000000 --- a/storage/caching/lock_map_test.go +++ /dev/null @@ -1,45 +0,0 @@ -// Package caching implements a caching wrapper around another Storage. -package caching - -import ( - "fmt" - "sync" - "testing" - "time" -) - -func TestLockMap(t *testing.T) { - lockMap := newLockMap() - var wg sync.WaitGroup - - // 100 goroutines competing for 10 buckets - workers := 100 - lockedBlocks := 10 - workTime := 20 * time.Millisecond - - counters := make([]int, lockedBlocks) - - wg.Add(workers) - for i := 0; i < workers; i++ { - go func(i int) { - b := i % lockedBlocks - blockID := fmt.Sprintf("block-%v", b) - for i := 0; i < 3; i++ { - lockMap.Lock(blockID) - if counters[b] != 0 { - t.Errorf("*** multiple goroutines entered block %v", blockID) - } - counters[b]++ - time.Sleep(workTime) - counters[b]-- - if counters[b] != 0 { - t.Errorf("*** multiple goroutines entered block %v", blockID) - } - lockMap.Unlock(blockID) - } - defer wg.Done() - }(i) - } - - wg.Wait() -}