[+] Missing LZ4 compressor
[*] Various compression related bugs under the 1/4th of the AuCompression platform related to compression stream objects. Now all 1/4ths match up.
This commit is contained in:
parent
8fb7b7e1ee
commit
ad4c18abe7
@ -73,6 +73,8 @@ namespace Aurora::Compression
|
||||
|
||||
AuUInt8 threads {1};
|
||||
|
||||
bool bLZ4AutoFlush {false};
|
||||
|
||||
inline CompressionInfo(ECompressionType alg) : type(alg)
|
||||
{
|
||||
|
||||
|
||||
@ -50,7 +50,7 @@ namespace Aurora::Compression
|
||||
AuUInt32 BaseStream::GetAvailableProcessedBytes()
|
||||
{
|
||||
AU_LOCK_GUARD(this->_spinlock);
|
||||
return this->_outbuffer.RemainingBytes();
|
||||
return this->_outbuffer.RemainingBytes(true);
|
||||
}
|
||||
|
||||
AuUInt32 BaseStream::Read(const Memory::MemoryViewWrite & /*opt*/ destination)
|
||||
|
||||
@ -48,7 +48,7 @@ namespace Aurora::Compression
|
||||
#endif
|
||||
#if defined(_AUHAS_LZ4)
|
||||
case ECompressionType::eLZ4:
|
||||
///ret = new LZ4Deflate(info);
|
||||
ret = new LZ4Deflate(info);
|
||||
break;
|
||||
#endif
|
||||
#if defined(_AUHAS_ZLIB)
|
||||
|
||||
@ -121,6 +121,27 @@ namespace Aurora::Compression
|
||||
{
|
||||
if (!this->ctx_.avail_in)
|
||||
{
|
||||
do
|
||||
{
|
||||
this->ctx_.avail_out = AuArraySize(this->dout_);
|
||||
this->ctx_.next_out = this->dout_;
|
||||
|
||||
auto ret = BZ2_bzCompress(&this->ctx_, type);
|
||||
if (ret < Z_OK)
|
||||
{
|
||||
SysPushErrorIO("Error: {}", BShitToString(ret));
|
||||
return false;
|
||||
}
|
||||
|
||||
auto have = AuArraySize(this->dout_) - this->ctx_.avail_out;
|
||||
|
||||
if (!Write(reinterpret_cast<const AuUInt8*>(this->dout_),
|
||||
have))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
} while (this->ctx_.avail_out == 0);
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
@ -111,6 +111,27 @@ namespace Aurora::Compression
|
||||
{
|
||||
if (!this->ctx_.avail_in)
|
||||
{
|
||||
do
|
||||
{
|
||||
this->ctx_.avail_out = AuArraySize(this->dout_);
|
||||
this->ctx_.next_out = this->dout_;
|
||||
|
||||
auto ret = deflate(&this->ctx_, type);
|
||||
if (ret < Z_OK)
|
||||
{
|
||||
SysPushErrorIO("Error: {}", zError(ret));
|
||||
return false;
|
||||
}
|
||||
|
||||
auto have = AuArraySize(this->dout_) - this->ctx_.avail_out;
|
||||
|
||||
if (!Write(reinterpret_cast<const AuUInt8*>(this->dout_),
|
||||
have))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
} while (this->ctx_.avail_out == 0);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
@ -7,3 +7,217 @@
|
||||
***/
|
||||
#pragma once
|
||||
|
||||
#include "lz4.h"
|
||||
#include "lz4frame.h"
|
||||
|
||||
namespace Aurora::Compression
|
||||
{
|
||||
struct LZ4Deflate : public BaseStream
|
||||
{
|
||||
CompressionInfo meta;
|
||||
AuSPtr<char> bufferIn_;
|
||||
AuSPtr<char> bufferOut_;
|
||||
char* readPtr_;
|
||||
AuUInt32 bufferInAvail{};
|
||||
|
||||
LZ4Deflate(const CompressionInfo &meta) : meta(meta), BaseStream(meta.internalStreamSize)
|
||||
{
|
||||
if (meta.bLZ4AutoFlush)
|
||||
{
|
||||
pref.autoFlush = true;
|
||||
}
|
||||
}
|
||||
|
||||
~LZ4Deflate()
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
bool Init(const AuSPtr<IO::IStreamReader> &reader)
|
||||
{
|
||||
this->reader_ = reader;
|
||||
|
||||
auto err = LZ4F_createCompressionContext(&this->cctxPtr, LZ4F_getVersion());
|
||||
if (LZ4F_isError(err))
|
||||
{
|
||||
return {};
|
||||
}
|
||||
|
||||
auto bufferSize = meta.internalStreamSize;
|
||||
|
||||
this->bufferIn_ = AuSPtr<char>(new char[bufferSize], AuDefaultDeleter<char[]>());
|
||||
|
||||
if (!this->bufferIn_)
|
||||
{
|
||||
return {};
|
||||
}
|
||||
|
||||
this->bufferOut_ = AuSPtr<char>(new char[bufferSize], AuDefaultDeleter<char[]>());
|
||||
if (!this->bufferOut_)
|
||||
{
|
||||
return {};
|
||||
}
|
||||
|
||||
this->readPtr_ = this->bufferIn_.get();
|
||||
SetPointer(this->readPtr_, bufferSize);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool AU_NOINLINE BeginLZ4FrameIfNeeded()
|
||||
{
|
||||
if (bDead)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if (AuExchange(this->bHasEncodedFrame, true))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
return DoLZ4Start();
|
||||
}
|
||||
|
||||
bool AU_NOINLINE EndLZ4FrameIfNeeded()
|
||||
{
|
||||
if (!this->bHasEncodedFrame)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
if (AuExchange(this->bHasFinished, true))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
return DoLZ4End();
|
||||
}
|
||||
|
||||
bool AU_NOINLINE DoLZ4Start()
|
||||
{
|
||||
this->bytesRemInFrame = 0;
|
||||
this->bHasFinished = false;
|
||||
|
||||
auto written = LZ4F_compressBegin(this->cctxPtr, this->bufferOut_.get(), meta.internalStreamSize, &this->pref);
|
||||
if (LZ4F_isError(written))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
bytesRemInFrame = written;
|
||||
|
||||
if (written)
|
||||
{
|
||||
if (!Write(this->bufferOut_.get(), written))
|
||||
{
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
bool AU_NOINLINE DoLZ4End()
|
||||
{
|
||||
if (!this->bufferOut_)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
auto startPtr = this->bufferOut_.get() + this->bytesRemInFrame;
|
||||
AuUInt32 bufferedBytes = LZ4F_compressEnd(cctxPtr, startPtr, meta.internalStreamSize - bytesRemInFrame, &options);
|
||||
if (LZ4F_isError(bufferedBytes))
|
||||
{
|
||||
this->bDead = true;
|
||||
return false;
|
||||
}
|
||||
|
||||
this->bytesRemInFrame += bufferedBytes;
|
||||
this->bHasEncodedFrame = false;
|
||||
|
||||
if (bufferedBytes)
|
||||
{
|
||||
if (!Write(startPtr, bufferedBytes))
|
||||
{
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
AuStreamReadWrittenPair_t Ingest_s(AuUInt32 input) override
|
||||
{
|
||||
bool ret = true;
|
||||
AuUInt32 inputStat = 0, outputStat = 0;
|
||||
|
||||
auto startLen = bytesRemInFrame;
|
||||
|
||||
if (!BeginLZ4FrameIfNeeded())
|
||||
{
|
||||
return {};
|
||||
}
|
||||
|
||||
auto bufferSize = meta.internalStreamSize;
|
||||
|
||||
while (inputStat < input)
|
||||
{
|
||||
inputStat += IngestForInPointer<char, uInt>(this->reader_, this->readPtr_, this->bufferInAvail, input - inputStat);
|
||||
|
||||
if (!this->bufferInAvail)
|
||||
{
|
||||
return { inputStat, (startLen - bytesRemInFrame) };
|
||||
}
|
||||
|
||||
size_t frameSPtr = this->bufferInAvail;
|
||||
size_t frameS2Ptr = bufferSize - bytesRemInFrame;
|
||||
|
||||
auto startPtr = this->bufferOut_.get() + bytesRemInFrame;
|
||||
auto temp = LZ4F_compressUpdate(this->cctxPtr, startPtr, frameS2Ptr, this->readPtr_, frameSPtr, &options);
|
||||
if (LZ4F_isError(temp))
|
||||
{
|
||||
SysPushErrorGeneric("LZ4 internal stream size was too small. ingested too much data. must reset stream now");
|
||||
bDead = true;
|
||||
return {};
|
||||
}
|
||||
|
||||
bytesRemInFrame += temp;
|
||||
|
||||
this->readPtr_ += this->bufferInAvail;
|
||||
this->bufferInAvail = 0;
|
||||
|
||||
if (temp)
|
||||
{
|
||||
if (!Write(startPtr, temp))
|
||||
{
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
outputStat += temp;
|
||||
}
|
||||
|
||||
return AuMakePair(inputStat, (startLen - bytesRemInFrame));
|
||||
}
|
||||
|
||||
bool Flush() override
|
||||
{
|
||||
return EndLZ4FrameIfNeeded();
|
||||
}
|
||||
|
||||
bool Finish() override
|
||||
{
|
||||
return Flush();
|
||||
}
|
||||
|
||||
private:
|
||||
|
||||
bool bHasEncodedFrame{};
|
||||
bool bHasFinished{};
|
||||
bool bDead{};
|
||||
size_t bytesRemInFrame{};
|
||||
AuSPtr<IO::IStreamReader> reader_;
|
||||
LZ4F_cctx* cctxPtr;
|
||||
LZ4F_preferences_t pref{};
|
||||
LZ4F_compressOptions_t options{};
|
||||
};
|
||||
}
|
||||
@ -100,6 +100,7 @@ namespace Aurora::Compression
|
||||
}
|
||||
|
||||
this->readPtr_ += frameSPtr;
|
||||
outputStat += bytesRemInFrame;
|
||||
this->bufferInAvail -= frameSPtr;
|
||||
|
||||
if (frameS2Ptr)
|
||||
@ -110,10 +111,10 @@ namespace Aurora::Compression
|
||||
}
|
||||
}
|
||||
|
||||
outputStat += frameS2Ptr;
|
||||
inputStat += frameS2Ptr;
|
||||
}
|
||||
|
||||
return {inputStat, outputStat};
|
||||
return AuMakePair(inputStat, outputStat);
|
||||
}
|
||||
|
||||
private:
|
||||
|
||||
@ -100,9 +100,14 @@ namespace Aurora::Compression
|
||||
return AuMakePair(read, 0);
|
||||
}
|
||||
|
||||
if (!output.pos)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
done += AuUInt32(output.pos);
|
||||
|
||||
if (!Write(reinterpret_cast<const AuUInt8 *>(output.dst),
|
||||
if (!Write(reinterpret_cast<const AuUInt8 *>(this->dout_),
|
||||
AuUInt32(output.pos)))
|
||||
{
|
||||
return AuMakePair(read, 0);
|
||||
@ -132,7 +137,24 @@ namespace Aurora::Compression
|
||||
|
||||
if (!this->availIn_)
|
||||
{
|
||||
return {};
|
||||
ZSTD_outBuffer output = { this->dout_, outFrameLength, 0 };
|
||||
|
||||
ZSTD_inBuffer input = { NULL, 0, 0 };
|
||||
auto ret = ZSTD_compressStream2(this->cctx_, &output, &input, type);
|
||||
if (ZSTD_isError(ret))
|
||||
{
|
||||
SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(ret));
|
||||
this->availIn_ -= AuUInt32(output.pos);
|
||||
return {};
|
||||
}
|
||||
|
||||
if (!Write(reinterpret_cast<const AuUInt8*>(output.dst),
|
||||
AuUInt32(output.pos)))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
this->input_ = ZSTD_inBuffer {this->curPtr_, this->availIn_, 0};
|
||||
@ -149,7 +171,7 @@ namespace Aurora::Compression
|
||||
return {};
|
||||
}
|
||||
|
||||
if (!Write(reinterpret_cast<const AuUInt8 *>(output.dst),
|
||||
if (!Write(reinterpret_cast<const AuUInt8 *>(this->dout_),
|
||||
AuUInt32(output.pos)))
|
||||
{
|
||||
return false;
|
||||
|
||||
@ -620,6 +620,7 @@ namespace Aurora::Compression
|
||||
|
||||
ret = true;
|
||||
pref.compressionLevel = info.compressionLevel;
|
||||
pref.autoFlush = true;
|
||||
|
||||
auto maxFrameSize = info.lz4BlockSize ? info.lz4BlockSize : 64 * 1024;
|
||||
auto buffer = AuSPtr<char>(new char[maxFrameSize], AuDefaultDeleter<char[]>());
|
||||
@ -655,6 +656,7 @@ namespace Aurora::Compression
|
||||
|
||||
if (!flush)
|
||||
{
|
||||
|
||||
AuUInt32 bufferedBytes = LZ4F_compressUpdate(cctxPtr, outBuffer.get(), maxOut, buffer.get(), read, &options);
|
||||
|
||||
if (LZ4F_isError(bufferedBytes))
|
||||
|
||||
@ -454,7 +454,7 @@ namespace Aurora::IO
|
||||
// a load of small deserializable packets at the start of a large buffer, for the CPU to immediately start failing OnDataAvailable's
|
||||
// much later into the stream, where a larger packet may overhang into memory we haven't reserved
|
||||
|
||||
// I really don't know how ReadNextAsync can be excepted to wrap around a ring buffer
|
||||
// I really don't know how ReadNextAsync can be expected to wrap around a ring buffer
|
||||
// We'd need to know if this pass failed, and if the read head is near the end, it'd know
|
||||
// to wrap back around to zero. An overengineered pain and liability.
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user