mirror of
https://github.com/restic/restic.git
synced 2026-10-07 11:07:11 +00:00
Add packer, update Design.md
This commit is contained in:
Alexander Neumann committed
1 parent
66b81a2a8b
commit
8f3b91ba32
3 files changed
+450
-51
No files matched your search
+270
@@ -0,0 +1,270 @@
|
||||
package pack
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"sync"
|
||||
|
||||
"github.com/restic/restic/backend"
|
||||
"github.com/restic/restic/crypto"
|
||||
)
|
||||
|
||||
type BlobType uint8
|
||||
|
||||
const (
|
||||
Data BlobType = 0
|
||||
Tree = 1
|
||||
)
|
||||
|
||||
func (t BlobType) MarshalJSON() ([]byte, error) {
|
||||
switch t {
|
||||
case Data:
|
||||
return []byte(`"data"`), nil
|
||||
case Tree:
|
||||
return []byte(`"tree"`), nil
|
||||
}
|
||||
|
||||
return nil, errors.New("unknown blob type")
|
||||
}
|
||||
|
||||
func (t *BlobType) UnmarshalJSON(buf []byte) error {
|
||||
switch string(buf) {
|
||||
case `"data"`:
|
||||
*t = Data
|
||||
case `"tree"`:
|
||||
*t = Tree
|
||||
default:
|
||||
return errors.New("unknown blob type")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Blob is a blob within a pack.
|
||||
type Blob struct {
|
||||
Type BlobType
|
||||
Length uint32
|
||||
ID backend.ID
|
||||
Offset uint
|
||||
}
|
||||
|
||||
// GetReader returns an io.Reader for the blob entry e.
|
||||
func (e Blob) GetReader(rd io.ReadSeeker) (io.Reader, error) {
|
||||
// seek to the correct location
|
||||
_, err := rd.Seek(int64(e.Offset), 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return io.LimitReader(rd, int64(e.Length)), nil
|
||||
}
|
||||
|
||||
// Packer is used to create a new Pack.
|
||||
type Packer struct {
|
||||
blobs []Blob
|
||||
|
||||
bytes uint
|
||||
k *crypto.Key
|
||||
wr io.Writer
|
||||
hw *backend.HashingWriter
|
||||
|
||||
m sync.Mutex
|
||||
}
|
||||
|
||||
// NewPacker returns a new Packer that can be used to pack blobs
|
||||
// together.
|
||||
func NewPacker(k *crypto.Key, w io.Writer) *Packer {
|
||||
return &Packer{k: k, wr: w, hw: backend.NewHashingWriter(w, sha256.New())}
|
||||
}
|
||||
|
||||
// Add saves the data read from rd as a new blob to the packer. Returned is the
|
||||
// number of bytes written to the pack.
|
||||
func (p *Packer) Add(t BlobType, id backend.ID, rd io.Reader) (int64, error) {
|
||||
p.m.Lock()
|
||||
defer p.m.Unlock()
|
||||
|
||||
c := Blob{Type: t, ID: id}
|
||||
|
||||
n, err := io.Copy(p.hw, rd)
|
||||
c.Length = uint32(n)
|
||||
c.Offset = p.bytes
|
||||
p.bytes += uint(n)
|
||||
p.blobs = append(p.blobs, c)
|
||||
|
||||
return n, err
|
||||
}
|
||||
|
||||
var entrySize = binary.Size(BlobType(0)) + binary.Size(uint32(0)) + backend.IDSize
|
||||
|
||||
// headerEntry is used with encoding/binary to read and write header entries
|
||||
type headerEntry struct {
|
||||
Type BlobType
|
||||
Length uint32
|
||||
ID [backend.IDSize]byte
|
||||
}
|
||||
|
||||
// Finalize writes the header for all added blobs and finalizes the pack.
|
||||
// Returned are the complete number of bytes written, including the header.
|
||||
// After Finalize() has finished, the ID of this pack can be obtained by
|
||||
// calling ID().
|
||||
func (p *Packer) Finalize() (int64, error) {
|
||||
p.m.Lock()
|
||||
defer p.m.Unlock()
|
||||
|
||||
// n tracks the number of bytes written for the header
|
||||
var n int64 = int64(p.bytes)
|
||||
|
||||
// create writer to encrypt header
|
||||
wr := crypto.EncryptTo(p.k, p.hw)
|
||||
|
||||
// write header
|
||||
for _, b := range p.blobs {
|
||||
entry := headerEntry{
|
||||
Type: b.Type,
|
||||
Length: b.Length,
|
||||
}
|
||||
copy(entry.ID[:], b.ID)
|
||||
|
||||
err := binary.Write(wr, binary.LittleEndian, entry)
|
||||
if err != nil {
|
||||
return int64(n), err
|
||||
}
|
||||
|
||||
n += int64(entrySize)
|
||||
}
|
||||
|
||||
// finalize encrypted header
|
||||
err := wr.Close()
|
||||
if err != nil {
|
||||
return int64(n), err
|
||||
}
|
||||
|
||||
// account for crypto overhead
|
||||
n += crypto.Extension
|
||||
|
||||
// write length
|
||||
err = binary.Write(p.hw, binary.LittleEndian, uint32(len(p.blobs)*entrySize+crypto.Extension))
|
||||
if err != nil {
|
||||
return int64(n), err
|
||||
}
|
||||
n += int64(binary.Size(uint32(0)))
|
||||
|
||||
p.bytes = uint(n)
|
||||
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// ID returns the ID of all data written so far.
|
||||
func (p *Packer) ID() backend.ID {
|
||||
p.m.Lock()
|
||||
defer p.m.Unlock()
|
||||
|
||||
return p.hw.Sum(nil)
|
||||
}
|
||||
|
||||
// Size returns the number of bytes written so far.
|
||||
func (p *Packer) Size() uint {
|
||||
p.m.Lock()
|
||||
defer p.m.Unlock()
|
||||
|
||||
return p.bytes
|
||||
}
|
||||
|
||||
// Count returns the number of blobs in this packer.
|
||||
func (p *Packer) Count() int {
|
||||
p.m.Lock()
|
||||
defer p.m.Unlock()
|
||||
|
||||
return len(p.blobs)
|
||||
}
|
||||
|
||||
// Blobs returns the slice of blobs that have been written.
|
||||
func (p *Packer) Blobs() []Blob {
|
||||
p.m.Lock()
|
||||
defer p.m.Unlock()
|
||||
|
||||
return p.blobs
|
||||
}
|
||||
|
||||
// Writer returns the underlying writer.
|
||||
func (p *Packer) Writer() io.Writer {
|
||||
return p.wr
|
||||
}
|
||||
|
||||
func (p *Packer) String() string {
|
||||
return fmt.Sprintf("<Packer %d blobs, %d bytes>", len(p.blobs), p.bytes)
|
||||
}
|
||||
|
||||
// Unpacker is used to read individual blobs from a pack.
|
||||
type Unpacker struct {
|
||||
rd io.ReadSeeker
|
||||
Entries []Blob
|
||||
k *crypto.Key
|
||||
}
|
||||
|
||||
// NewUnpacker returns a pointer to Unpacker which can be used to read
|
||||
// individual Blobs from a pack.
|
||||
func NewUnpacker(k *crypto.Key, entries []Blob, rd io.ReadSeeker) (*Unpacker, error) {
|
||||
var err error
|
||||
ls := binary.Size(uint32(0))
|
||||
|
||||
// reset to the end to read header length
|
||||
_, err = rd.Seek(-int64(ls), 2)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("seeking to read header length failed: %v", err)
|
||||
}
|
||||
|
||||
// read length
|
||||
var l uint32
|
||||
err = binary.Read(rd, binary.LittleEndian, &l)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("reading header length failed: %v", err)
|
||||
}
|
||||
|
||||
// reset to the beginning of the header
|
||||
_, err = rd.Seek(-int64(ls)-int64(l), 2)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("seeking to read header length failed: %v", err)
|
||||
}
|
||||
|
||||
// read header
|
||||
hrd, err := crypto.DecryptFrom(k, io.LimitReader(rd, int64(l)))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if entries == nil {
|
||||
pos := uint(0)
|
||||
for {
|
||||
e := headerEntry{}
|
||||
err = binary.Read(hrd, binary.LittleEndian, &e)
|
||||
if err == io.EOF {
|
||||
break
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
entries = append(entries, Blob{
|
||||
Type: e.Type,
|
||||
Length: e.Length,
|
||||
ID: e.ID[:],
|
||||
Offset: pos,
|
||||
})
|
||||
|
||||
pos += uint(e.Length)
|
||||
}
|
||||
}
|
||||
|
||||
p := &Unpacker{
|
||||
rd: rd,
|
||||
k: k,
|
||||
Entries: entries,
|
||||
}
|
||||
|
||||
return p, nil
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
package pack_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"testing"
|
||||
|
||||
"github.com/restic/restic/backend"
|
||||
"github.com/restic/restic/crypto"
|
||||
"github.com/restic/restic/pack"
|
||||
. "github.com/restic/restic/test"
|
||||
)
|
||||
|
||||
var lengths = []int{23, 31650, 25860, 10928, 13769, 19862, 5211, 127, 13690, 30231}
|
||||
|
||||
// var lengths = []int{200}
|
||||
|
||||
func TestCreatePack(t *testing.T) {
|
||||
type Buf struct {
|
||||
data []byte
|
||||
id backend.ID
|
||||
}
|
||||
|
||||
bufs := []Buf{}
|
||||
|
||||
for _, l := range lengths {
|
||||
b := make([]byte, l)
|
||||
_, err := io.ReadFull(rand.Reader, b)
|
||||
OK(t, err)
|
||||
h := sha256.Sum256(b)
|
||||
bufs = append(bufs, Buf{data: b, id: h[:]})
|
||||
}
|
||||
|
||||
file := bytes.NewBuffer(nil)
|
||||
|
||||
// create random keys
|
||||
k := crypto.NewKey()
|
||||
|
||||
// pack blobs
|
||||
p := pack.NewPacker(k, file)
|
||||
for _, b := range bufs {
|
||||
p.Add(pack.Tree, b.id, bytes.NewReader(b.data))
|
||||
}
|
||||
|
||||
// write file
|
||||
n, err := p.Finalize()
|
||||
OK(t, err)
|
||||
|
||||
written := 0
|
||||
// data
|
||||
for _, l := range lengths {
|
||||
written += l
|
||||
}
|
||||
// header length
|
||||
written += binary.Size(uint32(0))
|
||||
// header
|
||||
written += len(lengths) * (binary.Size(pack.BlobType(0)) + binary.Size(uint32(0)) + backend.IDSize)
|
||||
// header crypto
|
||||
written += crypto.Extension
|
||||
|
||||
// check length
|
||||
Equals(t, int64(written), n)
|
||||
Equals(t, uint(written), p.Size())
|
||||
|
||||
// read and parse it again
|
||||
rd := bytes.NewReader(file.Bytes())
|
||||
np, err := pack.NewUnpacker(k, nil, rd)
|
||||
OK(t, err)
|
||||
Equals(t, len(np.Entries), len(bufs))
|
||||
|
||||
for i, b := range bufs {
|
||||
e := np.Entries[i]
|
||||
Equals(t, b.id, e.ID)
|
||||
|
||||
brd, err := e.GetReader(rd)
|
||||
OK(t, err)
|
||||
data, err := ioutil.ReadAll(brd)
|
||||
OK(t, err)
|
||||
|
||||
Assert(t, bytes.Equal(b.data, data),
|
||||
"data for blob %v doesn't match", i)
|
||||
}
|
||||
}
|
||||
|
||||
var blobTypeJson = []struct {
|
||||
t pack.BlobType
|
||||
res string
|
||||
}{
|
||||
{pack.Data, `"data"`},
|
||||
{pack.Tree, `"tree"`},
|
||||
}
|
||||
|
||||
func TestBlobTypeJSON(t *testing.T) {
|
||||
for _, test := range blobTypeJson {
|
||||
// test serialize
|
||||
buf, err := json.Marshal(test.t)
|
||||
OK(t, err)
|
||||
Equals(t, test.res, string(buf))
|
||||
|
||||
// test unserialize
|
||||
var v pack.BlobType
|
||||
err = json.Unmarshal([]byte(test.res), &v)
|
||||
OK(t, err)
|
||||
Equals(t, test.t, v)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user