diff --git a/uppsrc/Core/CharSet.cpp b/uppsrc/Core/CharSet.cpp index ff4a0e94d..bfc80e625 100644 --- a/uppsrc/Core/CharSet.cpp +++ b/uppsrc/Core/CharSet.cpp @@ -2209,7 +2209,7 @@ bool utf8check(const char *_s, int len) if(s + 1 >= lim || s[0] < 0x80 || s[0] >= 0xc0 || s[1] < 0x80 || s[1] >= 0xc0) - return false; + return false; codePoint = ((code - 0xE0) << 12) + ((s[0] - 0x80) << 6) + s[1] - 0x80; if(codePoint < 0x0800 || codePoint > 0xFFFF) return false; @@ -2221,7 +2221,7 @@ bool utf8check(const char *_s, int len) s[0] < 0x80 || s[0] >= 0xc0 || s[1] < 0x80 || s[1] >= 0xc0 || s[2] < 0x80 || s[2] >= 0xc0) - return false; + return false; codePoint = ((code - 0xf0) << 18) + ((s[0] - 0x80) << 12) + ((s[1] - 0x80) << 6) + s[2] - 0x80; if(codePoint < 0x010000 || codePoint > 0x10FFFF) diff --git a/uppsrc/Core/CoWork.cpp b/uppsrc/Core/CoWork.cpp index 1958e8c10..302734301 100644 --- a/uppsrc/Core/CoWork.cpp +++ b/uppsrc/Core/CoWork.cpp @@ -79,7 +79,7 @@ bool CoWork::Pool::DoJob() LLOG("Quit thread"); return true; } - LLOG("DoJob " << p.scheduled - 1 << ", todo: " << job.work->todo << " (CoWork " << FormatIntHex(job.work) << ")"); + LLOG("DoJob " << p.scheduled - 1 << " (CoWork " << FormatIntHex(job.work) << ")"); finlock = false; Function fn = pick(job.fn); CoWork *work = job.work; @@ -147,7 +147,9 @@ CoWork::MJob& CoWork::PushJob(Function&& fn) Pool& p = GetPool(); MJob& job = p.jobs[p.scheduled++]; job.fn = pick(fn); - LLOG("Adding job " << p.scheduled - 1 << "; todo: " << todo << " (CoWork " << FormatIntHex(this) << ")"); + job.work = NULL; + job.started = NULL; + LLOG("Adding job " << p.scheduled - 1); if(p.waiting_threads) { LLOG("Releasing thread waiting for job: " << p.waiting_threads); p.waiting_threads--; @@ -237,11 +239,14 @@ CoWork::CoWork() { LLOG("CoWork constructed " << FormatHex(this)); todo = 0; +// SetMagic(magic, sizeof(magic)); } CoWork::~CoWork() { +// CheckMagic(magic, sizeof(magic)); Finish(); +// CheckMagic(magic, sizeof(magic)); LLOG("~CoWork " << FormatIntHex(this)); } diff --git a/uppsrc/Core/CoWork.h b/uppsrc/Core/CoWork.h index 26b2e2835..f4a114527 100644 --- a/uppsrc/Core/CoWork.h +++ b/uppsrc/Core/CoWork.h @@ -39,6 +39,7 @@ public: static thread_local bool is_worker; static thread_local Pool *pool; +// byte magic[sizeof(ConditionVariable)]; ConditionVariable waitforfinish; int todo; diff --git a/uppsrc/Core/Debug.cpp b/uppsrc/Core/Debug.cpp index 273f69c21..5963370e3 100644 --- a/uppsrc/Core/Debug.cpp +++ b/uppsrc/Core/Debug.cpp @@ -190,6 +190,19 @@ void LogHex(const WString& s) HexDump(VppLog(), ~s, sizeof(wchar) * s.GetLength()); } +void SetMagic(byte *t, int count) +{ + for(int i = 0; i < count; i++) + t[i] = i; +} + +void CheckMagic(byte *t, int count) +{ + for(int i = 0; i < count; i++) + if(t[i] != i) + Panic("Failed magic area!"); +} + #if defined(PLATFORM_WIN32) && !defined(PLATFORM_WINCE) template diff --git a/uppsrc/Core/Diag.h b/uppsrc/Core/Diag.h index 87ee8e6c2..71e151945 100644 --- a/uppsrc/Core/Diag.h +++ b/uppsrc/Core/Diag.h @@ -33,6 +33,9 @@ void HexDump(Stream& s, const void *ptr, int size, int maxsize = INT_MAX); void LogHex(const String& s); void LogHex(const WString& s); +void SetMagic(byte *t, int count); +void CheckMagic(byte *t, int count); + String GetTypeName(const char *type_name); inline String GetTypeName(const ::std::type_info& tinfo) { return GetTypeName(tinfo.name()); } diff --git a/uppsrc/Core/Stream.cpp b/uppsrc/Core/Stream.cpp index 414f60fff..8bb603d06 100644 --- a/uppsrc/Core/Stream.cpp +++ b/uppsrc/Core/Stream.cpp @@ -147,24 +147,22 @@ bool Stream::GetAll(Huge& h, size_t size) String Stream::Get(int size) { - if(size < 32 * 1024*1024) { - StringBuffer b(size); - int n = Get(~b, size); - b.SetCount(n); - String s = b; - s.Trim(n); - return s; - } - else { - Huge h; - Get(h, size); - return h.Get(); - } + StringBuffer b(size); + int n = Get(~b, size); + b.SetCount(n); + return b; } String Stream::GetAll(int size) { - String result = Get(size); + String result; + if(size < 4 * 1024*1024) + result = Get(size); + else { + Huge h; + Get(h, size); + result = h.Get(); + } if(result.GetCount() != size) { LoadError(); result = String::GetVoid(); diff --git a/uppsrc/Core/lheap.cpp b/uppsrc/Core/lheap.cpp index bb3762681..be5349114 100644 --- a/uppsrc/Core/lheap.cpp +++ b/uppsrc/Core/lheap.cpp @@ -173,7 +173,7 @@ void *Heap::LAlloc(size_t& size) Header *b = (Header *)((byte *)h + BIGHDRSZ - sizeof(Header)); b->size = 0; // header contains large header with size = 0, to detect big during free b->free = false; - LLOG("Big alloc " << (void *)b->GetBlock()); + LLOG("Big alloc " << size << ": " << (void *)b->GetBlock()); return b->GetBlock(); } int bini = SizeToBin((int)size); // get the bin diff --git a/uppsrc/Core/xxHsh.cpp b/uppsrc/Core/xxHsh.cpp index efadbba47..f0213a8aa 100644 --- a/uppsrc/Core/xxHsh.cpp +++ b/uppsrc/Core/xxHsh.cpp @@ -17,6 +17,7 @@ void xxHashStream::Reset(dword seed) void xxHashStream::Out(const void *data, dword size) { + RTIMING("xxh2"); XXH32_update((XXH32_state_t *)context, data, size); } diff --git a/uppsrc/plugin/lz4/Compress.cpp b/uppsrc/plugin/lz4/Compress.cpp index 502c6282e..85a5bb178 100644 --- a/uppsrc/plugin/lz4/Compress.cpp +++ b/uppsrc/plugin/lz4/Compress.cpp @@ -1,5 +1,7 @@ #include "lz4.h" +#ifdef flagCOMPLEXLZ4 + #define LLOG(x) // LOG(x) namespace Upp { @@ -18,7 +20,6 @@ void LZ4CompressStream::Init() void LZ4CompressStream::FinishBlock(char *outbuf, int clen, const char *origdata, int origsize) { - RTIMING("FinishBlock"); if(IsError() || out && out->IsError()) return; if(clen < 0) { @@ -162,3 +163,5 @@ LZ4CompressStream::~LZ4CompressStream() } }; + +#endif \ No newline at end of file diff --git a/uppsrc/plugin/lz4/Compress2.cpp b/uppsrc/plugin/lz4/Compress2.cpp new file mode 100644 index 000000000..99ccb2531 --- /dev/null +++ b/uppsrc/plugin/lz4/Compress2.cpp @@ -0,0 +1,160 @@ +#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(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 \ No newline at end of file diff --git a/uppsrc/plugin/lz4/Decompress.cpp b/uppsrc/plugin/lz4/Decompress.cpp index 39061449d..dd667465c 100644 --- a/uppsrc/plugin/lz4/Decompress.cpp +++ b/uppsrc/plugin/lz4/Decompress.cpp @@ -4,12 +4,17 @@ namespace Upp { + void LZ4DecompressStream::Init() { + for(int i = 0; i < 16; i++) + wb[i].Clear(); + ii = 0; + count = 0; + dlen = 0; pos = 0; eof = false; - buffer.Clear(); - ptr = rdlim = (byte *)~buffer; + ptr = rdlim = NULL; xxh.Reset(); ClearError(); } @@ -58,45 +63,105 @@ bool LZ4DecompressStream::Open(Stream& in_) return true; } -String LZ4DecompressStream::Read(int& blksz) +bool LZ4DecompressStream::Next() { - if(IsError() || in->IsError()) - return Null; - blksz = in->Get32le(); - if(blksz == 0) // This is EOF - return Null; - int len = blksz & 0x7fffffff; - if(len > maxblock) { - SetError(); - return Null; + RTIMING("Next"); + if(ii < count) { + pos += dlen; + ptr = (byte *)~wb[ii].d; + dlen = wb[ii].dlen; + rdlim = ptr + dlen; + ii++; + return true; } - String data = in->GetAll(len); - if(IsNull(data)) { - SetError(); - return Null; - } - if(lz4hdr & LZ4F_BLOCKCHECKSUM) - in->Get32le(); - return data; + return false; } -String LZ4DecompressStream::Fetch() + +void LZ4DecompressStream::Fetch() { - RTIMING("Fetch"); - int blksz; - String data = Read(blksz); - if((blksz & 0x80000000) || IsNull(data)) - return data; - StringBuffer b(maxblock); - int sz = LZ4_decompress_safe(~data, ~b, data.GetCount(), maxblock); - if(sz < 0) { - SetError(); - return 0; + if(Next()) + return; + if(eof) + return; +#ifdef _MULTITHREADED + CoWork co; +#endif + bool error = false; + bool last = false; + ii = 0; + count = concurrent ? 16 : 1; + for(int i = 0; i < count; i++) { + Workblock& t = wb[i]; + int blksz = in->Get32le(); + if(blksz == 0) { // This is EOF + last = true; + count = i; + break; + } + t.clen = blksz & 0x7fffffff; + if(t.clen > maxblock) { + SetError(); + return; + } + if(!t.c) { + RTIMING("Alloc"); + t.c.Alloc(maxblock); + t.d.Alloc(maxblock); + } + if(blksz & 0x80000000) { // block is not compressed + t.dlen = t.clen; + if(!in->GetAll(~t.d, t.clen)) { + SetError(); + return; + } + } + else { + { RTIMING("GetAll"); + if(!in->GetAll(~t.c, t.clen)) { + SetError(); + return; + } + } +#ifdef _MULTITHREADED + if(concurrent) + co & [=, &error] { + Workblock& t = wb[i]; + t.dlen = LZ4_decompress_safe(~t.c, ~t.d, t.clen, maxblock); + CoWork::FinLock(); + if(t.dlen < 0) + error = true; + }; + else +#endif + { + RTIMING("LZ4 decompress"); + t.dlen = LZ4_decompress_safe(~t.c, ~t.d, t.clen, maxblock); + if(t.dlen < 0) + error = true; + } + } + if(lz4hdr & LZ4F_BLOCKCHECKSUM) + in->Get32le(); // just skip it + } +#ifdef _MULTITHREADED + if(concurrent) + co.Finish(); +#endif + if(error) + SetError(); + else { + for(int i = 0; i < count; i++) { + RTIMING("xxh"); + xxh.Put(wb[i].d, wb[i].dlen); + } + if(last) { + if(in->Get32le() != xxh.Finish()) + SetError(); + eof = true; + } + Next(); } - if(sz <= 0) - return Null; - b.SetLength(sz); - return b; } bool LZ4DecompressStream::IsOpen() const @@ -104,33 +169,11 @@ bool LZ4DecompressStream::IsOpen() const return in->IsOpen() && !IsError(); } -void LZ4DecompressStream::CheckEof() -{ - if(!eof) { - if(in->Get32le() != xxh.Finish()) - SetError(); - eof = true; - } -} - -void LZ4DecompressStream::NewBuffer(const String& s) -{ - RTIMING("NewBuffer"); - pos += buffer.GetCount(); - buffer = s; - ptr = (byte *)buffer.begin(); - rdlim = (byte *)buffer.end(); - { RTIMING("XXH"); - xxh.Put(s, s.GetCount()); } - if(ptr == rdlim) - CheckEof(); -} - int LZ4DecompressStream::_Term() { if(eof) return -1; - NewBuffer(Fetch()); + Fetch(); return ptr == rdlim ? -1 : *ptr; } @@ -138,34 +181,32 @@ int LZ4DecompressStream::_Get() { if(eof) return -1; - NewBuffer(Fetch()); + Fetch(); return ptr == rdlim ? -1 : *ptr++; } -int64 copied; -EXITBLOCK { RDUMP(copied); } - dword LZ4DecompressStream::_Get(void *data, dword size) { RTIMING("_Get"); byte *t = (byte *)data; while(size) { - if(IsError() || in->IsError() || IsEof()) + if(IsError() || in->IsError() || ptr == rdlim && ii == count && eof) break; dword n = dword(rdlim - ptr); if(size < n) { - { RTIMING("memcpy1"); - memcpy(t, ptr, size); } + RTIMING("memcpy1"); + memcpy(t, ptr, size); t += size; ptr += size; break; } else { - { RTIMING("memcpy2"); copied += n; + { RTIMING("memcpy2"); memcpy(t, ptr, n); } t += n; size -= n; - NewBuffer(Fetch()); + ptr = rdlim; + Fetch(); } } @@ -175,6 +216,7 @@ dword LZ4DecompressStream::_Get(void *data, dword size) LZ4DecompressStream::LZ4DecompressStream() { in = NULL; + concurrent = false; } LZ4DecompressStream::~LZ4DecompressStream() diff --git a/uppsrc/plugin/lz4/lz4.h b/uppsrc/plugin/lz4/lz4.h index 267bb1d27..42c9c79c5 100644 --- a/uppsrc/plugin/lz4/lz4.h +++ b/uppsrc/plugin/lz4/lz4.h @@ -84,6 +84,7 @@ public: ~Lz4(); }; +#ifdef flagCOMPLEXLZ4 class LZ4CompressStream : public Stream { public: virtual void Close(); @@ -113,7 +114,7 @@ protected: int inblock; #endif - void Init(); + void Init(); void SetupBuffer(); void FinishBlock(char *outbuf, int clen, const char *origdata, int origsize); void FlushOut(); @@ -130,6 +131,47 @@ public: LZ4CompressStream(Stream& out) : LZ4CompressStream() { Open(out); } ~LZ4CompressStream(); }; +#else +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; + + Buffer buffer; + Buffer outbuf; + Buffer outsz; + + enum { BLOCK_BYTES = 1024 * 1024 }; + + xxHashStream xxh; + +#ifdef _MULTITHREADED + bool concurrent; +#endif + + void Alloc(); + void Init(); + void FlushOut(); + +public: + Event WhenPos; + +#ifdef _MULTITHREADED + void Concurrent(bool b = true); +#endif + void Open(Stream& out_); + + LZ4CompressStream(); + LZ4CompressStream(Stream& out) : LZ4CompressStream() { Open(out); } + ~LZ4CompressStream(); +}; +#endif class LZ4DecompressStream : public Stream { public: @@ -142,8 +184,17 @@ protected: private: Stream *in; - String buffer; - Vector ahead; + struct Workblock { + Buffer c, d; // compressed, decompressed data + int clen = 0, dlen = 0; // compressed, decompressed len + + bool Decompress(); + void Clear() { c.Clear(); d.Clear(); } + }; + Workblock wb[16]; + int count; // count of workblocks fetched + int ii; // next workblock to be read + int dlen; // length of current workblock enum { BLOCK_BYTES = 1024*1024 }; @@ -152,14 +203,14 @@ private: int blockchksumsz; byte lz4hdr; bool eof; + + bool concurrent; void TryHeader(); void Init(); - String Read(int& blksz); - String Fetch(); - void CheckEof(); - void NewBuffer(const String& s); + bool Next(); + void Fetch(); public: Callback2 WhenOut; @@ -167,7 +218,7 @@ public: bool Open(Stream& in); #ifdef _MULTITHREADED - void Concurrent(bool b = true) { co = b; } + void Concurrent(bool b = true) { concurrent = b; } #endif LZ4DecompressStream(); diff --git a/uppsrc/plugin/lz4/lz4.upp b/uppsrc/plugin/lz4/lz4.upp index 4dbcfad90..66afcd908 100644 --- a/uppsrc/plugin/lz4/lz4.upp +++ b/uppsrc/plugin/lz4/lz4.upp @@ -7,6 +7,7 @@ file lz4.h, lz4upp.cpp, Compress.cpp, + Compress2.cpp, Decompress.cpp, util.cpp, lib\LICENSE, diff --git a/uppsrc/plugin/lz4/util.cpp b/uppsrc/plugin/lz4/util.cpp index b59631aa2..f7c332deb 100644 --- a/uppsrc/plugin/lz4/util.cpp +++ b/uppsrc/plugin/lz4/util.cpp @@ -4,10 +4,11 @@ namespace Upp { static void sCopy(Stream& out, Stream& in, Gate2 progress) { + const int CHUNK = 16384; + Buffer b(CHUNK); while(!in.IsEof()) { // TODO: progress!!! - String h; - h = in.Get(4 * 1024*1024); - out.Put(h); + int n = in.Get(b, CHUNK); + out.Put(b, n); } }