mirror of
https://github.com/kopia/kopia.git
synced 2026-10-21 05:34:42 -04:00
removed caching storage
This commit is contained in:
Jarek Kowalski committed
1 parent
a700508048
commit
a0cc09b673
5 files changed
-544
No files matched your search
@@ -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
|
||||
}
|
||||
@@ -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{}
|
||||
@@ -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")
|
||||
}
|
||||
@@ -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{},
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
Reference in new issue
Block a user