github/restic
4
0 Fork 0
mirror of https://github.com/restic/restic.git synced 2026-11-14 06:44:03 +00:00
Code Issues Releases Wiki Activity

Refactor repository structure

Browse Source 
Merge Map data type into Tree.
This commit is contained in:
Alexander Neumann committed 2015-01-14 16:34:30 +01:00
1 parent bdcdcdea7d
commit 203a911de9
26 files changed
+939 -681
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

archiver.go
+161 -68
View File
@@ -1,6 +1,7 @@
package restic
import (
"encoding/json"
"errors"
"fmt"
"io"
@@ -18,10 +19,8 @@ const (
)
type Archiver struct {
s Server
ch *ContentHandler
bl *BlobList // blobs used for the current snapshot
s Server
m *Map
fileToken chan struct{}
blobToken chan struct{}
@@ -32,7 +31,7 @@ type Archiver struct {
p *Progress
}
func NewArchiver(s Server, bl *BlobList, p *Progress) (*Archiver, error) {
func NewArchiver(s Server, p *Progress) (*Archiver, error) {
var err error
arch := &Archiver{
s: s,
@@ -50,74 +49,110 @@ func NewArchiver(s Server, bl *BlobList, p *Progress) (*Archiver, error) {
arch.blobToken <- struct{}{}
}
// create new map to store all blobs in
arch.m = NewMap()
// abort on all errors
arch.Error = func(string, os.FileInfo, error) error { return err }
// allow all files
arch.Filter = func(string, os.FileInfo) bool { return true }
arch.bl = NewBlobList()
if bl != nil {
arch.bl.Merge(bl)
}
arch.ch = NewContentHandler(s)
// load all blobs from all snapshots
// TODO: only use bloblist from old snapshot if available
err = arch.ch.LoadAllMaps()
if err != nil {
return nil, err
}
return arch, nil
}
func (arch *Archiver) Save(t backend.Type, data []byte) (Blob, error) {
blob, err := arch.ch.Save(t, data)
if err != nil {
return Blob{}, err
// compute plaintext hash
id := backend.Hash(data)
debug("Save(%v, %v)\n", t, id.Str())
// test if this blob is already known
blob, err := arch.m.FindID(id)
if err == nil {
debug("Save(%v, %v): reusing %v\n", t, id.Str(), blob.Storage.Str())
id.Free()
return blob, nil
}
// store blob in storage map for current snapshot
arch.bl.Insert(blob)
// else encrypt and save data
blob, err = arch.s.Save(t, data, id)
return blob, nil
// store blob in storage map
smapblob := arch.m.Insert(blob)
// if the map has a different storage id for this plaintext blob, use that
// one and remove the other. This happens if the same plaintext blob was
// stored concurrently and finished earlier than this blob.
if blob.Storage.Compare(smapblob.Storage) != 0 {
debug("using other block, removing %v\n", blob.Storage.Str())
// remove the blob again
// TODO: implement a list of blobs in transport, so this doesn't happen so often
err = arch.s.Remove(t, blob.Storage)
if err != nil {
return Blob{}, err
}
}
debug(": Save(%v, %v): new blob %v\n", t, id.Str(), blob)
return smapblob, nil
}
func (arch *Archiver) SaveJSON(t backend.Type, item interface{}) (Blob, error) {
blob, err := arch.ch.SaveJSON(t, item)
func (arch *Archiver) SaveTreeJSON(item interface{}) (Blob, error) {
// convert to json
data, err := json.Marshal(item)
if err != nil {
return Blob{}, err
}
// store blob in storage map for current snapshot
arch.bl.Insert(blob)
// check if tree has been saved before
buf := backend.Compress(data)
id := backend.Hash(buf)
blob, err := arch.m.FindID(id)
// return the blob if we found it
if err == nil {
return blob, nil
}
// otherwise save the data
blob, err = arch.s.Save(backend.Tree, buf, id)
if err != nil {
return Blob{}, err
}
// store blob in storage map
arch.m.Insert(blob)
return blob, nil
}
// SaveFile stores the content of the file on the backend as a Blob by calling
// Save for each chunk.
func (arch *Archiver) SaveFile(node *Node) error {
func (arch *Archiver) SaveFile(node *Node) (Blobs, error) {
debug("SaveFile(%q)\n", node.path)
file, err := os.Open(node.path)
defer file.Close()
if err != nil {
return err
return nil, err
}
// check file again
fi, err := file.Stat()
if err != nil {
return err
return nil, err
}
if fi.ModTime() != node.ModTime {
e2 := arch.Error(node.path, fi, errors.New("file was updated, using new version\n"))
e2 := arch.Error(node.path, fi, errors.New("file was updated, using new version"))
if e2 == nil {
// create new node
n, err := NodeFromFileInfo(node.path, fi)
if err != nil {
return err
return nil, err
}
// copy node
@@ -139,16 +174,16 @@ func (arch *Archiver) SaveFile(node *Node) error {
defer FreeChunkBuf("blob single file", buf)
n, err := io.ReadFull(file, buf)
if err != nil && err != io.ErrUnexpectedEOF && err != io.EOF {
return arrar.Annotate(err, "SaveFile() read small file")
return nil, arrar.Annotate(err, "SaveFile() read small file")
}
if err == io.EOF {
// use empty blob list for empty files
blobs = Blobs{}
} else {
blob, err := arch.ch.Save(backend.Data, buf[:n])
blob, err := arch.Save(backend.Data, buf[:n])
if err != nil {
return arrar.Annotate(err, "SaveFile() save chunk")
return nil, arrar.Annotate(err, "SaveFile() save chunk")
}
arch.p.Report(Stat{Bytes: blob.Size})
@@ -173,7 +208,7 @@ func (arch *Archiver) SaveFile(node *Node) error {
if err != nil {
FreeChunkBuf("blob chunker", buf)
return arrar.Annotate(err, "SaveFile() chunker.Next()")
return nil, arrar.Annotate(err, "SaveFile() chunker.Next()")
}
chunks++
@@ -183,7 +218,7 @@ func (arch *Archiver) SaveFile(node *Node) error {
resCh := make(chan Blob, 1)
go func(ch chan<- Blob) {
blob, err := arch.ch.Save(backend.Data, chunk.Data)
blob, err := arch.Save(backend.Data, chunk.Data)
// TODO handle error
if err != nil {
panic(err)
@@ -205,7 +240,7 @@ func (arch *Archiver) SaveFile(node *Node) error {
}
if len(blobs) != chunks {
return fmt.Errorf("chunker returned %v chunks, but only %v blobs saved", chunks, len(blobs))
return nil, fmt.Errorf("chunker returned %v chunks, but only %v blobs saved", chunks, len(blobs))
}
}
@@ -214,54 +249,111 @@ func (arch *Archiver) SaveFile(node *Node) error {
node.Content = make([]backend.ID, len(blobs))
for i, blob := range blobs {
node.Content[i] = blob.ID
arch.bl.Insert(blob)
bytes += blob.Size
}
if bytes != node.Size {
return fmt.Errorf("errors saving node %q: saved %d bytes, wanted %d bytes", node.path, bytes, node.Size)
return nil, fmt.Errorf("errors saving node %q: saved %d bytes, wanted %d bytes", node.path, bytes, node.Size)
}
return nil
debug("SaveFile(%q): %v\n", node.path, blobs)
return blobs, nil
}
func (arch *Archiver) saveTree(t *Tree) (Blob, error) {
debug("saveTree(%v)\n", t)
var wg sync.WaitGroup
for _, node := range *t {
if node.tree != nil && node.Subtree == nil {
// add all blobs to global map
arch.m.Merge(t.Map)
// TODO: do all this in parallel
for _, node := range t.Nodes {
if node.tree != nil {
b, err := arch.saveTree(node.tree)
if err != nil {
return Blob{}, err
}
node.Subtree = b.ID
t.Map.Insert(b)
arch.p.Report(Stat{Dirs: 1})
} else if node.Type == "file" && len(node.Content) == 0 {
// get token
token := <-arch.fileToken
} else if node.Type == "file" {
if len(node.Content) > 0 {
removeContent := false
// start goroutine
wg.Add(1)
go func(n *Node) {
defer wg.Done()
defer func() {
arch.fileToken <- token
}()
// check content
for _, id := range node.Content {
blob, err := t.Map.FindID(id)
if err != nil {
debug("unable to find storage id for data blob %v", id.Str())
arch.Error(node.path, nil, fmt.Errorf("unable to find storage id for data blob %v", id.Str()))
removeContent = true
t.Map.DeleteID(id)
arch.m.DeleteID(id)
continue
}
node.err = arch.SaveFile(n)
arch.p.Report(Stat{Files: 1})
}(node)
if ok, err := arch.s.Test(backend.Data, blob.Storage); !ok || err != nil {
debug("blob %v not in repository (error is %v)", blob, err)
arch.Error(node.path, nil, fmt.Errorf("blob %v not in repository (error is %v)", blob.Storage.Str(), err))
removeContent = true
t.Map.DeleteID(id)
arch.m.DeleteID(id)
}
}
if removeContent {
debug("removing content for %s", node.path)
node.Content = node.Content[:0]
}
}
if len(node.Content) == 0 {
// get token
token := <-arch.fileToken
// start goroutine
wg.Add(1)
go func(n *Node) {
defer wg.Done()
defer func() {
arch.fileToken <- token
}()
var blobs Blobs
blobs, node.err = arch.SaveFile(n)
for _, b := range blobs {
t.Map.Insert(b)
}
arch.p.Report(Stat{Files: 1})
}(node)
}
}
}
wg.Wait()
usedIDs := backend.NewIDSet()
// check for invalid file nodes
for _, node := range *t {
for _, node := range t.Nodes {
if node.Type == "file" && node.Content == nil && node.err == nil {
return Blob{}, fmt.Errorf("node %v has empty content", node.Name)
}
// remember used hashes
if node.Type == "file" && node.Content != nil {
for _, id := range node.Content {
usedIDs.Insert(id)
}
}
if node.Type == "dir" && node.Subtree != nil {
usedIDs.Insert(node.Subtree)
}
if node.err != nil {
err := arch.Error(node.path, nil, node.err)
if err != nil {
@@ -273,7 +365,15 @@ func (arch *Archiver) saveTree(t *Tree) (Blob, error) {
}
}
blob, err := arch.SaveJSON(backend.Tree, t)
before := len(t.Map.IDs())
t.Map.Prune(usedIDs)
after := len(t.Map.IDs())
if before != after {
debug("pruned %d ids from map for tree %v\n", before-after, t)
}
blob, err := arch.SaveTreeJSON(t)
if err != nil {
return Blob{}, err
}
@@ -296,17 +396,10 @@ func (arch *Archiver) Snapshot(dir string, t *Tree, parentSnapshot backend.ID) (
if err != nil {
return nil, nil, err
}
sn.Tree = blob.ID
// save bloblist
blob, err = arch.SaveJSON(backend.Map, arch.bl)
if err != nil {
return nil, nil, err
}
sn.Map = blob.Storage
sn.Tree = blob
// save snapshot
blob, err = arch.SaveJSON(backend.Snapshot, sn)
blob, err = arch.s.SaveJSON(backend.Snapshot, sn)
if err != nil {
return nil, nil, err
}