/// Content-defined chunking.
pub(all) struct ChunkerStats {
input_bytes : Int
chunk_count : Int
smallest : Int
largest : Int
average : Int
boundaries : Array[Int]
} derive(Show)
pub fn ChunkerStats::empty() -> ChunkerStats {
{ input_bytes: 0, chunk_count: 0, smallest: 0, largest: 0, average: 0, boundaries: [] }
}
pub fn ChunkerStats::input(self : ChunkerStats) -> Int { self.input_bytes }
pub fn ChunkerStats::count(self : ChunkerStats) -> Int { self.chunk_count }
pub fn ChunkerStats::min_size(self : ChunkerStats) -> Int { self.smallest }
pub fn ChunkerStats::max_size(self : ChunkerStats) -> Int { self.largest }
pub fn ChunkerStats::mean(self : ChunkerStats) -> Int { self.average }
pub fn ChunkerStats::boundary_offsets(self : ChunkerStats) -> Array[Int] { self.boundaries.copy() }
fn boundary_mask(average : Int) -> UInt {
let mut power = 1
let mut bits = 0
while power < average {
power = power * 2
bits = bits + 1
}
if bits <= 1 { 1U } else { (1U << (bits - 1)) - 1U }
}
fn gear_value(index : Int, seed : UInt) -> UInt {
let mut x = seed ^ index.reinterpret_as_uint()
x = x ^ (x >> 16)
x = x * 0x7feb352dU
x = x ^ (x >> 15)
x = x * 0x846ca68bU
x ^ (x >> 16)
}
fn update_roll(hash : UInt, byte : Byte, position : Int, seed : UInt) -> UInt {
let table_value = gear_value(byte.to_int(), seed)
let mixed = (hash << 1) | (hash >> 31) ^ table_value
mixed + position.reinterpret_as_uint() * 0x9e3779b9U
}
fn should_cut(hash : UInt, size : Int, cfg : ChunkerConfig, mask : UInt) -> Bool {
size >= cfg.minimum() && ((hash & mask) == 0U || size >= cfg.maximum())
}
fn take_slice(source : Array[Byte], start : Int, end : Int) -> Array[Byte] {
let out = []
for i in start.. Array[Chunk] {
let _ = cfg.validate()
let chunks = []
let mut start = 0
let mut hash = cfg.seed
let mask = boundary_mask(cfg.average())
let mut ordinal = 0
for position, byte in source {
hash = update_roll(hash, byte, position, cfg.seed)
let size = position + 1 - start
if should_cut(hash, size, cfg, mask) {
let data = take_slice(source, start, position + 1)
chunks.push(Chunk::new(ordinal, Span::new(start, position + 1), data))
ordinal = ordinal + 1
start = position + 1
hash = cfg.seed ^ ordinal.reinterpret_as_uint()
}
}
if start < source.length() {
chunks.push(Chunk::new(ordinal, Span::new(start, source.length()), take_slice(source, start, source.length())))
}
chunks
}
/// Incremental chunker for readers that arrive in arbitrary input fragments.
pub(all) struct StreamingChunker {
config : ChunkerConfig
mut buffer : Array[Byte]
mut total_seen : Int
mut next_ordinal : Int
mut rolling : UInt
mut boundaries : Array[Int]
}
pub fn StreamingChunker::new(config : ChunkerConfig) -> StreamingChunker {
{
config,
buffer: [],
total_seen: 0,
next_ordinal: 0,
rolling: config.seed,
boundaries: [],
}
}
pub fn StreamingChunker::bytes_seen(self : StreamingChunker) -> Int { self.total_seen }
pub fn StreamingChunker::pending_size(self : StreamingChunker) -> Int { self.buffer.length() }
pub fn StreamingChunker::config(self : StreamingChunker) -> ChunkerConfig { self.config }
fn StreamingChunker::emit_if_ready(self : StreamingChunker, final_chunk : Bool) -> Array[Chunk] {
let output = []
let mask = boundary_mask(self.config.average())
let mut cut = -1
if final_chunk && self.buffer.length() > 0 {
cut = self.buffer.length()
} else if self.buffer.length() >= self.config.minimum() {
for i in 0.. 0 {
let data = take_slice(self.buffer, 0, cut)
let start = self.total_seen - self.buffer.length()
output.push(Chunk::new(self.next_ordinal, Span::new(start, start + cut), data))
self.next_ordinal = self.next_ordinal + 1
self.boundaries.push(start + cut)
self.buffer = take_slice(self.buffer, cut, self.buffer.length())
}
output
}
/// Feed one fragment. It may return zero or more complete chunks.
pub fn StreamingChunker::push(self : StreamingChunker, fragment : Array[Byte]) -> Array[Chunk] {
for b in fragment { self.buffer.push(b) }
self.total_seen = self.total_seen + fragment.length()
let output = []
while self.buffer.length() >= self.config.maximum() {
let emitted = self.emit_if_ready(false)
if emitted.length() == 0 {
let data = take_slice(self.buffer, 0, self.config.maximum())
let start = self.total_seen - self.buffer.length()
output.push(Chunk::new(self.next_ordinal, Span::new(start, start + data.length()), data))
self.next_ordinal = self.next_ordinal + 1
self.buffer = take_slice(self.buffer, data.length(), self.buffer.length())
} else {
for c in emitted { output.push(c) }
}
}
output
}
/// Finish the stream and return the final partial chunk.
pub fn StreamingChunker::finish(self : StreamingChunker) -> Array[Chunk] {
let output = []
while self.buffer.length() > 0 {
let emitted = self.emit_if_ready(true)
if emitted.length() == 0 { break }
for c in emitted { output.push(c) }
}
output
}
pub fn StreamingChunker::finish_all(self : StreamingChunker) -> Array[Chunk] {
let out = self.finish()
out
}
pub fn chunk_stats(chunks : Array[Chunk], input_bytes : Int) -> ChunkerStats {
if chunks.length() == 0 {
return { ..ChunkerStats::empty(), input_bytes }
}
let mut smallest = chunks[0].size()
let mut largest = chunks[0].size()
let mut total = 0
let boundaries = []
for chunk in chunks {
let size = chunk.size()
if size < smallest { smallest = size }
if size > largest { largest = size }
total = total + size
boundaries.push(chunk.range().end)
}
{ input_bytes, chunk_count: chunks.length(), smallest, largest, average: total / chunks.length(), boundaries }
}
pub fn chunk_histogram(chunks : Array[Chunk], buckets : Int) -> Array[Int] {
let count = if buckets <= 0 { 1 } else { buckets }
let out = Array::make(count, 0)
for chunk in chunks {
let slot = chunk.size() % count
out[slot] = out[slot] + 1
}
out
}
pub fn chunk_boundaries(chunks : Array[Chunk]) -> Array[Int] {
chunks.map(c => c.range().end)
}
pub fn chunk_total_size(chunks : Array[Chunk]) -> Int {
let mut total = 0
for c in chunks { total = total + c.size() }
total
}