ayo/kitty
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Wrap the rsync Go API to use with byte streams and arbitrary hash functions

Browse source
This commit is contained in:
Kovid Goyal 2023-07-02 20:27:17 +05:30
parent d539620c9d
commit 7e12972414
No known key found for this signature in database
GPG key ID: 06BC317B515ACE7C
4 changed files with 278 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
go.mod
View file

@ -13,6 +13,7 @@ require (
github.com/jamesruan/go-rfc1924 v0.0.0-20170108144916-2767ca7c638f
github.com/seancfoley/ipaddress-go v1.5.4
github.com/shirou/gopsutil/v3 v3.23.6
github.com/zeebo/xxh3 v1.0.2
golang.org/x/exp v0.0.0-20230321023759-10a507213a29
golang.org/x/sys v0.10.0
golang.org/x/image v0.9.0
@ -21,6 +22,7 @@ require (
require (
github.com/go-ole/go-ole v1.2.6 // indirect
github.com/klauspost/cpuid/v2 v2.0.9 // indirect
github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a // indirect
github.com/power-devops/perfstat v0.0.0-20221212215047-62379fc7944b // indirect
github.com/seancfoley/bintree v1.2.1 // indirect

5
go.sum
View file

@ -24,6 +24,8 @@ github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUq
github.com/jamesruan/go-rfc1924 v0.0.0-20170108144916-2767ca7c638f h1:Ko4+g6K16vSyUrtd/pPXuQnWsiHe5BYptEtTxfwYwCc=
github.com/jamesruan/go-rfc1924 v0.0.0-20170108144916-2767ca7c638f/go.mod h1:eHzfhOKbTGJEGPSdMHzU6jft192tHHt2Bu2vIZArvC0=
github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a h1:N9zuLhTvBSRt0gWSiJswwQ2HqDmtX/ZCDJURnKUt1Ik=
github.com/lufia/plan9stats v0.0.0-20230326075908-cb1d2100619a/go.mod h1:JKx41uQRwqlTZabZc+kILPrO/3jlKnQ2Z8b7YiVw5cE=
@ -56,6 +58,9 @@ github.com/tklauser/numcpus v0.6.0/go.mod h1:FEZLMke0lhOUG6w2JadTzp0a+Nl8PF/GFkQ
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
github.com/yusufpapurcu/wmi v1.2.3 h1:E1ctvB7uKFMOJw3fdOW32DwGE9I7t++CRUEMKvFoFiw=
github.com/yusufpapurcu/wmi v1.2.3/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/exp v0.0.0-20230321023759-10a507213a29 h1:ooxPy7fPvB4kwsA2h+iBNHkAbp/4JxTSwCmvdjEYmug=

129
tools/rsync/algorithm.go
View file

@ -12,9 +12,12 @@ package rsync
import (
"bytes"
"crypto/md5"
"encoding/binary"
"fmt"
"hash"
"io"
"kitty/tools/utils"
)
// If no BlockSize is specified in the RSync instance, this value is used.
@ -42,6 +45,40 @@ type Operation struct {
Data []byte
}
var bin = binary.LittleEndian
func (self Operation) Serialize() string {
ans := make([]byte, 24+len(self.Data))
bin.PutUint32(ans, uint32(len(self.Data)))
bin.PutUint32(ans[4:], uint32(self.Type))
bin.PutUint64(ans[8:], self.BlockIndex)
bin.PutUint64(ans[16:], self.BlockIndexEnd)
copy(ans[24:], self.Data)
return utils.UnsafeBytesToString(ans)
}
func (self *Operation) Unserialize(data []byte) (n int, err error) {
if len(data) < 24 {
return -1, fmt.Errorf("record too small to be an Operation %d < 24", len(data))
}
dlen := int(bin.Uint32(data))
if len(data) < 24+dlen {
return -1, fmt.Errorf("record too small to be an Operation %d < %d", len(data), 24+dlen)
}
t := OpType(bin.Uint32(data[4:]))
switch t {
case OpBlock, OpData, OpHash, OpBlockRange:
self.Type = t
default:
return 0, fmt.Errorf("record has unknown operation type: %d", t)
}
self.BlockIndex = bin.Uint64(data[8:])
self.BlockIndexEnd = bin.Uint64(data[16:])
self.Data = data[24 : 24+dlen]
n = 24 + dlen
return
}
// Signature hash item generated from target.
type BlockHash struct {
Index uint64
@ -49,6 +86,24 @@ type BlockHash struct {
WeakHash uint32
}
func (self BlockHash) Serialize() string {
ans := make([]byte, 12+len(self.StrongHash))
bin.PutUint64(ans, self.Index)
bin.PutUint32(ans[8:], self.WeakHash)
copy(ans[12:], self.StrongHash)
return utils.UnsafeBytesToString(ans)
}
func (self BlockHash) Unserialize(data []byte, hash_size int) (err error) {
if len(data) < 12+hash_size {
return fmt.Errorf("record too small to be a BlockHash: %d < %d", len(data), 12+hash_size)
}
self.Index = bin.Uint64(data)
self.WeakHash = bin.Uint32(data[8:])
self.StrongHash = data[12 : 12+hash_size]
return
}
// Write signatures as they are generated.
type SignatureWriter func(bl BlockHash) error
type OperationWriter func(op Operation) error
@ -60,7 +115,7 @@ type RSync struct {
BlockSize int
MaxDataOp int
// If this is nil an MD5 hash is used.
// This must be non-nil before using any functions
UniqueHasher hash.Hash
buffer []byte
@ -68,12 +123,10 @@ type RSync struct {
// If the target length is known the number of hashes in the
// signature can be determined.
func (r *RSync) BlockHashCount(targetLength int) (count int) {
if r.BlockSize <= 0 {
r.BlockSize = DefaultBlockSize
}
count = (targetLength / r.BlockSize)
if targetLength%r.BlockSize != 0 {
func (r *RSync) BlockHashCount(targetLength int64) (count int64) {
bs := int64(r.BlockSize)
count = targetLength / bs
if targetLength%bs != 0 {
count++
}
return
@ -81,12 +134,6 @@ func (r *RSync) BlockHashCount(targetLength int) (count int) {
// Calculate the signature of target.
func (r *RSync) CreateSignature(target io.Reader, sw SignatureWriter) error {
if r.BlockSize <= 0 {
r.BlockSize = DefaultBlockSize
}
if r.UniqueHasher == nil {
r.UniqueHasher = md5.New()
}
var err error
var n int
@ -124,10 +171,7 @@ func (r *RSync) CreateSignature(target io.Reader, sw SignatureWriter) error {
}
// Apply the difference to the target.
func (r *RSync) ApplyDelta(alignedTarget io.Writer, target io.ReadSeeker, ops chan Operation) error {
if r.BlockSize <= 0 {
r.BlockSize = DefaultBlockSize
}
func (r *RSync) ApplyDelta(alignedTarget io.Writer, target io.ReadSeeker, op Operation) error {
var err error
var n int
var block []byte
@ -154,34 +198,32 @@ func (r *RSync) ApplyDelta(alignedTarget io.Writer, target io.ReadSeeker, ops ch
return nil
}
for op := range ops {
switch op.Type {
case OpBlockRange:
for i := op.BlockIndex; i <= op.BlockIndexEnd; i++ {
err = writeBlock(Operation{
Type: OpBlock,
BlockIndex: i,
})
if err != nil {
if err == io.EOF {
break
}
return err
}
}
case OpBlock:
err = writeBlock(op)
switch op.Type {
case OpBlockRange:
for i := op.BlockIndex; i <= op.BlockIndexEnd; i++ {
err = writeBlock(Operation{
Type: OpBlock,
BlockIndex: i,
})
if err != nil {
if err == io.EOF {
break
}
return err
}
case OpData:
_, err = alignedTarget.Write(op.Data)
if err != nil {
return err
}
case OpBlock:
err = writeBlock(op)
if err != nil {
if err == io.EOF {
break
}
return err
}
case OpData:
_, err = alignedTarget.Write(op.Data)
if err != nil {
return err
}
}
return nil
@ -193,15 +235,6 @@ func (r *RSync) ApplyDelta(alignedTarget io.Writer, target io.ReadSeeker, ops ch
// data is reused. The sourceSum create a complete hash sum of the source if
// present.
func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops OperationWriter) (err error) {
if r.BlockSize <= 0 {
r.BlockSize = DefaultBlockSize
}
if r.MaxDataOp <= 0 {
r.MaxDataOp = DefaultMaxDataOp
}
if r.UniqueHasher == nil {
r.UniqueHasher = md5.New()
}
minBufferSize := (r.BlockSize * 2) + (r.MaxDataOp)
if len(r.buffer) < minBufferSize {
r.buffer = make([]byte, minBufferSize)

190
tools/rsync/api.go Normal file
View file

@ -0,0 +1,190 @@
// License: GPLv3 Copyright: 2023, Kovid Goyal, <kovid at kovidgoyal.net>
package rsync
import (
"encoding/json"
"fmt"
"io"
"math"
"os"
"strings"
"github.com/zeebo/xxh3"
"kitty/tools/utils"
)
var _ = fmt.Print
const MaxBlockSize int = 256 * 1024
type Api struct {
rsync RSync
signature []BlockHash
delta_output io.Writer
delta_input io.ReadSeeker
delta_buffer []byte
Strong_hash_name, Weak_hash_name string
}
type SignatureHeader struct {
Weak_hash_name string `json:"weak_hash,omitempty"`
Strong_hash_name string `json:"strong_hash,omitempty"`
Block_size int `json:"block_size,omitempty"`
}
func (self *Api) StartDelta(delta_output io.Writer, delta_input io.ReadSeeker) {
self.delta_output = delta_output
self.delta_input = delta_input
self.delta_buffer = self.delta_buffer[:0]
}
func (self *Api) update_delta(data []byte) (consumed int, err error) {
op := Operation{}
for len(data) > 0 {
n, uerr := op.Unserialize(data)
if uerr == nil {
consumed += n
data = data[n:]
if err = self.rsync.ApplyDelta(self.delta_output, self.delta_input, op); err != nil {
return
}
} else {
if n < 0 {
return consumed, nil
}
return consumed, uerr
}
}
return
}
func (self *Api) UpdateDelta(data []byte) (err error) {
if len(self.delta_buffer) == 0 {
consumed, err := self.update_delta(data)
if err != nil {
return err
}
data = data[consumed:]
if len(data) > 0 {
self.delta_buffer = append(self.delta_buffer, data...)
}
} else {
self.delta_buffer = append(self.delta_buffer, data...)
consumed, err := self.update_delta(self.delta_buffer)
if err != nil {
return err
}
leftover := len(self.delta_buffer) - consumed
copy(self.delta_buffer, self.delta_buffer[consumed:])
self.delta_buffer = self.delta_buffer[:leftover]
}
return
}
func (self *Api) FinishDelta() (err error) {
if len(self.delta_buffer) > 0 {
data := self.delta_buffer
self.delta_buffer = self.delta_buffer[:0]
if err = self.UpdateDelta(data); err != nil {
return err
}
if len(self.delta_buffer) > 0 {
return fmt.Errorf("There are %d leftover bytes in the delta", len(self.delta_buffer))
}
}
self.delta_input = nil
self.delta_output = nil
return
}
func (self *Api) CreateDelta(src io.Reader, output_callback func(string) error) (err error) {
if len(self.signature) == 0 {
return fmt.Errorf("Cannot call CreateDelta() before loading a signature")
}
self.rsync.CreateDelta(src, self.signature, func(op Operation) error {
return output_callback(op.Serialize())
})
return
}
func (self *Api) setup_from_signature(signature string) (err error) {
h := SignatureHeader{}
dec := json.NewDecoder(strings.NewReader(signature))
if err = dec.Decode(&h); err != nil {
return fmt.Errorf("rsync signature header not valid JSON with error: %w", err)
}
signature = signature[dec.InputOffset():]
if h.Block_size == 0 {
return fmt.Errorf("rsync signature header has no or zero block size")
}
if h.Block_size > MaxBlockSize {
return fmt.Errorf("rsync signature header has too large block size %d > %d", h.Block_size, MaxBlockSize)
}
self.rsync.BlockSize = h.Block_size
self.rsync.MaxDataOp = 10 * h.Block_size
if h.Weak_hash_name != "" && h.Weak_hash_name != "beta" {
return fmt.Errorf("rsync signature header has unknown weak hash algorithm: %#v", h.Weak_hash_name)
}
self.Weak_hash_name = h.Weak_hash_name
switch h.Strong_hash_name {
case "", "xxh3":
self.rsync.UniqueHasher = xxh3.New()
self.Strong_hash_name = "xxh3"
default:
return fmt.Errorf("rsync signature header has unknown strong hash algorithm: %#v", h.Strong_hash_name)
}
self.signature = make([]BlockHash, 0, 64)
hash_size := self.rsync.UniqueHasher.Size()
block_hash_size := self.rsync.UniqueHasher.Size() + 12
for ; len(signature) >= block_hash_size; signature = signature[block_hash_size:] {
data := utils.UnsafeStringToBytes(signature[:block_hash_size])
bl := BlockHash{}
bl.Unserialize(data, hash_size)
self.signature = append(self.signature, bl)
}
return
}
func NewFromSignature(signature string) (ans *Api, err error) {
ans = &Api{}
err = ans.setup_from_signature(signature)
return
}
func New(src io.Reader) (ans *Api, err error) {
bs := DefaultBlockSize
var sz int64
if v, ok := src.(io.ReadSeeker); ok {
if pos, err := v.Seek(0, os.SEEK_CUR); err != nil {
return nil, err
} else {
if sz, err = v.Seek(0, os.SEEK_END); err != nil {
sz -= pos
bs = int(math.Round(math.Sqrt(float64(sz))))
if _, err = v.Seek(pos, os.SEEK_SET); err != nil {
return nil, err
}
}
}
}
ans = &Api{Weak_hash_name: "beta", Strong_hash_name: "xxh3"}
ans.rsync.BlockSize = utils.Min(bs, MaxBlockSize)
ans.rsync.UniqueHasher = xxh3.New()
if ans.rsync.UniqueHasher.BlockSize() > 0 && ans.rsync.UniqueHasher.BlockSize() < ans.rsync.BlockSize {
ans.rsync.BlockSize = (ans.rsync.BlockSize / ans.rsync.UniqueHasher.BlockSize()) * ans.rsync.UniqueHasher.BlockSize()
}
ans.rsync.MaxDataOp = ans.rsync.BlockSize * 10
if sz > 0 {
ans.signature = make([]BlockHash, 0, ans.rsync.BlockHashCount(sz))
}
err = ans.rsync.CreateSignature(src, func(bl BlockHash) error {
ans.signature = append(ans.signature, bl)
return nil
})
return
}