github-starred/ultimatepp
2
0
Fork 0
mirror of https://github.com/ultimatepp/ultimatepp.git synced 2026-07-11 22:03:01 -06:00
Code Issues 33 Projects Packages Wiki Activity Actions

.developing LZ4

Browse source 
git-svn-id: svn://ultimatepp.org/upp/trunk@10097 f0d560ea-af0d-0410-9eb7-867de7ffcac7
This commit is contained in:
cxl 2016-07-25 11:20:00 +00:00
parent de28182b3f
commit 82a6cc93dc
7 changed files with 78 additions and 657 deletions
Download patch file Download diff file Expand all files Collapse all files

1
uppsrc/Core/xxHsh.cpp
View file

@ -17,7 +17,6 @@ void xxHashStream::Reset(dword seed)
void xxHashStream::Out(const void *data, dword size)
{
RTIMING("xxh2");
XXH32_update((XXH32_state_t *)context, data, size);
}

159
uppsrc/plugin/lz4/Compress.cpp
View file

@ -1,119 +1,112 @@
#include "lz4.h"
#ifdef flagCOMPLEXLZ4
#ifndef flagCOMPLEXLZ4
#define LLOG(x) // LOG(x)
namespace Upp {
void LZ4CompressStream::Init()
void LZ4CompressStream::Open(Stream& out_)
{
out = &out_;
ClearError();
pos = 0;
header = false;
xxh.Reset();
#ifdef _MULTITHREADED
outblock = inblock = 0;
#endif
SetupBuffer();
Alloc();
pos = 0;
byte h[7];
Poke32le(h, LZ4F_MAGIC);
h[4] = LZ4F_VERSION | LZ4F_BLOCKINDEPENDENCE | LZ4F_CONTENTCHECKSUM;
h[5] = LZ4F_MAXSIZE_1024KB;
h[6] = xxHash(h + 4, 2) >> 8;
out->Put(h, 7);
}
void LZ4CompressStream::FinishBlock(char *outbuf, int clen, const char *origdata, int origsize)
void LZ4CompressStream::Alloc()
{
if(IsError() || out && out->IsError())
return;
if(clen < 0) {
SetError();
return;
}
if(clen >= origsize) {
Poke32le(outbuf, 0x80000000 | origsize);
memcpy(outbuf + 4, origdata, origsize);
out->Put(outbuf, origsize + 4);
}
else {
Poke32le(outbuf, clen);
out->Put(outbuf, clen + 4);
}
int N = 16;
int sz = concurrent ? N * BLOCK_BYTES : BLOCK_BYTES;
buffer.Alloc(sz);
outbuf.Alloc(N * LZ4_compressBound(BLOCK_BYTES));
outsz.Alloc(N);
wrlim = ~buffer + sz;
ptr = ~buffer;
}
void LZ4CompressStream::SetupBuffer()
void LZ4CompressStream::Concurrent(bool b)
{
buffer.SetCount(BLOCK_BYTES);
wrlim = (byte *)~buffer + BLOCK_BYTES;
ptr = (byte *)~buffer;
FlushOut();
concurrent = b;
Alloc();
}
void LZ4CompressStream::FlushOut()
{
if(!header) {
byte h[7];
Poke32le(h, LZ4F_MAGIC);
h[4] = LZ4F_VERSION | LZ4F_BLOCKINDEPENDENCE | LZ4F_CONTENTCHECKSUM;
h[5] = LZ4F_MAXSIZE_1024KB;
h[6] = xxHash(h + 4, 2) >> 8;
out->Put(h, 7);
header = true;
}
if(ptr == (byte *)~buffer)
return;
int origsize = int((char *)ptr - ~buffer);
CoWork co;
int osz = LZ4_compressBound(BLOCK_BYTES);
byte *t = ~outbuf;
int ii = 0;
for(byte *s = ~buffer; s < ptr; s += BLOCK_BYTES) {
int origsize = min((int)BLOCK_BYTES, int(ptr - s));
if(concurrent)
co & [=] {
outsz[ii] = LZ4_compress_default((char *)s, (char *)t, origsize, osz);
};
else
outsz[ii] = LZ4_compress_default((char *)s, (char *)t, origsize, osz);
ii++;
t += osz;
}
if(concurrent)
co.Finish();
byte *s = ~buffer;
t = ~outbuf;
for(int i = 0; i < ii; i++) {
int origsize = min((int)BLOCK_BYTES, int(ptr - s));
int clen = outsz[i];
if(clen < 0) {
SetError();
return;
}
if(clen >= origsize || clen == 0) {
out->Put32le(0x80000000 | origsize);
out->Put(s, origsize);
}
else {
out->Put32le(clen);
out->Put(t, clen);
}
s += BLOCK_BYTES;
t += osz;
}
int origsize = int(ptr - ~buffer);
xxh.Put(~buffer, origsize);
pos += origsize;
WhenPos(pos);
#ifdef _MULTITHREADED
if(co) {
String bs = buffer;
int inblk = inblock++;
CoWork::Start([=] {
Buffer<char> outbuf(4 + LZ4_compressBound(BLOCK_BYTES));
int clen = LZ4_compress(~bs, ~outbuf + 4, origsize);
Mutex::Lock __(lock);
while(outblock != inblk)
cond.Wait(lock);
FinishBlock(outbuf, clen, ~bs, origsize);
outblock++;
cond.Broadcast();
});
xxh.Put(~bs, origsize);
SetupBuffer();
}
else
#endif
{
Buffer<char> outbuf(4 + LZ4_compressBound(BLOCK_BYTES));
xxh.Put(~buffer, origsize);
int clen = LZ4_compress(~buffer, ~outbuf + 4, origsize);
FinishBlock(outbuf, clen, buffer, origsize);
SetupBuffer();
}
ptr = ~buffer;
}
void LZ4CompressStream::Close()
{
ASSERT(compress >= 0);
FlushOut();
#ifdef _MULTITHREADED
if(co) {
Mutex::Lock __(lock);
while(outblock != inblock)
cond.Wait(lock);
if(out) {
FlushOut();
out->Put32le(0);
out->Put32le(xxh.Finish());
out = NULL;
}
#endif
byte h[8];
Poke32le(h, 0);
Poke32le(h + 4, xxh.Finish());
out->Put(h, 8);
buffer.Clear();
}
bool LZ4CompressStream::IsOpen() const
{
return (!out || out->IsOpen()) && buffer.GetCount();
return out && out->IsOpen();
}
void LZ4CompressStream::_Put(int w)
@ -133,7 +126,7 @@ void LZ4CompressStream::_Put(const void *data, dword size)
while(size > 0) {
if(IsError() || out && out->IsError())
return;
dword n = BLOCK_BYTES - int((char *)ptr - ~buffer);
dword n = dword(wrlim - ptr);
if(size >= n) {
memcpy(ptr, s, n);
ptr = wrlim;
@ -152,14 +145,14 @@ void LZ4CompressStream::_Put(const void *data, dword size)
LZ4CompressStream::LZ4CompressStream()
{
#ifdef _MULTITHREADED
co = false;
concurrent = false;
#endif
out = NULL;
Init();
}
LZ4CompressStream::~LZ4CompressStream()
{
Close();
}
};

160
uppsrc/plugin/lz4/Compress2.cpp
View file

@ -1,160 +0,0 @@
#include "lz4.h"
#ifndef flagCOMPLEXLZ4
#define LLOG(x) // LOG(x)
namespace Upp {
void LZ4CompressStream::Open(Stream& out_)
{
out = &out_;
ClearError();
pos = 0;
xxh.Reset();
Alloc();
pos = 0;
byte h[7];
Poke32le(h, LZ4F_MAGIC);
h[4] = LZ4F_VERSION | LZ4F_BLOCKINDEPENDENCE | LZ4F_CONTENTCHECKSUM;
h[5] = LZ4F_MAXSIZE_1024KB;
h[6] = xxHash(h + 4, 2) >> 8;
out->Put(h, 7);
}
void LZ4CompressStream::Alloc()
{
int N = 16;
int sz = concurrent ? N * BLOCK_BYTES : BLOCK_BYTES;
buffer.Alloc(sz);
outbuf.Alloc(N * LZ4_compressBound(BLOCK_BYTES));
outsz.Alloc(N);
wrlim = ~buffer + sz;
ptr = ~buffer;
}
void LZ4CompressStream::Concurrent(bool b)
{
FlushOut();
concurrent = b;
Alloc();
}
void LZ4CompressStream::FlushOut()
{
if(ptr == (byte *)~buffer)
return;
CoWork co;
int osz = LZ4_compressBound(BLOCK_BYTES);
byte *t = ~outbuf;
int ii = 0;
for(byte *s = ~buffer; s < ptr; s += BLOCK_BYTES) {
int origsize = min((int)BLOCK_BYTES, int(ptr - s));
if(concurrent)
co & [=] {
outsz[ii] = LZ4_compress_default((char *)s, (char *)t, origsize, osz);
};
else
outsz[ii] = LZ4_compress_default((char *)s, (char *)t, origsize, osz);
ii++;
t += osz;
}
if(concurrent)
co.Finish();
byte *s = ~buffer;
t = ~outbuf;
for(int i = 0; i < ii; i++) {
int origsize = min((int)BLOCK_BYTES, int(ptr - s));
int clen = outsz[i];
if(clen < 0) {
SetError();
return;
}
if(clen >= origsize || clen == 0) {
out->Put32le(0x80000000 | origsize);
out->Put(s, origsize);
}
else {
out->Put32le(clen);
out->Put(t, clen);
}
s += BLOCK_BYTES;
t += osz;
}
int origsize = int(ptr - ~buffer);
xxh.Put(~buffer, origsize);
pos += origsize;
WhenPos(pos);
ptr = ~buffer;
}
void LZ4CompressStream::Close()
{
ASSERT(compress >= 0);
if(out) {
FlushOut();
out->Put32le(0);
out->Put32le(xxh.Finish());
out = NULL;
}
}
bool LZ4CompressStream::IsOpen() const
{
return out && out->IsOpen();
}
void LZ4CompressStream::_Put(int w)
{
FlushOut();
*ptr++ = w;
}
void LZ4CompressStream::_Put(const void *data, dword size)
{
LLOG("Put " << size);
ASSERT(compress >= 0);
const char *s = reinterpret_cast<const char *>(data);
while(size > 0) {
if(IsError() || out && out->IsError())
return;
dword n = dword(wrlim - ptr);
if(size >= n) {
memcpy(ptr, s, n);
ptr = wrlim;
FlushOut();
size -= n;
s += n;
}
else {
memcpy(ptr, s, size);
ptr += size;
break;
}
}
}
LZ4CompressStream::LZ4CompressStream()
{
#ifdef _MULTITHREADED
concurrent = false;
#endif
out = NULL;
}
LZ4CompressStream::~LZ4CompressStream()
{
Close();
}
};
#endif

14
uppsrc/plugin/lz4/Decompress.cpp
View file

@ -65,7 +65,6 @@ bool LZ4DecompressStream::Open(Stream& in_)
bool LZ4DecompressStream::Next()
{
RTIMING("Next");
if(ii < count) {
pos += dlen;
ptr = (byte *)~wb[ii].d;
@ -105,7 +104,6 @@ void LZ4DecompressStream::Fetch()
return;
}
if(!t.c) {
RTIMING("Alloc");
t.c.Alloc(maxblock);
t.d.Alloc(maxblock);
}
@ -117,12 +115,10 @@ void LZ4DecompressStream::Fetch()
}
}
else {
{ RTIMING("GetAll");
if(!in->GetAll(~t.c, t.clen)) {
SetError();
return;
}
}
#ifdef _MULTITHREADED
if(concurrent)
co & [=, &error] {
@ -135,7 +131,6 @@ void LZ4DecompressStream::Fetch()
else
#endif
{
RTIMING("LZ4 decompress");
t.dlen = LZ4_decompress_safe(~t.c, ~t.d, t.clen, maxblock);
if(t.dlen < 0)
error = true;
@ -151,10 +146,8 @@ void LZ4DecompressStream::Fetch()
if(error)
SetError();
else {
for(int i = 0; i < count; i++) {
RTIMING("xxh");
for(int i = 0; i < count; i++)
xxh.Put(wb[i].d, wb[i].dlen);
}
if(last) {
if(in->Get32le() != xxh.Finish())
SetError();
@ -187,22 +180,19 @@ int LZ4DecompressStream::_Get()
dword LZ4DecompressStream::_Get(void *data, dword size)
{
RTIMING("_Get");
byte *t = (byte *)data;
while(size) {
if(IsError() || in->IsError() || ptr == rdlim && ii == count && eof)
break;
dword n = dword(rdlim - ptr);
if(size < n) {
RTIMING("memcpy1");
memcpy(t, ptr, size);
t += size;
ptr += size;
break;
}
else {
{ RTIMING("memcpy2");
memcpy(t, ptr, n); }
memcpy(t, ptr, n);
t += n;
size -= n;
ptr = rdlim;

109
uppsrc/plugin/lz4/lz4.h
View file

@ -24,114 +24,6 @@ enum {
LZ4F_MAXSIZE_4096KB = 0x70,
};
class Lz4 { // Filter is deprecated, use Streams instead, will be removed soon
StringBuffer buffer; // to be able to pass as String
int8 compress;
bool error;
enum { BLOCK_BYTES = 1024 * 1024 };
xxHashStream xxh;
int maxblock;
int pos;
int blockchksumsz;
bool header;
byte lz4hdr;
String header_data;
String out;
#ifdef _MULTITHREADED
bool parallel;
CoWork co;
Mutex lock;
ConditionVariable cond;
int outblock;
int inblock;
#endif
void TryHeader();
void Init();
void FinishBlock(char *outbuf, int clen, const char *origdata, int origsize);
void FlushOut();
void PutOut(const void *ptr, int size);
public:
Callback2<const void *, int> WhenOut;
void Put(const void *ptr, int size);
void Put(const String& s) { Put(~s, s.GetCount()); }
void End();
void Clear();
const String& Get() const { return out; }
operator const String&() const { return out; }
const String& operator~() const { return out; }
void ClearOut() { out.Clear(); }
void Compress();
void Decompress();
#ifdef _MUTLITHREADED
void Parallel(bool b = true) { parallel = b; }
#endif
bool IsError() const { return error; }
Lz4();
~Lz4();
};
#ifdef flagCOMPLEXLZ4
class LZ4CompressStream : public Stream {
public:
virtual void Close();
virtual bool IsOpen() const;
protected:
virtual void _Put(int w);
virtual void _Put(const void *data, dword size);
Stream *out;
StringBuffer buffer; // to be able to pass it as String
enum { BLOCK_BYTES = 1024 * 1024 };
xxHashStream xxh;
int blockchksumsz;
bool header;
byte lz4hdr;
String header_data;
#ifdef _MULTITHREADED
bool co;
Mutex lock;
ConditionVariable cond;
int outblock;
int inblock;
#endif
void Init();
void SetupBuffer();
void FinishBlock(char *outbuf, int clen, const char *origdata, int origsize);
void FlushOut();
public:
Event<int64> WhenPos;
#ifdef _MULTITHREADED
void Concurrent(bool b = true) { co = b; }
#endif
void Open(Stream& out_) { Init(); out = &out_; }
LZ4CompressStream();
LZ4CompressStream(Stream& out) : LZ4CompressStream() { Open(out); }
~LZ4CompressStream();
};
#else
class LZ4CompressStream : public Stream {
public:
virtual void Close();
@ -171,7 +63,6 @@ public:
LZ4CompressStream(Stream& out) : LZ4CompressStream() { Open(out); }
~LZ4CompressStream();
};
#endif
class LZ4DecompressStream : public Stream {
public:

2
uppsrc/plugin/lz4/lz4.upp
View file

@ -5,9 +5,7 @@ uses
file
lz4.h,
lz4upp.cpp,
Compress.cpp,
Compress2.cpp,
Decompress.cpp,
util.cpp,
lib\LICENSE,

290
uppsrc/plugin/lz4/lz4upp.cpp
View file

@ -1,290 +0,0 @@
#include "lz4.h"
#define LLOG(x) // LOG(x)
namespace Upp {
void Lz4::Init()
{
error = false;
pos = 0;
header = false;
xxh.Reset();
#ifdef _MULTITHREADED
outblock = inblock = 0;
#endif
}
void Lz4::Compress()
{
compress = 1;
buffer.SetCount(BLOCK_BYTES);
Init();
}
void Lz4::Decompress()
{
compress = 0;
header_data.Clear();
Init();
}
void Lz4::PutOut(const void *ptr, int size)
{
LLOG("LZ4 PutOut " << out.GetCount());
out.Cat((const char *)ptr, (int)size);
}
void Lz4::FinishBlock(char *outbuf, int clen, const char *origdata, int origsize)
{
RTIMING("FinishBlock");
if(error)
return;
if(clen < 0) {
error = true;
return;
}
if(clen >= origsize) {
Poke32le(outbuf, 0x80000000 | origsize);
memcpy(outbuf + 4, origdata, origsize);
WhenOut(outbuf, origsize + 4);
}
else {
Poke32le(outbuf, clen);
WhenOut(outbuf, clen + 4);
}
}
void Lz4::FlushOut()
{
if(!header) {
byte h[7];
Poke32le(h, LZ4F_MAGIC);
h[4] = LZ4F_VERSION | LZ4F_BLOCKINDEPENDENCE | LZ4F_CONTENTCHECKSUM;
h[5] = LZ4F_MAXSIZE_1024KB;
h[6] = xxHash(h + 4, 2) >> 8;
WhenOut(h, 7);
header = true;
maxblock = BLOCK_BYTES;
}
if(!pos)
return;
int origsize = pos;
pos = 0;
#ifdef _MULTITHREADED
if(parallel) {
String bs = buffer;
int inblk = inblock++;
// DLOG("Scheduling " << inblk);
co.Start([=] {
Buffer<char> outbuf(4 + LZ4_compressBound(maxblock));
int clen = LZ4_compress(~bs, ~outbuf + 4, origsize);
Mutex::Lock __(lock);
{ RTIMING("Waiting for order");
while(outblock != inblk)
cond.Wait(lock);
}
FinishBlock(outbuf, clen, ~bs, origsize);
outblock++;
cond.Broadcast();
});
RTIMING("xxh");
xxh.Put(~bs, origsize);
buffer.SetCount(BLOCK_BYTES);
}
else
#endif
{
Buffer<char> outbuf(4 + LZ4_compressBound(maxblock));
xxh.Put(~buffer, origsize);
int clen = LZ4_compress(~buffer, ~outbuf + 4, origsize);
FinishBlock(outbuf, clen, buffer, origsize);
}
}
void Lz4::End()
{
ASSERT(compress >= 0);
if(compress) {
FlushOut();
#ifdef _MULTITHREADED
Mutex::Lock __(lock);
{ RTIMING("Waiting for order");
while(outblock != inblock)
cond.Wait(lock);
}
#endif
byte h[8];
Poke32le(h, 0);
Poke32le(h + 4, xxh.Finish());
WhenOut(h, 8);
}
else
if(pos)
error = true;
}
void Lz4::TryHeader()
{
if(header_data.GetCount() < 7)
return;
if(Peek32le(~header_data) != LZ4F_MAGIC) {
error = true;
return;
}
lz4hdr = header_data[4];
if((lz4hdr & LZ4F_VERSIONMASK) != LZ4F_VERSION) {
error = true;
return;
}
if(!(lz4hdr & LZ4F_BLOCKINDEPENDENCE)) {
error = true; // dependent blocks not supported
return;
}
maxblock = header_data[5];
maxblock = decode(maxblock & LZ4F_MAXSIZEMASK,
LZ4F_MAXSIZE_64KB, 1024 * 64,
LZ4F_MAXSIZE_256KB, 1024 * 256,
LZ4F_MAXSIZE_1024KB, 1024 * 1024,
LZ4F_MAXSIZE_4096KB, 1024 * 4096,
-1);
if(maxblock < 0) {
error = true;
return;
}
if((lz4hdr & LZ4F_CONTENTSIZE) && header_data.GetCount() < 15)
return; // need to skip content length
header = true;
blockchksumsz = 4 * !!(lz4hdr & LZ4F_BLOCKCHECKSUM);
buffer.SetCount(maxblock + 4 + blockchksumsz);
}
void Lz4::Put(const void *ptr_, int size)
{
LLOG("Put " << size);
ASSERT(compress >= 0);
if(error)
return;
const char *ptr = reinterpret_cast<const char *>(ptr_);
if(compress) {
while(size > 0) {
if(error)
return;
int n = BLOCK_BYTES - pos;
if(size >= n) {
memcpy(~buffer + pos, ptr, n);
pos = BLOCK_BYTES;
FlushOut();
size -= n;
ptr += n;
}
else {
memcpy(buffer + pos, ptr, size);
pos += size;
break;
}
}
}
else {
while(!header && size > 0 && !error) {
header_data.Cat(*ptr++);
size--;
TryHeader();
}
while(size > 0 && !error) { // TODO might try to avoid memcpy
if(pos < 4) { // Get length first
int n = min(size, 4 - pos);
memcpy(~buffer + pos, ptr, n);
pos += n;
ptr += n;
size -= n;
}
else {
int count = Peek32le(~buffer);
int len = count & 0x7fffffff;
if(len > maxblock) {
error = true;
break;
}
if(count == 0) { // reached the end
int need = 4 - pos + 4;
if(size >= need) { // we have enough data for final checksum
memcpy(~buffer + pos, ptr, need);
if(Peek32le(~buffer + 4) != xxh.Finish()) {
error = true;
return;
}
size -= need;
ptr += need;
pos = 0;
return; // decompression completed
}
}
else {
int need = len - (pos - 4 - blockchksumsz);
if(size >= need) { // we have enough data to finish the block
memcpy(~buffer + pos, ptr, need);
const char *src = ~buffer + 4;
if(count & 0x80000000) {
WhenOut(src, len);
xxh.Put(src, len);
}
else {
int sz;
Buffer<char> outbuf(maxblock);
sz = LZ4_decompress_safe(src, outbuf, len, maxblock);
if(sz < 0) {
error = true;
return;
}
WhenOut(outbuf, sz);
xxh.Put(outbuf, sz);
}
ptr += need;
size -= need;
pos = 0;
continue;
}
}
memcpy(~buffer + pos, ptr, size);
pos += size;
break;
}
}
}
}
Lz4::Lz4()
{
compress = -1;
#ifdef _MULTITHREADED
parallel = false;
#endif
WhenOut = callback(this, &Lz4::PutOut);
}
Lz4::~Lz4()
{
}
bool IsLZ4(Stream& s)
{
int64 pos = s.GetPos();
bool b = s.Get32le() == LZ4F_MAGIC;
s.Seek(pos);
return b;
}
};