ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make seekable streams work w/ small buffers, misc fixes

Browse Source
This commit is contained in:
Sean Purcell
2017-04-11 13:53:09 -07:00
parent c3ba15e48f
commit a3b7c22604
6 changed files with 110 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdseek_compress.c
+63 -31
View File
@@ -39,6 +39,8 @@ struct ZSTD_seekable_CStream_s {
U32 maxChunkSize;
int checksumFlag;
int writingSeekTable;
};
ZSTD_seekable_CStream* ZSTD_seekable_createCStream()
@@ -52,6 +54,7 @@ ZSTD_seekable_CStream* ZSTD_seekable_createCStream()
zcs->cstream = ZSTD_createCStream();
if (zcs->cstream == NULL) goto failed1;
/* allocate some initial space */
{ size_t const CHUNKLOG_STARTING_CAPACITY = 16;
zcs->chunklog.entries =
malloc(sizeof(chunklogEntry_t) * CHUNKLOG_STARTING_CAPACITY);
@@ -70,7 +73,7 @@ failed1:
size_t ZSTD_seekable_freeCStream(ZSTD_seekable_CStream* zcs)
{
if (zcs == NULL) return 0; /* support free on NULL */
if (zcs == NULL) return 0; /* support free on null */
ZSTD_freeCStream(zcs->cstream);
free(zcs->chunklog.entries);
free(zcs);
@@ -80,12 +83,14 @@ size_t ZSTD_seekable_freeCStream(ZSTD_seekable_CStream* zcs)
size_t ZSTD_seekable_initCStream(ZSTD_seekable_CStream* zcs,
int compressionLevel,
int checksumFlag,
U32 maxChunkSize)
{
zcs->chunklog.size = 0;
zcs->chunkCSize = 0;
zcs->chunkDSize = 0;
/* make sure maxChunkSize has a reasonable value */
if (maxChunkSize > ZSTD_SEEKABLE_MAX_CHUNK_DECOMPRESSED_SIZE) {
return ERROR(compressionParameter_unsupported);
}
@@ -94,11 +99,13 @@ size_t ZSTD_seekable_initCStream(ZSTD_seekable_CStream* zcs,
? maxChunkSize
: ZSTD_SEEKABLE_MAX_CHUNK_DECOMPRESSED_SIZE;
zcs->checksumFlag = 0;
zcs->checksumFlag = checksumFlag;
if (zcs->checksumFlag) {
XXH64_reset(&zcs->xxhState, 0);
}
zcs->writingSeekTable = 0;
return ZSTD_initCStream(zcs->cstream, compressionLevel);
}
@@ -114,10 +121,13 @@ static size_t ZSTD_seekable_logChunk(ZSTD_seekable_CStream* zcs)
};
if (zcs->checksumFlag)
zcs->chunklog.entries[zcs->chunklog.size].checksum =
/* take lower 32 bits of digest */
XXH64_digest(&zcs->xxhState) & 0xFFFFFFFFU;
zcs->chunklog.size++;
/* grow the buffer if required */
if (zcs->chunklog.size == zcs->chunklog.capacity) {
/* exponential size increase for constant amortized runtime */
size_t const newCapacity = zcs->chunklog.capacity * 2;
chunklogEntry_t* const newEntries = realloc(zcs->chunklog.entries,
sizeof(chunklogEntry_t) * newCapacity);
@@ -134,6 +144,7 @@ static size_t ZSTD_seekable_logChunk(ZSTD_seekable_CStream* zcs)
size_t ZSTD_seekable_endChunk(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output)
{
size_t const prevOutPos = output->pos;
/* end the frame */
size_t ret = ZSTD_endStream(zcs->cstream, output);
zcs->chunkCSize += output->pos - prevOutPos;
@@ -165,6 +176,7 @@ size_t ZSTD_seekable_compressStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer*
inLen = MIN(inLen, (size_t)(zcs->maxChunkSize - zcs->chunkDSize));
/* if we haven't finished flushing the last chunk, don't start writing a new one */
if (inLen > 0) {
ZSTD_inBuffer inTmp = { inBase, inLen, 0 };
size_t const prevOutPos = output->pos;
@@ -184,8 +196,12 @@ size_t ZSTD_seekable_compressStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer*
}
if (zcs->maxChunkSize == zcs->chunkDSize) {
/* log the chunk and start over */
size_t const ret = ZSTD_seekable_endChunk(zcs, output);
if (ZSTD_isError(ret)) return ret;
/* get the client ready for the next chunk */
return (size_t)zcs->maxChunkSize;
}
return (size_t)(zcs->maxChunkSize - zcs->chunkDSize);
@@ -204,62 +220,78 @@ static size_t ZSTD_seekable_seekTableSize(ZSTD_seekable_CStream* zcs)
static size_t ZSTD_seekable_writeSeekTable(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output)
{
BYTE* op = (BYTE*) output->dst;
BYTE tmp[4]; /* so that we can work with buffers too small to write a whole word to */
/* repurpose
* zcs->chunkDSize: the current index in the table and
* zcs->chunkCSize: the amount of the table written so far */
* zcs->chunkCSize: the amount of the table written so far
*
* This function is written this way so that if it has to return early
* because of a small buffer, it can keep going where it left off.
*/
size_t const sizePerChunk = 8 + (zcs->checksumFlag?4:0);
size_t const seekTableLen = ZSD_seekable_seekTableSize(zcs);
size_t const seekTableLen = ZSTD_seekable_seekTableSize(zcs);
if (zcs->chunkCSize == 0) {
if (output->size - output->pos < 4) return seekTableLen - zcs->chunkCSize;
MEM_writeLE32(op + output->pos, ZSTD_MAGIC_SKIPPABLE_START);
output->pos += 4;
zcs->chunkCSize += 4;
}
if (zcs->chunkCSize == 4) {
if (output->size - output->pos < 4) return seekTableLen - zcs->chunkCSize;
MEM_writeLE32(op + output->pos, seekTableLen - ZSTD_skippableHeaderSize);
output->pos += 4;
zcs->chunkCSize += 4;
}
#define st_write32(x, o) \
do { \
if (zcs->chunkCSize < (o) + 4) { \
size_t const lenWrite = MIN(output->size - output->pos, \
(o) + 4 - zcs->chunkCSize); \
MEM_writeLE32(tmp, (x)); \
memcpy(op + output->pos, tmp + (zcs->chunkCSize - (o)), lenWrite); \
zcs->chunkCSize += lenWrite; \
output->pos += lenWrite; \
if (lenWrite < 4) return seekTableLen - zcs->chunkCSize; \
} \
} while (0)
st_write32(ZSTD_MAGIC_SKIPPABLE_START, 0);
st_write32(seekTableLen - ZSTD_skippableHeaderSize, 4);
while (zcs->chunkDSize < zcs->chunklog.size) {
if (output->size - output->pos < sizePerChunk) return seekTableLen - zcs->chunkCSize;
MEM_writeLE32(op + output->pos + 0, zcs->chunklog.entries[zcs->chunkDSize].cSize);
MEM_writeLE32(op + output->pos + 4, zcs->chunklog.entries[zcs->chunkDSize].dSize);
st_write32(zcs->chunklog.entries[zcs->chunkDSize].cSize,
ZSTD_skippableHeaderSize + sizePerChunk * zcs->chunkDSize);
st_write32(zcs->chunklog.entries[zcs->chunkDSize].dSize,
ZSTD_skippableHeaderSize + sizePerChunk * zcs->chunkDSize + 4);
if (zcs->checksumFlag) {
MEM_writeLE32(op + output->pos + 8, zcs->chunklog.entries[zcs->chunkDSize].checksum);
st_write32(zcs->chunklog.entries[zcs->chunkDSize].checksum,
ZSTD_skippableHeaderSize + sizePerChunk * zcs->chunkDSize + 8);
}
output->pos += sizePerChunk;
zcs->chunkCSize += sizePerChunk;
zcs->chunkDSize++;
}
if (output->size - output->pos < ZSTD_seekTableFooterSize) return seekTableLen - zcs->chunkCSize;
MEM_writeLE32(op + output->pos, zcs->chunklog.size);
{ BYTE sfd = 0;
st_write32(zcs->chunklog.size, seekTableLen - ZSTD_seekTableFooterSize);
if (output->size - output->pos < 1) return seekTableLen - zcs->chunkCSize;
if (zcs->chunkCSize < seekTableLen - 4) {
BYTE sfd = 0;
sfd |= (zcs->checksumFlag) << 7;
op[output->pos + 4] = sfd;
op[output->pos] = sfd;
output->pos++;
zcs->chunkCSize++;
}
MEM_writeLE32(op + output->pos + 5, ZSTD_SEEKABLE_MAGICNUMBER);
output->pos += ZSTD_seekTableFooterSize;
zcs->chunkCSize += ZSTD_seekTableFooterSize;
st_write32(ZSTD_SEEKABLE_MAGICNUMBER, seekTableLen - 4);
if (zcs->chunkCSize != seekTableLen) return ERROR(GENERIC);
return 0;
#undef st_write32
}
size_t ZSTD_seekable_endStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output)
{
if (zcs->chunkDSize) {
if (!zcs->writingSeekTable && zcs->chunkDSize) {
const size_t endChunk = ZSTD_seekable_endChunk(zcs, output);
if (ZSTD_isError(endChunk)) return endChunk;
/* return an accurate size hint */
if (endChunk) return endChunk + ZSTD_seekable_seekTableLen(zcs);
if (endChunk) return endChunk + ZSTD_seekable_seekTableSize(zcs);
}
zcs->writingSeekTable = 1;
return ZSTD_seekable_writeSeekTable(zcs, output);
}