diff --git a/go.mod b/go.mod index 70cbd4a8f..02731fca0 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 20b228993..0367aeced 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/tools/rsync/algorithm.go b/tools/rsync/algorithm.go index be85f7444..ad47380d7 100644 --- a/tools/rsync/algorithm.go +++ b/tools/rsync/algorithm.go @@ -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) diff --git a/tools/rsync/api.go b/tools/rsync/api.go new file mode 100644 index 000000000..2283a8855 --- /dev/null +++ b/tools/rsync/api.go @@ -0,0 +1,190 @@ +// License: GPLv3 Copyright: 2023, Kovid Goyal, + +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 +}