diff --git a/Sources/CompressionCore/StreamingCompressor.swift b/Sources/CompressionCore/StreamingCompressor.swift index 7c86418..0814006 100644 --- a/Sources/CompressionCore/StreamingCompressor.swift +++ b/Sources/CompressionCore/StreamingCompressor.swift @@ -25,6 +25,12 @@ public protocol StreamingCompressor: ~Copyable, Sendable { mutating func finish( into output: inout OutputSpan ) throws(Failure) -> Bool + + /// Discard all stream state and start over, keeping `configuration`. + /// + /// Any data compressed after this is independent of what came before, so it + /// decodes on its own. A stream that was never finished is dropped. + mutating func reset() } extension StreamingCompressor where Self: ~Copyable { diff --git a/Sources/CompressionCore/StreamingDecompressor.swift b/Sources/CompressionCore/StreamingDecompressor.swift index 99eaae9..6488213 100644 --- a/Sources/CompressionCore/StreamingDecompressor.swift +++ b/Sources/CompressionCore/StreamingDecompressor.swift @@ -29,6 +29,12 @@ public protocol StreamingDecompressor: ~Copyable, Sendable { _ chunk: Span, into output: inout OutputSpan ) throws(Failure) -> Int + + /// Discard all stream state and start over, keeping `configuration`. + /// + /// Any data compressed after this is independent of what came before, so it + /// decodes on its own. A stream that was never finished is dropped. + mutating func reset() } extension StreamingDecompressor where Self: ~Copyable { diff --git a/Sources/Zlib/DeflateConfiguration.swift b/Sources/Zlib/DeflateConfiguration.swift index c580755..90d19ae 100644 --- a/Sources/Zlib/DeflateConfiguration.swift +++ b/Sources/Zlib/DeflateConfiguration.swift @@ -8,7 +8,14 @@ extension Deflate { public var memory: MemoryUsage public var strategy: Strategy - public init( + public init() { + self.level = .default + self.format = .zlib + self.memory = .default + self.strategy = .default + } + + package init( level: Level = .default, format: Format = .zlib, memory: MemoryUsage = .default, diff --git a/Sources/Zlib/DeflateStreamingCompressor.swift b/Sources/Zlib/DeflateStreamingCompressor.swift index ea7c8ae..f6e3d03 100644 --- a/Sources/Zlib/DeflateStreamingCompressor.swift +++ b/Sources/Zlib/DeflateStreamingCompressor.swift @@ -8,9 +8,13 @@ extension Deflate { @usableFromInline let stream: ZStreamBox + @usableFromInline + var isFinished: Bool + public init(configuration: Configuration = .default) { self.configuration = configuration self.stream = .init() + self.isFinished = false unsafe stream.value.zalloc = nil unsafe stream.value.zfree = nil unsafe stream.value.opaque = nil @@ -35,6 +39,8 @@ extension Deflate { _ chunk: Span, into output: inout OutputSpan ) throws(Deflate.Error) -> Int { + precondition(!self.isFinished, "Stream is finished, call reset if you want to start over") + guard !output.isFull else { return 0 } unsafe stream.value.avail_in = UInt32(chunk.count) @@ -68,31 +74,33 @@ extension Deflate { ) throws(Deflate.Error) -> Bool { guard output.freeCapacity > 0 else { return false } - let streamRef = stream - unsafe streamRef.value.avail_in = 0 - unsafe streamRef.value.next_in = nil + unsafe stream.value.avail_in = 0 + unsafe stream.value.next_in = nil try deflateStep(flag: CZLIB_Z_SYNC_FLUSH, into: &output) // zlib signals a complete flush by leaving output space unused. - return unsafe streamRef.value.avail_out != 0 + return unsafe stream.value.avail_out != 0 } @inlinable public func finish( into output: inout OutputSpan ) throws(Deflate.Error) -> Bool { - let streamRef = stream - unsafe streamRef.value.avail_in = 0 - - guard !output.isFull else { - return false - } + guard !self.isFinished else { return true } + unsafe stream.value.avail_in = 0 + guard !output.isFull else { return false } let status = try deflateStep(flag: CZLIB_Z_FINISH, into: &output) return status == CZLIB_Z_STREAM_END } + @inlinable + public mutating func reset() { + unsafe czlib_z_deflateReset(&self.stream.value) + self.isFinished = false + } + /// Run a single `deflate` call into the free space of `output`. /// /// `output` must have free capacity. diff --git a/Sources/Zlib/DeflateStreamingDecompressor.swift b/Sources/Zlib/DeflateStreamingDecompressor.swift index ec2c2ec..245956a 100644 --- a/Sources/Zlib/DeflateStreamingDecompressor.swift +++ b/Sources/Zlib/DeflateStreamingDecompressor.swift @@ -130,5 +130,12 @@ extension Deflate { return chunk.count - Int(unsafe stream.value.avail_in) } + + @inlinable + public mutating func reset() { + unsafe czlib_z_inflateReset(&self.stream.value) + self.decompressedBytesCount = 0 + self._isFinished = false + } } } diff --git a/Sources/Zstd/ZstdStreamingCompressor.swift b/Sources/Zstd/ZstdStreamingCompressor.swift index c58a8da..68b1d69 100644 --- a/Sources/Zstd/ZstdStreamingCompressor.swift +++ b/Sources/Zstd/ZstdStreamingCompressor.swift @@ -29,6 +29,8 @@ extension Zstd: StreamingCompressionAlgorithm { @inlinable public mutating func compress(_ chunk: Span, into output: inout OutputSpan) throws(Zstd.Error) -> Int { + precondition(!self.isFinished, "Stream is finished, call reset if you want to start over") + var consumed = 0 while consumed < chunk.count { @@ -87,13 +89,19 @@ extension Zstd: StreamingCompressionAlgorithm { return isFinished } + @inlinable + public mutating func reset() { + unsafe ZSTD_CCtx_reset(self.stream.value, ZSTD_reset_session_only) + self.isFinished = false + } + @inlinable public mutating func flush(into output: inout OutputSpan) throws(Zstd.Error) -> Bool { + precondition(!isFinished, "Stream is finished, call reset if you want to start over") guard !output.isFull else { return false } let free = output.freeCapacity let result = unsafe output.withUnsafeMutableBufferPointer { tail, initialisedCount -> CZstd_StreamResult? in - guard free > 0, let base = tail.baseAddress else { return nil } let result = unsafe CZstd_compressStream2NoInput( diff --git a/Sources/Zstd/ZstdStreamingDecompressor.swift b/Sources/Zstd/ZstdStreamingDecompressor.swift index 6a6e8b0..041be0e 100644 --- a/Sources/Zstd/ZstdStreamingDecompressor.swift +++ b/Sources/Zstd/ZstdStreamingDecompressor.swift @@ -116,5 +116,12 @@ extension Zstd: StreamingDecompressionAlgorithm { return consumed } + + @inlinable + public mutating func reset() { + unsafe ZSTD_DCtx_reset(self.stream.value, ZSTD_reset_session_only) + self.isFinished = false + self.decompressedBytesCount = 0 + } } } diff --git a/Tests/DeflateTests/DeflateResetTests.swift b/Tests/DeflateTests/DeflateResetTests.swift new file mode 100644 index 0000000..7139991 --- /dev/null +++ b/Tests/DeflateTests/DeflateResetTests.swift @@ -0,0 +1,136 @@ +import CompressionDeflate +import Testing + +private let payload: [UInt8] = { + var bytes = [UInt8]() + while bytes.count < 4_096 { bytes += Array("the quick brown fox jumps over the lazy dog, ".utf8) } + return Array(bytes.prefix(4_096)) +}() + +private let other: [UInt8] = Array(repeating: 0x5A, count: 2_048) + +private func bytes(_ span: Span) -> [UInt8] { + var out = [UInt8]() + out.reserveCapacity(span.count) + for i in span.indices { out.append(span[i]) } + return out +} + +@Suite("Deflate reset") +struct DeflateResetTests { + @Test("A stream compressed after reset decodes on its own") + func streamAfterResetIsSelfContained() throws { + var compressor = Deflate.StreamingCompressor() + var firstStream = [UInt8]() + var secondStream = [UInt8]() + + try compressor.compress(payload.span) { firstStream += bytes($0) } + try compressor.finish { firstStream += bytes($0) } + compressor.reset() + try compressor.compress(payload.span) { secondStream += bytes($0) } + try compressor.finish { secondStream += bytes($0) } + + let decompressor = Deflate.Decompressor() + #expect(try decompressor.decompress(firstStream) == payload) + #expect(try decompressor.decompress(secondStream) == payload) + } + + @Test("reset discards a stream that was never finished") + func resetDiscardsUnfinishedStream() throws { + var compressor = Deflate.StreamingCompressor() + var discarded = [UInt8]() + try compressor.compress(payload.span) { discarded += bytes($0) } + + compressor.reset() + + var stream = [UInt8]() + try compressor.compress(other.span) { stream += bytes($0) } + try compressor.finish { stream += bytes($0) } + + #expect(try Deflate.Decompressor().decompress(stream) == other) + } + + @Test("After reset the compressor behaves like a fresh one with the same configuration") + func resetKeepsConfiguration() throws { + let configuration = Deflate.CompressionConfiguration(level: .best, format: .gzip) + + var reused = Deflate.StreamingCompressor(configuration: configuration) + var throwaway = [UInt8]() + try reused.compress(payload.span) { throwaway += bytes($0) } + try reused.finish { throwaway += bytes($0) } + reused.reset() + var reusedStream = [UInt8]() + try reused.compress(payload.span) { reusedStream += bytes($0) } + try reused.finish { reusedStream += bytes($0) } + + var fresh = Deflate.StreamingCompressor(configuration: configuration) + var freshStream = [UInt8]() + try fresh.compress(payload.span) { freshStream += bytes($0) } + try fresh.finish { freshStream += bytes($0) } + + #expect(reusedStream == freshStream, "reset must not drop level, format or memory") + let decompressor = Deflate.Decompressor(configuration: .init(format: .gzip)) + #expect(try decompressor.decompress(reusedStream) == payload) + } + + @Test("After reset a second independent stream decodes") + func resetAllowsASecondStream() throws { + let stream = try Deflate.Compressor().compress(payload) + var decompressor = Deflate.StreamingDecompressor() + + var first = [UInt8]() + try decompressor.decompress(stream.span) { first += bytes($0) } + // #expect can't read a property off a ~Copyable value, so bind it first. + let finishedAfterFirst = decompressor.isFinished + #expect(first == payload) + #expect(finishedAfterFirst) + + decompressor.reset() + let finishedAfterReset = decompressor.isFinished + #expect(!finishedAfterReset) + + var second = [UInt8]() + try decompressor.decompress(stream.span) { second += bytes($0) } + let finishedAfterSecond = decompressor.isFinished + #expect(second == payload) + #expect(finishedAfterSecond) + } + + /// Without this, a reused decompressor keeps counting bytes from the stream + /// before it and trips `maxDecompressedSize` on a stream that is within the limit. + @Test("reset clears the byte count so maxDecompressedSize applies per stream") + func resetClearsTheByteCount() throws { + let stream = try Deflate.Compressor().compress(payload) + let configuration = Deflate.DecompressionConfiguration(maxDecompressedSize: payload.count) + var decompressor = Deflate.StreamingDecompressor(configuration: configuration) + + var first = [UInt8]() + try decompressor.decompress(stream.span) { first += bytes($0) } + #expect(first == payload) + + decompressor.reset() + + var second = [UInt8]() + try decompressor.decompress(stream.span) { second += bytes($0) } + #expect(second == payload) + } + + @Test("reset discards a partially fed stream") + func resetDiscardsPartialInput() throws { + let stream = try Deflate.Compressor().compress(payload) + var decompressor = Deflate.StreamingDecompressor() + + var partial = [UInt8]() + try decompressor.decompress(Array(stream.prefix(stream.count / 2)).span) { partial += bytes($0) } + let finishedAfterPartial = decompressor.isFinished + #expect(!finishedAfterPartial) + + decompressor.reset() + + var whole = [UInt8]() + try decompressor.decompress(stream.span) { whole += bytes($0) } + let finishedAtEnd = decompressor.isFinished + #expect(whole == payload) + #expect(finishedAtEnd) + } +} diff --git a/Tests/ZstdTests/ZstdResetTests.swift b/Tests/ZstdTests/ZstdResetTests.swift new file mode 100644 index 0000000..4755200 --- /dev/null +++ b/Tests/ZstdTests/ZstdResetTests.swift @@ -0,0 +1,140 @@ +import Testing +import Zstandard + +/// 4 KB of repetitive text. Both payloads are identical so that, if `reset()` +/// left the previous frame's match history in place, the second frame would be +/// encoded against it and stop being self-contained. +private let payload: [UInt8] = { + var bytes = [UInt8]() + while bytes.count < 4_096 { bytes += Array("the quick brown fox jumps over the lazy dog, ".utf8) } + return Array(bytes.prefix(4_096)) +}() + +private let other: [UInt8] = Array(repeating: 0x5A, count: 2_048) + +private func bytes(_ span: Span) -> [UInt8] { + var out = [UInt8]() + out.reserveCapacity(span.count) + for i in span.indices { out.append(span[i]) } + return out +} + +@Suite("Zstd reset") +struct ZstdResetTests { + @Test("A frame compressed after reset decodes on its own") + func frameAfterResetIsSelfContained() throws { + var compressor = Zstd.StreamingCompressor() + var firstFrame = [UInt8](), secondFrame = [UInt8]() + + try compressor.compress(payload.span) { firstFrame += bytes($0) } + try compressor.finish { firstFrame += bytes($0) } + compressor.reset() + try compressor.compress(payload.span) { secondFrame += bytes($0) } + try compressor.finish { secondFrame += bytes($0) } + + let decompressor = Zstd.Decompressor() + #expect(try decompressor.decompress(firstFrame) == payload) + #expect(try decompressor.decompress(secondFrame) == payload) + } + + @Test("reset discards a frame that was never finished") + func resetDiscardsUnfinishedFrame() throws { + var compressor = Zstd.StreamingCompressor() + var discarded = [UInt8]() + try compressor.compress(payload.span) { discarded += bytes($0) } + + compressor.reset() + + var frame = [UInt8]() + try compressor.compress(other.span) { frame += bytes($0) } + try compressor.finish { frame += bytes($0) } + + #expect(try Zstd.Decompressor().decompress(frame) == other) + } + + @Test("After reset the compressor behaves like a fresh one with the same configuration") + func resetKeepsConfiguration() throws { + let configuration = ZstdCompressionConfiguration(strategy: .btopt, level: 9) + + var reused = Zstd.StreamingCompressor(configuration: configuration) + var throwaway = [UInt8]() + try reused.compress(payload.span) { throwaway += bytes($0) } + try reused.finish { throwaway += bytes($0) } + reused.reset() + var reusedFrame = [UInt8]() + try reused.compress(payload.span) { reusedFrame += bytes($0) } + try reused.finish { reusedFrame += bytes($0) } + + var fresh = Zstd.StreamingCompressor(configuration: configuration) + var freshFrame = [UInt8]() + try fresh.compress(payload.span) { freshFrame += bytes($0) } + try fresh.finish { freshFrame += bytes($0) } + + #expect(reusedFrame == freshFrame, "reset must not drop strategy or level") + #expect(try Zstd.Decompressor().decompress(reusedFrame) == payload) + } + + @Test("After reset a second independent stream decodes") + func resetAllowsASecondStream() throws { + let frame = try Zstd.Compressor().compress(payload) + var decompressor = Zstd.StreamingDecompressor() + + var first = [UInt8]() + try decompressor.decompress(frame.span) { first += bytes($0) } + // #expect can't read a property off a ~Copyable value, so bind it first. + let finishedAfterFirst = decompressor.isFinished + #expect(first == payload) + #expect(finishedAfterFirst) + + decompressor.reset() + let finishedAfterReset = decompressor.isFinished + #expect(!finishedAfterReset) + + var second = [UInt8]() + try decompressor.decompress(frame.span) { second += bytes($0) } + let finishedAfterSecond = decompressor.isFinished + #expect(second == payload) + #expect(finishedAfterSecond) + } + + /// Without this, a reused decompressor keeps counting bytes from the stream + /// before it and trips `maxDecompressedSize` on a stream that is within the limit. + @Test("reset clears the byte count so maxDecompressedSize applies per stream") + func resetClearsTheByteCount() throws { + let frame = try Zstd.Compressor().compress(payload) + let configuration = Zstd.DecompressionConfiguration( + windowLogMax: 0, + maxDecompressedSize: payload.count + ) + var decompressor = Zstd.StreamingDecompressor(configuration: configuration) + + var first = [UInt8]() + try decompressor.decompress(frame.span) { first += bytes($0) } + #expect(first == payload) + + decompressor.reset() + + var second = [UInt8]() + try decompressor.decompress(frame.span) { second += bytes($0) } + #expect(second == payload) + } + + @Test("reset discards a partially fed stream") + func resetDiscardsPartialInput() throws { + let frame = try Zstd.Compressor().compress(payload) + var decompressor = Zstd.StreamingDecompressor() + + var partial = [UInt8]() + try decompressor.decompress(Array(frame.prefix(frame.count / 2)).span) { partial += bytes($0) } + let finishedAfterPartial = decompressor.isFinished + #expect(!finishedAfterPartial) + + decompressor.reset() + + var whole = [UInt8]() + try decompressor.decompress(frame.span) { whole += bytes($0) } + let finishedAtEnd = decompressor.isFinished + #expect(whole == payload) + #expect(finishedAtEnd) + } +}