AuroraRuntime/Source/Compression/Compressors/ZSTDCompressor.hpp
Reece Wilson 745b9f974a [+] ByteBuffer::WriteFromEx
[+] ByteBuffer::WriteFrom
[*] Fix known path resolution quirks and missing compression staged change
[+] ICompressionInterceptor::LimitHasHit
[+] ICompressionInterceptor::LimitReset
[+] ICompressionInterceptor::LimitSet
[+] ICompressionInterceptor::LimitGetIndex
2022-11-07 22:46:35 +00:00

221 lines
6.7 KiB
C++

/***
Copyright (C) 2022 J Reece Wilson (a/k/a "Reece"). All rights reserved.
File: ZSTDCompressor.hpp
Date: 2022-2-15
Author: Reece
***/
#pragma once
#include "zstd.h"
namespace Aurora::Compression
{
struct ZSTDDeflate : BaseStream
{
CompressInfo meta;
ZSTDDeflate(const CompressInfo &meta) : meta(meta), BaseStream(meta.uInternalStreamSize)
{}
~ZSTDDeflate()
{
if (auto cctx = AuExchange(this->cctx_, {}))
{
ZSTD_freeCCtx(cctx);
}
}
bool Init(const AuSPtr<IO::IStreamReader> &pReader) override
{
AuUInt uRet;
if (!this->IsValid())
{
SysPushErrorMem();
return false;
}
this->pReader_ = pReader;
this->cctx_ = ZSTD_createCCtx();
if (!this->cctx_)
{
SysPushErrorGen("Couldn't create compressor");
}
uRet = ZSTD_CCtx_setParameter(this->cctx_, ZSTD_c_compressionLevel, meta.uCompressionLevel);
if (ZSTD_isError(uRet))
{
SysPushErrorGen("Invalid compression level");
return false;
}
uRet = ZSTD_CCtx_setParameter(this->cctx_, ZSTD_c_checksumFlag, 1);
if (ZSTD_isError(uRet))
{
SysPushErrorGen("Invalid option");
return false;
}
uRet = ZSTD_CCtx_setParameter(this->cctx_, ZSTD_c_nbWorkers, AuMax(meta.uThreads, AuUInt8(1u)));
if (ZSTD_isError(uRet))
{
SysPushErrorGen();
return false;
}
this->pIterator_ = this->din_;
this->uAvailableIn_ = 0;
SetArray(this->din_);
return true;
}
AuStreamReadWrittenPair_t Ingest_s(AuUInt32 input) override
{
AuUInt32 uLength = AuUInt32(ZSTD_DStreamInSize());
AuUInt32 uOutFrameLength = AuUInt32(ZSTD_DStreamOutSize());
AuUInt32 done {}, read {};
if (!this->pReader_)
{
return {};
}
while (read < input)
{
read += IngestForInPointer<char, AuUInt32>(this->pReader_, this->pIterator_, this->uAvailableIn_, input - read);
if (!this->uAvailableIn_)
{
Flush();
return {read, done};
}
this->input_ = ZSTD_inBuffer {this->pIterator_, this->uAvailableIn_, 0};
size_t uRet {};
bool bLastFrame = input - read == 0;
while ((this->input_.pos != this->input_.size) ||
(/*bLastFrame && uRet*/ false))
{
ZSTD_outBuffer output = { this->dout_, uOutFrameLength, 0 };
ZSTD_EndDirective mode = ZSTD_e_continue;// bLastFrame ? ZSTD_e_flush : ZSTD_e_continue;
uRet = ZSTD_compressStream2(this->cctx_, &output, &this->input_, mode);
if (ZSTD_isError(uRet))
{
SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(uRet));
this->uAvailableIn_ -= AuUInt32(output.pos);
this->pReader_.reset();
return AuMakePair(read, 0);
}
this->bLastFrameHasNotFinished_ = bLastFrame && uRet;
if (!output.pos)
{
continue;
}
done += AuUInt32(output.pos);
if (!Write(reinterpret_cast<const AuUInt8 *>(this->dout_),
AuUInt32(output.pos)))
{
this->pReader_.reset();
return AuMakePair(read, 0);
}
}
this->pIterator_ += this->input_.pos;
this->uAvailableIn_ -= AuUInt32(this->input_.pos);
}
return {read, done};
}
bool Flush() override
{
AuUInt32 uLength = AuUInt32(ZSTD_DStreamInSize());
AuUInt32 uOutFrameLength = AuUInt32(ZSTD_DStreamOutSize());
this->input_ = ZSTD_inBuffer { this->pIterator_, this->uAvailableIn_, 0 };
AuUInt uRet {};
while ((this->input_.pos < this->input_.size) ||
(this->bLastFrameHasNotFinished_) ||
(uRet))
{
ZSTD_outBuffer output = { this->dout_, uOutFrameLength, 0 };
this->bLastFrameHasNotFinished_ = false;
uRet = ZSTD_compressStream2(this->cctx_, &output, &this->input_, ZSTD_e_flush);
if (ZSTD_isError(uRet))
{
SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(uRet));
this->uAvailableIn_ -= AuUInt32(output.pos);
return {};
}
if (!output.pos)
{
continue;
}
if (!Write(reinterpret_cast<const AuUInt8 *>(this->dout_),
AuUInt32(output.pos)))
{
this->pReader_.reset();
return false;
}
}
this->pIterator_ += this->input_.pos;
this->uAvailableIn_ -= this->input_.pos;
return true;// RunFlush(ZSTD_e_continue);
}
bool Finish() override
{
AuUInt32 uOutFrameLength = AuUInt32(ZSTD_DStreamOutSize());
ZSTD_outBuffer output = { this->dout_, uOutFrameLength, 0 };
Flush();
size_t uRet;
do
{
uRet = ZSTD_endStream(this->cctx_, &output);
if (ZSTD_isError(uRet))
{
SysPushErrorIO("Compression error: {}", ZSTD_getErrorName(uRet));
return {};
}
if (!Write(reinterpret_cast<const AuUInt8 *>(this->dout_),
AuUInt32(output.pos)))
{
return false;
}
}
while (uRet);
return true;
}
private:
AuSPtr<IO::IStreamReader> pReader_;
ZSTD_CCtx *cctx_ {};
char din_[ZSTD_BLOCKSIZE_MAX];
char dout_[ZSTD_COMPRESSBOUND(ZSTD_BLOCKSIZE_MAX) + 3 +/*ZSTD_BLOCKHEADERSIZE*/ + 4 /*32bit hash*/];
char *pIterator_ {};
AuUInt32 uAvailableIn_ {};
ZSTD_inBuffer input_ {};
bool bLastFrameHasNotFinished_ {};
};
}