Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Compress to buffer812 #18034

Closed
wants to merge 3 commits into from
Closed

Compress to buffer812 #18034

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
caeed48
HPCC-29917 Refactor compressToBuffer to support different compression…
richardkchapman Nov 1, 2023
bd6d36d
HPCC-29917 Refactor compressToBuffer to support different compression…
richardkchapman Nov 15, 2023
2f70a84
WIP
richardkchapman Nov 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions common/dllserver/thorplugin.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -676,10 +676,7 @@ extern DLLSERVER_API bool decompressResource(size32_t len, const void *data, Str
extern DLLSERVER_API void appendResource(MemoryBuffer & mb, size32_t len, const void *data, bool compress)
{
mb.append((byte)0x80).append(resourceHeaderVersion);
if (compress)
compressToBuffer(mb, len, data);
else
appendToBuffer(mb, len, data);
compressToBuffer(mb, len, data, compress ? COMPRESS_METHOD_LZW : COMPRESS_METHOD_NONE);
}

extern DLLSERVER_API void compressResource(MemoryBuffer & compressed, size32_t len, const void *data)
Expand Down
8 changes: 4 additions & 4 deletions system/jlib/jfcmp.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,8 @@ class jlib_decl CFcmpCompressor : public CSimpleInterfaceOf<ICompressor>

virtual void open(void *buf,size32_t max)
{
if (max<1024)
throw MakeStringException(-1,"CFcmpCompressor::open - block size (%d) not large enough", max);
// if (max<1024)
// throw MakeStringException(-1,"CFcmpCompressor::open - block size (%d) not large enough", max);
wrmax = max;
originalMax = max;
if (buf)
Expand Down Expand Up @@ -103,8 +103,8 @@ class jlib_decl CFcmpCompressor : public CSimpleInterfaceOf<ICompressor>
{
if (!initialSize)
initialSize = FCMP_BUFFER_SIZE; // 1MB
if (initialSize<1024)
throw MakeStringException(-1,"CFcmpCompressor::open - block size (%d) not large enough", initialSize);
//if (initialSize<1024)
// throw MakeStringException(-1,"CFcmpCompressor::open - block size (%d) not large enough", initialSize);
wrmax = initialSize;
if (bufalloc)
{
Expand Down
11 changes: 5 additions & 6 deletions system/jlib/jlz4.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,18 +33,17 @@ class CLZ4Compressor final : public CFcmpCompressor
protected:
virtual void setinmax() override
{
inmax = blksz-outlen-sizeof(size32_t);
if (inmax<256)
if (blksz <= outlen+sizeof(size32_t))
trailing = true; // too small to bother compressing
else
{
trailing = false;
inmax = blksz-outlen-sizeof(size32_t);
size32_t slack = LZ4_COMPRESSBOUND(inmax) - inmax;
int inmax2 = inmax - (slack + sizeof(size32_t));
if (inmax2<256)
if (inmax <= (slack + sizeof(size32_t)))
trailing = true;
else
inmax = inmax2;
inmax = inmax - (slack + sizeof(size32_t));
}
}

Expand Down Expand Up @@ -72,7 +71,7 @@ class CLZ4Compressor final : public CFcmpCompressor
if (toflush == 0)
return;

if (toflush < 256)
if (false && toflush < 256)
{
trailing = true;
return;
Expand Down
218 changes: 132 additions & 86 deletions system/jlib/jlzw.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -744,82 +744,76 @@ size32_t RLEExpand(void *dst,const void *src,size32_t expsize)
return (size32_t)(in-(const byte *)src);
}

void appendToBuffer(MemoryBuffer & out, size32_t len, const void * src)
void compressToBuffer(MemoryBuffer & out, size32_t len, const void * src, CompressionMethod method, const char *options)
{
out.append(false);
out.append(len);
out.append(len, src);
}

void compressToBuffer(MemoryBuffer & out, size32_t len, const void * src)
{
unsigned originalLength = out.length();
out.append(true);
out.append((size32_t)0);

if (len >= 32)
if (method != COMPRESS_METHOD_NONE && len >= 32)
{
ICompressHandler *handler = queryCompressHandler(method);
if (!handler)
{
VStringBuffer s("Unknown compression method %x requested in compressToBuffer", (byte) method);
throw makeStringException(0, s.str());
}
unsigned originalLength = out.length();
// For back-compatibility, we always store COMPRESS_METHOD_LZW as 1 as earlier versions stored a boolean here
// rather than an enum
// This means that compressToBuffer/decompressToBuffer cannot bs used for rowdiff compression - this is not likely to be an issue
// Alternative would be a separate enum for compressToBuffer formats, but that seems more likely to cause confusion
out.append((byte) (method == COMPRESS_METHOD_LZW ? COMPRESS_METHOD_LZWLEGACY : method));
out.append((size32_t)0);
size32_t newSize = len * 4 / 5; // Copy if compresses less than 80% ...
Owned<ICompressor> compressor = createLZWCompressor();
Owned<ICompressor> compressor = handler->getCompressor(options);
void *newData = out.reserve(newSize);
compressor->open(newData, newSize);
if (compressor->write(src, len)==len)
try
{
compressor->close();
size32_t compressedLen = compressor->buflen();
out.setWritePos(originalLength + sizeof(bool));
out.append(compressedLen);
out.setWritePos(originalLength + sizeof(bool) + sizeof(size32_t) + compressedLen);
return;
compressor->open(newData, newSize);
if (compressor->write(src, len)==len)
{
compressor->close();
size32_t compressedLen = compressor->buflen();
out.setWritePos(originalLength + sizeof(byte));
out.append(compressedLen);
out.setWritePos(originalLength + sizeof(byte) + sizeof(size32_t) + compressedLen);
return;
}
}
catch (IException *E)
{
E->Release();
}
// failed to compress...
out.setWritePos(originalLength);
}

// all or don't compress
out.setWritePos(originalLength);
appendToBuffer(out, len, src);
}

void decompressToBuffer(MemoryBuffer & out, const void * src)
{
Owned<IExpander> expander = createLZWExpander();
unsigned outSize = expander->init(src);
void * buff = out.reserve(outSize);
expander->expand(buff);
out.append((byte) COMPRESS_METHOD_NONE);
out.append(len);
out.append(len, src);
}


void decompressToBuffer(MemoryBuffer & out, MemoryBuffer & in)
void decompressToBuffer(MemoryBuffer & out, MemoryBuffer & in, const char *options)
{
bool compressed;
size32_t srcLen;
in.read(compressed).read(srcLen);
if (compressed)
decompressToBuffer(out, in.readDirect(srcLen));
else
unsigned char _method;
in.read(_method).read(srcLen);
CompressionMethod method = (CompressionMethod) _method;
if (method==COMPRESS_METHOD_NONE)
out.append(srcLen, in.readDirect(srcLen));
}

void decompressToAttr(MemoryAttr & out, const void * src)
{
Owned<IExpander> expander = createLZWExpander();
unsigned outSize = expander->init(src);
void * buff = out.allocate(outSize);
expander->expand(buff);
}

void decompressToBuffer(MemoryAttr & out, MemoryBuffer & in)
{
bool compressed;
size32_t srcLen;
in.read(compressed).read(srcLen);
if (compressed)
decompressToAttr(out, in.readDirect(srcLen));
else
out.set(srcLen, in.readDirect(srcLen));
{
if (method==COMPRESS_METHOD_LZWLEGACY)
method = COMPRESS_METHOD_LZW; // Back compatibilty
ICompressHandler *handler = queryCompressHandler(method);
if (!handler)
{
VStringBuffer s("Unknown decompression method %x required in decompressToBuffer", (byte) method);
throw makeStringException(0, s.str());
}
Owned<IExpander> expander = handler->getExpander(options);
unsigned outSize = expander->init(in.readDirect(srcLen));
void * buff = out.reserve(outSize);
expander->expand(buff);
}
}



/*
Simple Diff compression format is

Expand Down Expand Up @@ -2604,6 +2598,8 @@ class CAESCompressor : implements ICompressor, public CInterface
else
outbuf = outattr.allocate(blksize);
outBufMb = NULL;
if (blksize <= AES_PADDING_SIZE+sizeof(size32_t))
throw makeStringException(0, "CAESCompressor: target buffer too small");
size32_t subsz = blksize-AES_PADDING_SIZE-sizeof(size32_t);
comp->open(compattr.reserveTruncate(subsz),subsz);
}
Expand Down Expand Up @@ -2788,44 +2784,88 @@ IPropertyTree *getBlockedFileDetails(IFile *file)
return NULL;
}

class CCompressHandlerArray : public IArrayOf<ICompressHandler>
class CCompressHandlerArray
{
IArrayOf<ICompressHandler> registered; // Owns the relevant handler objects
ICompressHandler *byMethod[COMPRESS_METHOD_LAST] = { nullptr };
ICompressHandler *AESbyMethod[COMPRESS_METHOD_LAST] = { nullptr };

public:
ICompressHandler *lookup(const char *type) const
{
ForEachItemIn(h, *this)
ForEachItemIn(h, registered)
{
ICompressHandler &handler = item(h);
ICompressHandler &handler = registered.item(h);
if (0 == stricmp(type, handler.queryType()))
return &handler;
}
return NULL;
}
ICompressHandler *lookup(CompressionMethod method) const
{
if ((method & ~COMPRESS_METHOD_AES) >= COMPRESS_METHOD_LAST)
return nullptr;
else if (method & COMPRESS_METHOD_AES)
return AESbyMethod[method & ~COMPRESS_METHOD_AES];
else
return byMethod[method];
}
ICompressHandlerIterator *getIterator()
{
return new ArrayIIteratorOf<IArrayOf<ICompressHandler>, ICompressHandler, ICompressHandlerIterator>(registered);
}
bool addCompressor(ICompressHandler *handler)
{
CompressionMethod method = handler->queryMethod();
if (lookup(method))
{
handler->Release();
return false; // already registered
}
registered.append(* handler);
if ((method & ~COMPRESS_METHOD_AES) < COMPRESS_METHOD_LAST)
{
if (method & COMPRESS_METHOD_AES)
AESbyMethod[method & ~COMPRESS_METHOD_AES] = handler;
else
byMethod[method] = handler;
}
return true;
}
bool removeCompressor(ICompressHandler *handler)
{
CompressionMethod method = handler->queryMethod();
if (registered.zap(* handler))
{
if ((method & ~COMPRESS_METHOD_AES) < COMPRESS_METHOD_LAST)
{
if (method & COMPRESS_METHOD_AES)
AESbyMethod[method & ~COMPRESS_METHOD_AES] = handler;
else
byMethod[method] = handler;
}
return true;
}
else
return false;
}
} compressors;

typedef IIteratorOf<ICompressHandler> ICompressHandlerIterator;

ICompressHandlerIterator *getCompressHandlerIterator()
{
return new ArrayIIteratorOf<IArrayOf<ICompressHandler>, ICompressHandler, ICompressHandlerIterator>(compressors);
return compressors.getIterator();
}



bool addCompressorHandler(ICompressHandler *handler)
{
if (compressors.lookup(handler->queryType()))
{
handler->Release();
return false; // already registered
}
compressors.append(* handler);
return true;
return compressors.addCompressor(handler);
}

bool removeCompressorHandler(ICompressHandler *handler)
{
return compressors.zap(* handler);
return compressors.removeCompressor(handler);
}

Linked<ICompressHandler> defaultCompressor;
Expand All @@ -2834,38 +2874,38 @@ MODULE_INIT(INIT_PRIORITY_STANDARD)
{
class CCompressHandlerBase : implements ICompressHandler, public CInterface
{
StringAttr type;
public:
IMPLEMENT_IINTERFACE;
CCompressHandlerBase(const char *_type) : type(_type) { }
// ICompressHandler
virtual const char *queryType() const { return type; }
};
class CFLZCompressHandler : public CCompressHandlerBase
{
public:
CFLZCompressHandler() : CCompressHandlerBase("FLZ") { }
virtual const char *queryType() const { return "FLZ"; }
virtual CompressionMethod queryMethod() const { return COMPRESS_METHOD_FASTLZ; }
virtual ICompressor *getCompressor(const char *options) { return createFastLZCompressor(); }
virtual IExpander *getExpander(const char *options) { return createFastLZExpander(); }
};
class CLZ4CompressHandler : public CCompressHandlerBase
{
public:
CLZ4CompressHandler() : CCompressHandlerBase("LZ4") { }
virtual const char *queryType() const { return "LZ4"; }
virtual CompressionMethod queryMethod() const { return COMPRESS_METHOD_LZ4; }
virtual ICompressor *getCompressor(const char *options) { return createLZ4Compressor(false); }
virtual IExpander *getExpander(const char *options) { return createLZ4Expander(); }
};
class CLZ4HCCompressHandler : public CCompressHandlerBase
{
public:
CLZ4HCCompressHandler() : CCompressHandlerBase("LZ4HC") { }
virtual const char *queryType() const { return "LZ4HC"; }
virtual CompressionMethod queryMethod() const { return COMPRESS_METHOD_LZ4HC; }
virtual ICompressor *getCompressor(const char *options) { return createLZ4Compressor(true); }
virtual IExpander *getExpander(const char *options) { return createLZ4Expander(); }
};
class CAESCompressHandler : public CCompressHandlerBase
{
public:
CAESCompressHandler() : CCompressHandlerBase("AES") { }
virtual const char *queryType() const { return "AES"; }
virtual CompressionMethod queryMethod() const { return (CompressionMethod) (COMPRESS_METHOD_AES|COMPRESS_METHOD_LZW); }
virtual ICompressor *getCompressor(const char *options)
{
assertex(options);
Expand All @@ -2880,20 +2920,22 @@ MODULE_INIT(INIT_PRIORITY_STANDARD)
class CDiffCompressHandler : public CCompressHandlerBase
{
public:
CDiffCompressHandler() : CCompressHandlerBase("DIFF") { }
virtual const char *queryType() const { return "DIFF"; }
virtual CompressionMethod queryMethod() const { return COMPRESS_METHOD_ROWDIF; }
virtual ICompressor *getCompressor(const char *options) { return createRDiffCompressor(); }
virtual IExpander *getExpander(const char *options) { return createRDiffExpander(); }
};
class CLZWCompressHandler : public CCompressHandlerBase
{
public:
CLZWCompressHandler() : CCompressHandlerBase("LZW") { }
virtual const char *queryType() const { return "LZW"; }
virtual CompressionMethod queryMethod() const { return COMPRESS_METHOD_LZW; }
virtual ICompressor *getCompressor(const char *options) { return createLZWCompressor(true); }
virtual IExpander *getExpander(const char *options) { return createLZWExpander(true); }
};
addCompressorHandler(new CLZWCompressHandler());
addCompressorHandler(new CAESCompressHandler());
addCompressorHandler(new CDiffCompressHandler());
addCompressorHandler(new CLZWCompressHandler());
addCompressorHandler(new CFLZCompressHandler());
addCompressorHandler(new CLZ4HCCompressHandler());
ICompressHandler *lz4Compressor = new CLZ4CompressHandler();
Expand All @@ -2906,6 +2948,10 @@ ICompressHandler *queryCompressHandler(const char *type)
{
return compressors.lookup(type);
}
ICompressHandler *queryCompressHandler(CompressionMethod method)
{
return compressors.lookup(method);
}

void setDefaultCompressor(const char *type)
{
Expand Down
Loading