restic/cache.go

352 lines
7.1 KiB
Go

package restic
import (
"encoding/json"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
"github.com/restic/restic/backend"
"github.com/restic/restic/debug"
"github.com/restic/restic/server"
)
type Cache struct {
base string
}
func NewCache(be backend.Identifier) (c *Cache, err error) {
// try to get explicit cache dir from environment
dir := os.Getenv("RESTIC_CACHE")
// otherwise try OS specific default
if dir == "" {
dir, err = GetCacheDir()
if err != nil {
return nil, err
}
}
basedir := filepath.Join(dir, be.ID())
debug.Log("Cache.New", "opened cache at %v", basedir)
return &Cache{base: basedir}, nil
}
func (c *Cache) Has(t backend.Type, subtype string, id backend.ID) (bool, error) {
// try to open file
filename, err := c.filename(t, subtype, id)
if err != nil {
return false, err
}
fd, err := os.Open(filename)
defer fd.Close()
if err != nil {
if os.IsNotExist(err) {
debug.Log("Cache.Has", "test for file %v: not cached", filename)
return false, nil
}
debug.Log("Cache.Has", "test for file %v: error %v", filename, err)
return false, err
}
debug.Log("Cache.Has", "test for file %v: is cached", filename)
return true, nil
}
func (c *Cache) Store(t backend.Type, subtype string, id backend.ID) (io.WriteCloser, error) {
filename, err := c.filename(t, subtype, id)
if err != nil {
return nil, err
}
dirname := filepath.Dir(filename)
err = os.MkdirAll(dirname, 0700)
if err != nil {
return nil, err
}
file, err := os.Create(filename)
if err != nil {
debug.Log("Cache.Store", "error creating file %v: %v", filename, err)
return nil, err
}
debug.Log("Cache.Store", "created file %v", filename)
return file, nil
}
func (c *Cache) Load(t backend.Type, subtype string, id backend.ID) (io.ReadCloser, error) {
// try to open file
filename, err := c.filename(t, subtype, id)
if err != nil {
return nil, err
}
return os.Open(filename)
}
func (c *Cache) Purge(t backend.Type, subtype string, id backend.ID) error {
filename, err := c.filename(t, subtype, id)
if err != nil {
return err
}
err = os.Remove(filename)
debug.Log("Cache.Purge", "Remove file %v: %v", filename, err)
if err != nil && os.IsNotExist(err) {
return nil
}
return err
}
func (c *Cache) Clear(s *server.Server) error {
list, err := c.List(backend.Snapshot)
if err != nil {
return err
}
for _, entry := range list {
debug.Log("Cache.Clear", "found entry %v", entry)
if ok, err := s.Test(backend.Snapshot, entry.ID.String()); !ok || err != nil {
debug.Log("Cache.Clear", "snapshot %v doesn't exist any more, removing %v", entry.ID, entry)
err = c.Purge(backend.Snapshot, entry.Subtype, entry.ID)
if err != nil {
return err
}
}
}
return nil
}
type CacheEntry struct {
ID backend.ID
Subtype string
}
func (c CacheEntry) String() string {
if c.Subtype != "" {
return c.ID.Str() + "." + c.Subtype
}
return c.ID.Str()
}
func (c *Cache) List(t backend.Type) ([]CacheEntry, error) {
var dir string
switch t {
case backend.Snapshot:
dir = filepath.Join(c.base, "snapshots")
case backend.Tree:
dir = filepath.Join(c.base, "trees")
default:
return nil, fmt.Errorf("cache not supported for type %v", t)
}
fd, err := os.Open(dir)
if err != nil {
if os.IsNotExist(err) {
return []CacheEntry{}, nil
}
return nil, err
}
defer fd.Close()
fis, err := fd.Readdir(-1)
if err != nil {
return nil, err
}
entries := make([]CacheEntry, 0, len(fis))
for _, fi := range fis {
parts := strings.SplitN(fi.Name(), ".", 2)
id, err := backend.ParseID(parts[0])
// ignore invalid cache entries for now
if err != nil {
debug.Log("Cache.List", "unable to parse name %v as id: %v", parts[0], err)
continue
}
e := CacheEntry{ID: id}
if len(parts) == 2 {
e.Subtype = parts[1]
}
entries = append(entries, e)
}
return entries, nil
}
// Construct file name for given Type.
func (c *Cache) filename(t backend.Type, subtype string, id backend.ID) (string, error) {
filename := id.String()
if subtype != "" {
filename += "." + subtype
}
switch t {
case backend.Snapshot:
return filepath.Join(c.base, "snapshots", filename), nil
case backend.Tree:
return filepath.Join(c.base, "trees", filename), nil
}
return "", fmt.Errorf("cache not supported for type %v", t)
}
// high-level functions
// RefreshSnapshots loads the maps for all snapshots and saves them to the
// local cache. Old cache entries are purged.
func (c *Cache) RefreshSnapshots(s *server.Server, p *Progress) error {
defer p.Done()
// list cache entries
entries, err := c.List(backend.Snapshot)
if err != nil {
return err
}
// list snapshots first
done := make(chan struct{})
defer close(done)
// check that snapshot blobs are cached
for name := range s.List(backend.Snapshot, done) {
id, err := backend.ParseID(name)
if err != nil {
continue
}
// remove snapshot from list of entries
for i, e := range entries {
if e.ID.Equal(id) {
entries = append(entries[:i], entries[i+1:]...)
break
}
}
has, err := c.Has(backend.Snapshot, "blobs", id)
if err != nil {
return err
}
if has {
continue
}
// else start progress reporting
p.Start()
// build new cache
_, err = cacheSnapshotBlobs(p, s, c, id)
if err != nil {
debug.Log("Cache.RefreshSnapshots", "unable to cache snapshot blobs for %v: %v", id.Str(), err)
return err
}
}
// remove other entries
for _, e := range entries {
debug.Log("Cache.RefreshSnapshots", "remove entry %v", e)
err = c.Purge(backend.Snapshot, e.Subtype, e.ID)
if err != nil {
return err
}
}
return nil
}
// cacheSnapshotBlobs creates a cache of all the blobs used within the
// snapshot. It collects all blobs from all trees and saves the resulting map
// to the cache and returns the map.
func cacheSnapshotBlobs(p *Progress, s *server.Server, c *Cache, id backend.ID) (*Map, error) {
debug.Log("CacheSnapshotBlobs", "create cache for snapshot %v", id.Str())
sn, err := LoadSnapshot(s, id)
if err != nil {
debug.Log("CacheSnapshotBlobs", "unable to load snapshot %v: %v", id.Str(), err)
return nil, err
}
m := NewMap()
// add top-level node
m.Insert(sn.Tree)
p.Report(Stat{Trees: 1})
// start walker
var wg sync.WaitGroup
ch := make(chan WalkTreeJob)
wg.Add(1)
go func() {
WalkTree(s, sn.Tree, nil, ch)
wg.Done()
}()
for i := 0; i < maxConcurrencyPreload; i++ {
wg.Add(1)
go func() {
for job := range ch {
if job.Tree == nil {
continue
}
p.Report(Stat{Trees: 1})
debug.Log("CacheSnapshotBlobs", "got job %v", job)
m.Merge(job.Tree.Map)
}
wg.Done()
}()
}
wg.Wait()
// save blob list for snapshot
return m, c.StoreMap(id, m)
}
func (c *Cache) StoreMap(snid backend.ID, m *Map) error {
wr, err := c.Store(backend.Snapshot, "blobs", snid)
if err != nil {
return nil
}
defer wr.Close()
enc := json.NewEncoder(wr)
err = enc.Encode(m)
if err != nil {
return err
}
return nil
}
func (c *Cache) LoadMap(s *server.Server, snid backend.ID) (*Map, error) {
rd, err := c.Load(backend.Snapshot, "blobs", snid)
if err != nil {
return nil, err
}
m := &Map{}
err = json.NewDecoder(rd).Decode(m)
return m, err
}