diff --git a/tools/rsync/algorithm.go b/tools/rsync/algorithm.go index 539f3cfa7..c02530f96 100644 --- a/tools/rsync/algorithm.go +++ b/tools/rsync/algorithm.go @@ -17,6 +17,8 @@ import ( "hash" "io" "os" + + "golang.org/x/exp/slices" ) // If no BlockSize is specified in the RSync instance, this value is used. @@ -47,35 +49,73 @@ type Operation struct { var bin = binary.LittleEndian func (self Operation) Serialize() []byte { - 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) + var ans []byte + switch self.Type { + case OpBlock: + ans = make([]byte, 9) + bin.PutUint64(ans[1:], self.BlockIndex) + case OpBlockRange: + ans = make([]byte, 13) + bin.PutUint64(ans[1:], self.BlockIndex) + bin.PutUint32(ans[9:], uint32(self.BlockIndexEnd-self.BlockIndex)) + case OpHash: + ans = make([]byte, 3+len(self.Data)) + bin.PutUint16(ans[1:], uint16(len(self.Data))) + copy(ans[3:], self.Data) + case OpData: + ans = make([]byte, 5+len(self.Data)) + bin.PutUint32(ans[1:], uint32(len(self.Data))) + copy(ans[5:], self.Data) + } + ans[0] = byte(self.Type) return 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)) + if len(data) < 1 { + return -1, io.ErrShortBuffer } - 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 + switch OpType(data[0]) { + case OpBlock: + n = 9 + if len(data) < n { + return -1, io.ErrShortBuffer + } + self.BlockIndex = bin.Uint64(data[1:]) + case OpBlockRange: + n = 13 + if len(data) < n { + return -1, io.ErrShortBuffer + } + self.BlockIndex = bin.Uint64(data[1:]) + self.BlockIndexEnd = self.BlockIndex + uint64(bin.Uint32(data[9:])) + case OpHash: + n = 3 + if len(data) < n { + return -1, io.ErrShortBuffer + } + sz := int(bin.Uint16(data[1:])) + n += sz + if len(data) < n { + return -1, io.ErrShortBuffer + } + self.Data = data[3:n] + case OpData: + n = 5 + if len(data) < n { + return -1, io.ErrShortBuffer + } + sz := int(bin.Uint32(data[1:])) + n += sz + if len(data) < n { + return -1, io.ErrShortBuffer + } + self.Data = data[5:n] default: - return 0, fmt.Errorf("record has unknown operation type: %d", t) + return 0, fmt.Errorf("record has unknown operation type: %d", data[0]) } - self.BlockIndex = bin.Uint64(data[8:]) - self.BlockIndexEnd = bin.Uint64(data[16:]) - n = 24 + dlen - self.Data = data[24:n] - return n, nil + self.Type = OpType(data[0]) + return } // Signature hash item generated from target. @@ -235,6 +275,248 @@ func (r *RSync) set_buffer_to_size(sz int) { } } +type section struct { + tail int + head int +} + +type node struct { + op *Operation + next *node +} + +type list struct { + head *node +} + +func (self *list) push_back(op *Operation) { + n := &node{op: op} + n.next = self.head + self.head = n +} + +func (self *list) is_empty() bool { return self.head == nil } + +func (self *list) front() *Operation { + for c := self.head; c != nil; c = c.next { + if c.next == nil { + return c.op + } + } + return nil +} + +func (self *list) pop_front() *Operation { + c := self.head + var prev *node + for c != nil { + if c.next == nil { + if prev == nil { + self.head = nil + } else { + prev.next = nil + } + return c.op + } + prev = c + c = c.next + } + return nil +} + +type Diff struct { + buffer []byte + // A single β hash may correlate with many unique hashes. + hash_lookup map[uint32][]BlockHash + source io.Reader + max_data_op int + hasher hash.Hash + hash_buf []byte + + data, sum section + block_size int + n, valid_to int + alpha_pop, alpha_push, beta, beta1, beta2 uint32 + finished, rolling bool + + pending_op *Operation + ready_ops list +} + +func (self *Diff) Operation() *Operation { + return self.ready_ops.front() +} + +func (self *Diff) hash(b []byte) []byte { + self.hasher.Reset() + self.hasher.Write(b) + return self.hasher.Sum(self.hash_buf[:0]) +} + +// Combine OpBlock into OpBlockRange. To do this store the previous +// non-data operation and determine if it can be extended. +func (self *Diff) enqueue(op Operation) { + switch op.Type { + case OpBlock: + if self.pending_op != nil { + switch self.pending_op.Type { + case OpBlock: + if self.pending_op.BlockIndex+1 == op.BlockIndex { + self.pending_op = &Operation{ + Type: OpBlockRange, + BlockIndex: self.pending_op.BlockIndex, + BlockIndexEnd: op.BlockIndex, + } + return + } + case OpBlockRange: + if self.pending_op.BlockIndexEnd+1 == op.BlockIndex { + self.pending_op.BlockIndexEnd = op.BlockIndex + return + } + } + self.ready_ops.push_back(self.pending_op) + self.pending_op = nil + } + self.pending_op = &op + case OpData: + // Never save a data operation, as it would corrupt the buffer. + if self.pending_op != nil { + self.ready_ops.push_back(self.pending_op) + self.pending_op = nil + } + self.ready_ops.push_back(&op) + } + return + +} + +func (self *Diff) send_data() { + self.enqueue(Operation{Type: OpData, Data: slices.Clone(self.buffer[self.data.tail:self.data.head])}) + self.data.tail = self.data.head +} + +func (self *Diff) pump_till_op_available() error { + for self.ready_ops.is_empty() && !self.finished { + if err := self.read_at_least_one_operation(); err != nil { + return err + } + } + if self.finished && self.pending_op != nil { + self.ready_ops.push_back(self.pending_op) + self.pending_op = nil + } + return nil +} + +// See https://rsync.samba.org/tech_report/node4.html for the design of this algorithm +func (self *Diff) read_at_least_one_operation() error { + last_run := false + // Determine if the buffer should be extended. + if self.sum.tail+self.block_size > self.valid_to { + // Determine if the buffer should be wrapped. + if self.valid_to+self.block_size > len(self.buffer) { + // Before wrapping the buffer, send any trailing data off. + if self.data.tail < self.data.head { + self.send_data() + } + // Wrap the buffer. + l := self.valid_to - self.sum.tail + copy(self.buffer[:l], self.buffer[self.sum.tail:self.valid_to]) + + // Reset indexes. + self.valid_to = l + self.sum.tail = 0 + self.data.head = 0 + self.data.tail = 0 + } + + n, err := io.ReadAtLeast(self.source, self.buffer[self.valid_to:self.valid_to+self.block_size], self.block_size) + self.valid_to += n + if err != nil { + if err != io.EOF && err != io.ErrUnexpectedEOF { + return err + } + last_run = true + self.data.head = self.valid_to + } + if n == 0 { + if self.data.tail < self.data.head { + self.send_data() + } + } + } + + // Set the hash sum window head. Must either be a block size + // or be at the end of the buffer. + self.sum.head = min(self.sum.tail+self.block_size, self.valid_to) + + // Compute the rolling hash. + if !self.rolling { + self.beta, self.beta1, self.beta2 = βhash(self.buffer[self.sum.tail:self.sum.head]) + self.rolling = true + } else { + self.alpha_push = uint32(self.buffer[self.sum.head-1]) + self.beta1 = (self.beta1 - self.alpha_pop + self.alpha_push) % _M + self.beta2 = (self.beta2 - uint32(self.sum.head-self.sum.tail)*self.alpha_pop + self.beta1) % _M + self.beta = self.beta1 + _M*self.beta2 + } + + // Determine if there is a hash match. + found_hash := false + var block_index uint64 + if hh, ok := self.hash_lookup[self.beta]; ok && !last_run { + block_index, found_hash = findUniqueHash(hh, self.hash(self.buffer[self.sum.tail:self.sum.head])) + } + // Send data off if there is data available and a hash is found (so the buffer before it + // must be flushed first), or the data chunk size has reached it's maximum size (for buffer + // allocation purposes) or to flush the end of the data. + if self.data.tail < self.data.head && (found_hash || self.data.head-self.data.tail >= self.max_data_op || last_run) { + self.send_data() + } + + if found_hash { + self.enqueue(Operation{Type: OpBlock, BlockIndex: block_index}) + self.rolling = false + self.sum.tail += self.block_size + + // There is prior knowledge that any available data + // buffered will have already been sent. Thus we can + // assume data.head and data.tail are the same. + // May trigger "data wrap". + self.data.head = self.sum.tail + self.data.tail = self.sum.tail + } else { + // The following is for the next loop iteration, so don't try to calculate if last. + if !last_run && self.rolling { + self.alpha_pop = uint32(self.buffer[self.sum.tail]) + } + self.sum.tail += 1 + + // May trigger "data wrap". + self.data.head = self.sum.tail + } + if last_run { + self.finished = true + } + return nil +} + +func (r *RSync) CreateDiff(source io.Reader, signature []BlockHash) (ans *Diff) { + ans = &Diff{ + block_size: r.BlockSize, buffer: make([]byte, (r.BlockSize*2)+(r.MaxDataOp)), + hash_lookup: make(map[uint32][]BlockHash, len(signature)), + source: source, max_data_op: r.MaxDataOp, hasher: r.UniqueHasher, + hash_buf: make([]byte, 0, r.UniqueHasher.Size()), + } + for _, h := range signature { + key := h.WeakHash + ans.hash_lookup[key] = append(ans.hash_lookup[key], h) + } + + return +} + // Create the operation list to mutate the target signature into the source. // Any data operation from the OperationWriter must have the data copied out // within the span of the function; the data buffer underlying the operation @@ -251,11 +533,6 @@ func (r *RSync) CreateDelta(source io.Reader, signature []BlockHash, ops Operati hashLookup[key] = append(hashLookup[key], h) } - type section struct { - tail int - head int - } - var data, sum section var n, validTo int var αPop, αPush, β, β1, β2 uint32 @@ -449,9 +726,10 @@ func findUniqueHash(hh []BlockHash, hashValue []byte) (uint64, bool) { // Use a faster way to identify a set of bytes. func βhash(block []byte) (β uint32, β1 uint32, β2 uint32) { var a, b uint32 + sz := uint32(len(block) - 1) for i, val := range block { a += uint32(val) - b += (uint32(len(block)-1) - uint32(i) + 1) * uint32(val) + b += (sz - uint32(i) + 1) * uint32(val) } β = (a % _M) + (_M * (b % _M)) β1 = a % _M