Накопительное сжатие растущего буфера (C++, zlib)

Question

У меня есть буфер (строка), который растет с течением времени, и мне нужно отправить этот буфер, хотя канал с ограниченным размером ввода (4096 байт). Связь через этот канал является дорогостоящей, поэтому лучше отправлять сжатые данные. Рост буфера происходит блоками разного размера. Эти блоки нельзя разбить или потерять смысл.

Я на самом деле использую zlib в C++ для сжатия с ограничением размера буфера. Когда этот предел достигнут, строка сжимается и отправляется, думая о канале. Это работает, но это не оптимально, потому что предел довольно низок для того, чтобы не потерять информацию (ограничение входного канала - 4096 байт).

Моя идея - использовать zlib для создания растущего сжатого буфера с блоками сжатия разного размера и остановить процесс до достижения предела ввода канала. Может ли zlib работать с блоками сжатия разного размера или мне нужен еще один алгоритм?

Answer 1

Мне удалось создать компрессор, который отправляет растущую часть буфера по частям через канал с ограниченным размером ввода. Я поставил здесь ответ для тех, кто работает над одной и той же проблемой. Thx, чтобы отметить Адлера и до MSalters для того, чтобы привести меня к правильному пути.

class zStreamManager { 
    public: 
     zStreamManager(); 
     ~zStreamManager(); 
     void endStream(); 
     void addToStream(const void *inData, size_t inDataSize); 

    private: 
     // Size of base64 encoded is about 4*originalSize/3 + (3 to 6) 
     // so with maximum output size of 4096, 3050 max zipped out 
     // buffer will be fine 
     const size_t CHUNK_IN = 1024, CHUNK_OUT = 3050; 
     const std::string base64Chars = 
     "ABCDEFGHIJKLMNOPQRSTUVWXYZ" 
     "abcdefghijklmnopqrstuvwxyz" 
     "/"; 
     bool deallocated = true; 
     z_stream stream; 
     std::vector<uint8_t> outBuffer; 
     std::string base64Encode(std::vector<uint8_t> &str); 
}; 

zStreamManager::~zStreamManager() { 
    endStream(); 
} 

void zStreamManager::endStream() { 
    if(!deallocated) { 
     deallocated = true; 
     uint8_t tempBuffer[CHUNK_IN]; 
     int response = Z_OK; 
     unsigned int have; 

     while(response == Z_OK) { 
      if (stream.avail_out == 0) { 
       outBuffer.insert(outBuffer.end(), tempBuffer, tempBuffer + CHUNK_IN); 
       stream.next_out = tempBuffer; 
       stream.avail_out = CHUNK_IN; 
      } 
      response = deflate(&stream, Z_FINISH); 
     } 

     have = CHUNK_IN - stream.avail_out; 
     if(have) 
      outBuffer.insert(outBuffer.end(), tempBuffer, tempBuffer + have); 

     deflateEnd(&stream); 

     if(outBuffer.size()) 
      SEND << outBuffer << "$"; 
    } 
} 

void zStreamManager::addToStream(const void *inData, size_t inDataSize) { 
    if(deallocated) { 
     deallocated = false; 
     stream.zalloc = 0; 
     stream.zfree = 0; 
     stream.opaque = 0; 
     deflateInit(&stream, 9); 
    } 

    std::vector<uint8_t> tempBuffer(inDataSize); 
    unsigned int have; 

    stream.next_in = reinterpret_cast<uint8_t *>(const_cast<void*>(inData)); 
    stream.avail_in = inDataSize; 
    stream.next_out = &tempBuffer[0]; 
    stream.avail_out = inDataSize; 

    while (stream.avail_in != 0) { 
     deflate(&stream, Z_SYNC_FLUSH); 
     if (stream.avail_out == 0) { 
      outBuffer.insert(outBuffer.end(), tempBuffer.begin(), tempBuffer.begin() + inDataSize); 
      stream.next_out = &tempBuffer[0]; 
      stream.avail_out = inDataSize; 
     } 
    } 

    have = inDataSize - stream.avail_out; 
    if(have) 
     outBuffer.insert(outBuffer.end(), tempBuffer.begin(), tempBuffer.begin() + have); 

    while(outBuffer.size() >= CHUNK_OUT) { 
     std::vector<uint8_t> zipped; 

     zipped.insert(zipped.end(), outBuffer.begin(), outBuffer.begin() + CHUNK_OUT); 
     outBuffer.erase(outBuffer.begin(), outBuffer.begin() + CHUNK_OUT); 

     if(zipped.size()) 
      SEND << zipped << "|"; 
    } 
} 

std::string zStreamManager::base64Encode(std::vector<uint8_t> &str) { 
    /* ALTERED VERSION OF René Nyffenegger BASE64 CODE 
    Copyright (C) 2004-2008 René Nyffenegger 

    This source code is provided 'as-is', without any express or implied 
    warranty. In no event will the author be held liable for any damages 
    arising from the use of this software. 

    Permission is granted to anyone to use this software for any purpose, 
    including commercial applications, and to alter it and redistribute it 
    freely, subject to the following restrictions: 

    1. The origin of this source code must not be misrepresented; you must not 
     claim that you wrote the original source code. If you use this source code 
     in a product, an acknowledgment in the product documentation would be 
     appreciated but is not required. 

    2. Altered source versions must be plainly marked as such, and must not be 
     misrepresented as being the original source code. 

    3. This notice may not be removed or altered from any source distribution. 

    René Nyffenegger rene.nyffenegger@adp-gmbh.ch 
    */ 
    unsigned char const* bytes_to_encode = &str[0]; 
    unsigned int in_len = str.size(); 
    std::string ret; 
    int i = 0, j = 0; 
    unsigned char char_array_3[3], char_array_4[4]; 

    while(in_len--) { 
    char_array_3[i++] = *(bytes_to_encode++); 
    if (i == 3) { 
     char_array_4[0] = (char_array_3[0] & 0xfc) >> 2; 
     char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4); 
     char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6); 
     char_array_4[3] = char_array_3[2] & 0x3f; 

     for(i = 0; (i <4) ; i++) 
     ret += base64Chars[char_array_4[i]]; 
     i = 0; 
    } 
    } 

    if(i) { 
    for(j = i; j < 3; j++) 
     char_array_3[j] = '\0'; 

    char_array_4[0] = (char_array_3[0] & 0xfc) >> 2; 
    char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4); 
    char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6); 
    char_array_4[3] = char_array_3[2] & 0x3f; 

    for(j = 0; (j < i + 1); j++) 
     ret += base64Chars[char_array_4[j]]; 

    while((i++ < 3)) 
     ret += '='; 
    } 

    return ret; 
} 

Прецедент:

zStreamManager zm; 
string growingBuffer = ""; 
bool somethingToSend = true; 

while(somethingToSend) { 
    RECEIVE(&growingBuffer); 
    if(growingBuffer.size()) { 
    zm.addToStream(growingBuffer.c_str(), growingBuffer.size()); 
    growingBuffer.clear(); 
    } else { 
    somethingToSend = false; 
    } 
} 

zm.endStream(); 

RECEIVE С и SEND, методы, используемые для получения буфера и отправки его через канал. Для разжатия каждая часть разделяется символом '|' символ и конец всего буфера разделяются символом '$'. Каждая часть должна быть декодирована base64, а затем конкатенирована. Наконец, он может быть несжатым с zlib, как и любые другие сжатые данные.

Answer 2

Простейшим решением является преобразование внеполосного разграничения пакетов в внутриполосный формат. Самый простой способ сделать это - когда ваши блоки ввода не используют все 256 возможных байтов. Например. когда значение 00 не встречается в блоках, его можно использовать для разделения блоков до сжатия. В противном случае вам понадобится escape-код.

В любом случае, вы сжимаете непрерывный поток с разделителем блоков. На принимающей стороне вы разгружаете поток, распознаете разделители и собираете блоки.

Answer 3

Вы можете просто выполнять непрерывное сжатие zlib, отправляя данные по вашему каналу каждый раз, когда генерируется 4K сжатых данных. На другом конце вам нужно убедиться, что декомпрессор загружается в 4K блоков сжатых данных в правильном порядке.

Алгоритм спускания в zlib является взрывоопасным, накапливаясь от 16 до 64 КБ или более данных внутри, прежде чем испускать любые сжатые данные, а затем доставляет блок сжатых данных и затем накапливается снова. Таким образом, будет задержка, если вы не запросите сброс данных сбрасывания. Если вы хотите уменьшить латентность, вы можете иметь меньшие блоки путем промывки, с небольшим воздействием на сжатие.