AuroraMiddleware/zstd
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add raw seek table construction API and parallel compression example

Browse Source
This commit is contained in:
Sean Purcell 2017-04-28 12:17:09 -07:00
parent 7d37ca1d5b
commit 470993c9b1
6 changed files with 358 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
contrib/seekable_format/examples/.gitignore vendored
View File

@ -1,3 +1,4 @@
seekable_compression
seekable_decompression
parallel_processing
parallel_compression

8
contrib/seekable_format/examples/Makefile
View File

@ -9,7 +9,7 @@
# This Makefile presumes libzstd is built, using `make` in / or /lib/
LDFLAGS += -L../../../lib/ -lzstd
LDFLAGS += ../../../lib/libzstd.a
CPPFLAGS += -I../ -I../../../lib -I../../../lib/common
CFLAGS ?= -O3
@ -32,7 +32,11 @@ seekable_decompression : seekable_decompression.c $(SEEKABLE_OBJS)
parallel_processing : parallel_processing.c $(SEEKABLE_OBJS)
$(CC) $(CPPFLAGS) $(CFLAGS) $^ $(LDFLAGS) -o $@ -pthread
parallel_compression : parallel_compression.c $(SEEKABLE_OBJS)
$(CC) $(CPPFLAGS) $(CFLAGS) $^ $(LDFLAGS) -o $@ -pthread
clean:
@rm -f core *.o tmp* result* *.zst \
seekable_compression seekable_decompression parallel_processing
seekable_compression seekable_decompression \
parallel_processing parallel_compression
@echo Cleaning completed

214
contrib/seekable_format/examples/parallel_compression.c Normal file
View File

@ -0,0 +1,214 @@
/**
* Copyright 2017-present, Facebook, Inc.
* All rights reserved.
*
* This source code is licensed under the license found in the
* LICENSE-examples file in the root directory of this source tree.
*/
#include <stdlib.h> // malloc, free, exit, atoi
#include <stdio.h> // fprintf, perror, feof, fopen, etc.
#include <string.h> // strlen, memset, strcat
#define ZSTD_STATIC_LINKING_ONLY
#include <zstd.h> // presumes zstd library is installed
#include <zstd_errors.h>
#if defined(WIN32) || defined(_WIN32)
# include <windows.h>
# define SLEEP(x) Sleep(x)
#else
# include <unistd.h>
# define SLEEP(x) usleep(x * 1000)
#endif
#define XXH_NAMESPACE ZSTD_
#include "xxhash.h"
#include "pool.h" // use zstd thread pool for demo
#include "zstd_seekable.h"
static void* malloc_orDie(size_t size)
{
void* const buff = malloc(size);
if (buff) return buff;
/* error */
perror("malloc:");
exit(1);
}
static FILE* fopen_orDie(const char *filename, const char *instruction)
{
FILE* const inFile = fopen(filename, instruction);
if (inFile) return inFile;
/* error */
perror(filename);
exit(3);
}
static size_t fread_orDie(void* buffer, size_t sizeToRead, FILE* file)
{
size_t const readSize = fread(buffer, 1, sizeToRead, file);
if (readSize == sizeToRead) return readSize; /* good */
if (feof(file)) return readSize; /* good, reached end of file */
/* error */
perror("fread");
exit(4);
}
static size_t fwrite_orDie(const void* buffer, size_t sizeToWrite, FILE* file)
{
size_t const writtenSize = fwrite(buffer, 1, sizeToWrite, file);
if (writtenSize == sizeToWrite) return sizeToWrite; /* good */
/* error */
perror("fwrite");
exit(5);
}
static size_t fclose_orDie(FILE* file)
{
if (!fclose(file)) return 0;
/* error */
perror("fclose");
exit(6);
}
static void fseek_orDie(FILE* file, long int offset, int origin)
{
if (!fseek(file, offset, origin)) {
if (!fflush(file)) return;
}
/* error */
perror("fseek");
exit(7);
}
static long int ftell_orDie(FILE* file)
{
long int off = ftell(file);
if (off != -1) return off;
/* error */
perror("ftell");
exit(8);
}
struct job {
const void* src;
size_t srcSize;
void* dst;
size_t dstSize;
unsigned checksum;
int compressionLevel;
int done;
};
static void compressFrame(void* opaque)
{
struct job* job = opaque;
job->checksum = XXH64(job->src, job->srcSize, 0);
size_t ret = ZSTD_compress(job->dst, job->dstSize, job->src, job->srcSize, job->compressionLevel);
if (ZSTD_isError(ret)) {
fprintf(stderr, "ZSTD_compress() error : %s \n", ZSTD_getErrorName(ret));
exit(20);
}
job->dstSize = ret;
job->done = 1;
}
static void compressFile_orDie(const char* fname, const char* outName, int cLevel, unsigned frameSize, int nbThreads)
{
POOL_ctx* pool = POOL_create(nbThreads, nbThreads);
if (pool == NULL) { fprintf(stderr, "POOL_create() error \n"); exit(9); }
FILE* const fin = fopen_orDie(fname, "rb");
FILE* const fout = fopen_orDie(outName, "wb");
if (ZSTD_compressBound(frameSize) > 0xFFFFFFFFU) { fprintf(stderr, "Frame size too large \n"); exit(10); }
unsigned dstSize = ZSTD_compressBound(frameSize);
fseek_orDie(fin, 0, SEEK_END);
long int length = ftell_orDie(fin);
fseek_orDie(fin, 0, SEEK_SET);
size_t numFrames = (length + frameSize - 1) / frameSize;
struct job* jobs = malloc_orDie(sizeof(struct job) * numFrames);
size_t i;
for(i = 0; i < numFrames; i++) {
void* in = malloc_orDie(frameSize);
void* out = malloc_orDie(dstSize);
size_t inSize = fread_orDie(in, frameSize, fin);
jobs[i].src = in;
jobs[i].srcSize = inSize;
jobs[i].dst = out;
jobs[i].dstSize = dstSize;
jobs[i].compressionLevel = cLevel;
jobs[i].done = 0;
POOL_add(pool, compressFrame, &jobs[i]);
}
ZSTD_frameLog* fl = ZSTD_seekable_createFrameLog(1);
if (fl == NULL) { fprintf(stderr, "ZSTD_seekable_createFrameLog() failed \n"); exit(11); }
for (i = 0; i < numFrames; i++) {
while (!jobs[i].done) SLEEP(5); /* wake up every 5 milliseconds to check */
fwrite_orDie(jobs[i].dst, jobs[i].dstSize, fout);
free((void*)jobs[i].src);
free(jobs[i].dst);
size_t ret = ZSTD_seekable_logFrame(fl, jobs[i].dstSize, jobs[i].srcSize, jobs[i].checksum);
if (ZSTD_isError(ret)) { fprintf(stderr, "ZSTD_seekable_logFrame() error : %s \n", ZSTD_getErrorName(ret)); }
}
{ unsigned char seekTableBuff[1024];
ZSTD_outBuffer out = {seekTableBuff, 1024, 0};
while (ZSTD_seekable_writeSeekTable(fl, &out) != 0) {
fwrite_orDie(seekTableBuff, out.pos, fout);
out.pos = 0;
}
fwrite_orDie(seekTableBuff, out.pos, fout);
}
ZSTD_seekable_freeFrameLog(fl);
free(jobs);
fclose_orDie(fout);
fclose_orDie(fin);
}
static const char* createOutFilename_orDie(const char* filename)
{
size_t const inL = strlen(filename);
size_t const outL = inL + 5;
void* outSpace = malloc_orDie(outL);
memset(outSpace, 0, outL);
strcat(outSpace, filename);
strcat(outSpace, ".zst");
return (const char*)outSpace;
}
int main(int argc, const char** argv) {
const char* const exeName = argv[0];
if (argc!=4) {
printf("wrong arguments\n");
printf("usage:\n");
printf("%s FILE FRAME_SIZE NB_THREADS\n", exeName);
return 1;
}
{ const char* const inFileName = argv[1];
unsigned const frameSize = (unsigned)atoi(argv[2]);
int const nbThreads = atoi(argv[3]);
const char* const outFileName = createOutFilename_orDie(inFileName);
compressFile_orDie(inFileName, outFileName, 5, frameSize, nbThreads);
}
return 0;
}

2
contrib/seekable_format/examples/parallel_processing.c
View File

@ -95,8 +95,6 @@ static void fseek_orDie(FILE* file, long int offset, int origin) {
exit(7);
}
static const char* filename;
struct sum_job {
const char* fname;
unsigned long long sum;

23
contrib/seekable_format/zstd_seekable.h
View File

@ -82,6 +82,29 @@ ZSTDLIB_API size_t ZSTD_seekable_compressStream(ZSTD_seekable_CStream* zcs, ZSTD
ZSTDLIB_API size_t ZSTD_seekable_endFrame(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output);
ZSTDLIB_API size_t ZSTD_seekable_endStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output);
/*= Raw seek table API
* These functions allow for the seek table to be constructed directly.
* This table can then be appended to a file of concatenated frames.
* This allows the frames to be compressed independently, even in parallel,
* and compiled together afterward into a seekable archive.
*
* Use ZSTD_seekable_createFrameLog() to allocate and initialize a tracking
* structure.
*
* Call ZSTD_seekable_logFrame() once for each frame in the archive.
* checksum is optional, and will not be used if checksumFlag was 0 when the
* frame log was created. If present, it should be the least significant 32
* bits of the XXH64 hash of the uncompressed data.
*
* Call ZSTD_seekable_writeSeekTable to serialize the data into a seek table.
* If the entire table was written, the return value will be 0. Otherwise,
* it will be equal to the number of bytes left to write. */
typedef struct ZSTD_frameLog_s ZSTD_frameLog;
ZSTDLIB_API ZSTD_frameLog* ZSTD_seekable_createFrameLog(int checksumFlag);
ZSTDLIB_API size_t ZSTD_seekable_freeFrameLog(ZSTD_frameLog* fl);
ZSTDLIB_API size_t ZSTD_seekable_logFrame(ZSTD_frameLog* fl, unsigned compressedSize, unsigned decompressedSize, unsigned checksum);
ZSTDLIB_API size_t ZSTD_seekable_writeSeekTable(ZSTD_frameLog* fl, ZSTD_outBuffer* output);
/*-****************************************************************************
* Seekable decompression - HowTo
* A ZSTD_seekable object is required to tracking the seekTable.

185
contrib/seekable_format/zstdseek_compress.c
View File

@ -35,15 +35,21 @@ typedef struct {
U32 checksum;
} framelogEntry_t;
typedef struct {
struct ZSTD_frameLog_s {
framelogEntry_t* entries;
U32 size;
U32 capacity;
int checksumFlag;
/* for use when streaming out the seek table */
U32 seekTablePos;
U32 seekTableIndex;
} framelog_t;
struct ZSTD_seekable_CStream_s {
ZSTD_CStream* cstream;
framelog_t framelog;
ZSTD_frameLog framelog;
U32 frameCSize;
U32 frameDSize;
@ -52,13 +58,52 @@ struct ZSTD_seekable_CStream_s {
U32 maxFrameSize;
int checksumFlag;
int writingSeekTable;
U32 seekTablePos;
U32 seekTableIndex;
};
size_t ZSTD_seekable_frameLog_allocVec(ZSTD_frameLog* fl)
{
/* allocate some initial space */
size_t const FRAMELOG_STARTING_CAPACITY = 16;
fl->entries = (framelogEntry_t*)malloc(
sizeof(framelogEntry_t) * FRAMELOG_STARTING_CAPACITY);
if (fl->entries == NULL) return ERROR(memory_allocation);
fl->capacity = FRAMELOG_STARTING_CAPACITY;
return 0;
}
size_t ZSTD_seekable_frameLog_freeVec(ZSTD_frameLog* fl)
{
if (fl != NULL) free(fl->entries);
return 0;
}
ZSTD_frameLog* ZSTD_seekable_createFrameLog(int checksumFlag)
{
ZSTD_frameLog* fl = malloc(sizeof(ZSTD_frameLog));
if (fl == NULL) return NULL;
if (ZSTD_isError(ZSTD_seekable_frameLog_allocVec(fl))) {
free(fl);
return NULL;
}
fl->checksumFlag = checksumFlag;
fl->seekTablePos = 0;
fl->seekTableIndex = 0;
fl->size = 0;
return fl;
}
size_t ZSTD_seekable_freeFrameLog(ZSTD_frameLog* fl)
{
ZSTD_seekable_frameLog_freeVec(fl);
free(fl);
return 0;
}
ZSTD_seekable_CStream* ZSTD_seekable_createCStream()
{
ZSTD_seekable_CStream* zcs = malloc(sizeof(ZSTD_seekable_CStream));
@ -70,13 +115,7 @@ ZSTD_seekable_CStream* ZSTD_seekable_createCStream()
zcs->cstream = ZSTD_createCStream();
if (zcs->cstream == NULL) goto failed1;
/* allocate some initial space */
{ size_t const FRAMELOG_STARTING_CAPACITY = 16;
zcs->framelog.entries = (framelogEntry_t*)malloc(
sizeof(framelogEntry_t) * FRAMELOG_STARTING_CAPACITY);
if (zcs->framelog.entries == NULL) goto failed2;
zcs->framelog.capacity = FRAMELOG_STARTING_CAPACITY;
}
if (ZSTD_isError(ZSTD_seekable_frameLog_allocVec(&zcs->framelog))) goto failed2;
return zcs;
@ -91,7 +130,7 @@ size_t ZSTD_seekable_freeCStream(ZSTD_seekable_CStream* zcs)
{
if (zcs == NULL) return 0; /* support free on null */
ZSTD_freeCStream(zcs->cstream);
free(zcs->framelog.entries);
ZSTD_seekable_frameLog_freeVec(&zcs->framelog);
free(zcs);
return 0;
@ -115,44 +154,43 @@ size_t ZSTD_seekable_initCStream(ZSTD_seekable_CStream* zcs,
? maxFrameSize
: ZSTD_SEEKABLE_MAX_FRAME_DECOMPRESSED_SIZE;
zcs->checksumFlag = checksumFlag;
if (zcs->checksumFlag) {
zcs->framelog.checksumFlag = checksumFlag;
if (zcs->framelog.checksumFlag) {
XXH64_reset(&zcs->xxhState, 0);
}
zcs->seekTablePos = 0;
zcs->seekTableIndex = 0;
zcs->framelog.seekTablePos = 0;
zcs->framelog.seekTableIndex = 0;
zcs->writingSeekTable = 0;
return ZSTD_initCStream(zcs->cstream, compressionLevel);
}
static size_t ZSTD_seekable_logFrame(ZSTD_seekable_CStream* zcs)
size_t ZSTD_seekable_logFrame(ZSTD_frameLog* fl,
unsigned compressedSize,
unsigned decompressedSize,
unsigned checksum)
{
if (zcs->framelog.size == ZSTD_SEEKABLE_MAXFRAMES)
if (fl->size == ZSTD_SEEKABLE_MAXFRAMES)
return ERROR(frameIndex_tooLarge);
/* grow the buffer if required */
if (zcs->framelog.size == zcs->framelog.capacity) {
if (fl->size == fl->capacity) {
/* exponential size increase for constant amortized runtime */
size_t const newCapacity = zcs->framelog.capacity * 2;
framelogEntry_t* const newEntries = realloc(zcs->framelog.entries,
size_t const newCapacity = fl->capacity * 2;
framelogEntry_t* const newEntries = realloc(fl->entries,
sizeof(framelogEntry_t) * newCapacity);
if (newEntries == NULL) return ERROR(memory_allocation);
zcs->framelog.entries = newEntries;
zcs->framelog.capacity = newCapacity;
fl->entries = newEntries;
fl->capacity = newCapacity;
}
zcs->framelog.entries[zcs->framelog.size] = (framelogEntry_t){
zcs->frameCSize, zcs->frameDSize,
fl->entries[fl->size] = (framelogEntry_t){
compressedSize, decompressedSize, checksum
};
if (zcs->checksumFlag)
zcs->framelog.entries[zcs->framelog.size].checksum =
XXH64_digest(&zcs->xxhState) & 0xFFFFFFFFU; /* take lower 32 bits of digest */
zcs->framelog.size++;
fl->size++;
return 0;
}
@ -171,7 +209,11 @@ size_t ZSTD_seekable_endFrame(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output
/* frame done */
/* store the frame data for later */
ret = ZSTD_seekable_logFrame(zcs);
ret = ZSTD_seekable_logFrame(
&zcs->framelog, zcs->frameCSize, zcs->frameDSize,
zcs->framelog.checksumFlag
? XXH64_digest(&zcs->xxhState) & 0xFFFFFFFFU
: 0);
if (ret) return ret;
/* reset for the next frame */
@ -179,7 +221,7 @@ size_t ZSTD_seekable_endFrame(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output
zcs->frameDSize = 0;
ZSTD_resetCStream(zcs->cstream, 0);
if (zcs->checksumFlag)
if (zcs->framelog.checksumFlag)
XXH64_reset(&zcs->xxhState, 0);
return 0;
@ -199,7 +241,7 @@ size_t ZSTD_seekable_compressStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer*
size_t const ret = ZSTD_compressStream(zcs->cstream, output, &inTmp);
if (zcs->checksumFlag) {
if (zcs->framelog.checksumFlag) {
XXH64_update(&zcs->xxhState, inBase, inTmp.pos);
}
@ -223,36 +265,36 @@ size_t ZSTD_seekable_compressStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer*
return (size_t)(zcs->maxFrameSize - zcs->frameDSize);
}
static inline size_t ZSTD_seekable_seekTableSize(ZSTD_seekable_CStream const* zcs)
static inline size_t ZSTD_seekable_seekTableSize(const ZSTD_frameLog* fl)
{
size_t const sizePerFrame = 8 + (zcs->checksumFlag?4:0);
size_t const sizePerFrame = 8 + (fl->checksumFlag?4:0);
size_t const seekTableLen = ZSTD_skippableHeaderSize +
sizePerFrame * zcs->framelog.size +
sizePerFrame * fl->size +
ZSTD_seekTableFooterSize;
return seekTableLen;
}
static inline size_t ZSTD_stwrite32(ZSTD_seekable_CStream* zcs,
static inline size_t ZSTD_stwrite32(ZSTD_frameLog* fl,
ZSTD_outBuffer* output, U32 const value,
U32 const offset)
{
if (zcs->seekTablePos < offset + 4) {
if (fl->seekTablePos < offset + 4) {
BYTE tmp[4]; /* so that we can work with buffers too small to write a whole word to */
size_t const lenWrite =
MIN(output->size - output->pos, offset + 4 - zcs->seekTablePos);
MIN(output->size - output->pos, offset + 4 - fl->seekTablePos);
MEM_writeLE32(tmp, value);
memcpy((BYTE*)output->dst + output->pos,
tmp + (zcs->seekTablePos - offset), lenWrite);
tmp + (fl->seekTablePos - offset), lenWrite);
output->pos += lenWrite;
zcs->seekTablePos += lenWrite;
fl->seekTablePos += lenWrite;
if (lenWrite < 4) return ZSTD_seekable_seekTableSize(zcs) - zcs->seekTablePos;
if (lenWrite < 4) return ZSTD_seekable_seekTableSize(fl) - fl->seekTablePos;
}
return 0;
}
static size_t ZSTD_seekable_writeSeekTable(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* output)
size_t ZSTD_seekable_writeSeekTable(ZSTD_frameLog* fl, ZSTD_outBuffer* output)
{
/* seekTableIndex: the current index in the table and
* seekTableSize: the amount of the table written so far
@ -261,50 +303,51 @@ static size_t ZSTD_seekable_writeSeekTable(ZSTD_seekable_CStream* zcs, ZSTD_outB
* because of a small buffer, it can keep going where it left off.
*/
size_t const sizePerFrame = 8 + (zcs->checksumFlag?4:0);
size_t const seekTableLen = ZSTD_seekable_seekTableSize(zcs);
size_t const sizePerFrame = 8 + (fl->checksumFlag?4:0);
size_t const seekTableLen = ZSTD_seekable_seekTableSize(fl);
CHECK_Z(ZSTD_stwrite32(zcs, output, ZSTD_MAGIC_SKIPPABLE_START | 0xE, 0));
CHECK_Z(ZSTD_stwrite32(zcs, output, seekTableLen - ZSTD_skippableHeaderSize,
CHECK_Z(ZSTD_stwrite32(fl, output, ZSTD_MAGIC_SKIPPABLE_START | 0xE, 0));
CHECK_Z(ZSTD_stwrite32(fl, output, seekTableLen - ZSTD_skippableHeaderSize,
4));
while (zcs->seekTableIndex < zcs->framelog.size) {
CHECK_Z(ZSTD_stwrite32(
zcs, output, zcs->framelog.entries[zcs->seekTableIndex].cSize,
ZSTD_skippableHeaderSize + sizePerFrame * zcs->seekTableIndex));
while (fl->seekTableIndex < fl->size) {
CHECK_Z(ZSTD_stwrite32(fl, output,
fl->entries[fl->seekTableIndex].cSize,
ZSTD_skippableHeaderSize +
sizePerFrame * fl->seekTableIndex + 0));
CHECK_Z(ZSTD_stwrite32(
zcs, output, zcs->framelog.entries[zcs->seekTableIndex].dSize,
ZSTD_skippableHeaderSize + sizePerFrame * zcs->seekTableIndex + 4));
CHECK_Z(ZSTD_stwrite32(fl, output,
fl->entries[fl->seekTableIndex].dSize,
ZSTD_skippableHeaderSize +
sizePerFrame * fl->seekTableIndex + 4));
if (zcs->checksumFlag) {
if (fl->checksumFlag) {
CHECK_Z(ZSTD_stwrite32(
zcs, output,
zcs->framelog.entries[zcs->seekTableIndex].checksum,
ZSTD_skippableHeaderSize + sizePerFrame * zcs->seekTableIndex +
8));
fl, output, fl->entries[fl->seekTableIndex].checksum,
ZSTD_skippableHeaderSize +
sizePerFrame * fl->seekTableIndex + 8));
}
zcs->seekTableIndex++;
fl->seekTableIndex++;
}
CHECK_Z(ZSTD_stwrite32(zcs, output, zcs->framelog.size,
CHECK_Z(ZSTD_stwrite32(fl, output, fl->size,
seekTableLen - ZSTD_seekTableFooterSize));
if (output->size - output->pos < 1) return seekTableLen - zcs->seekTablePos;
if (zcs->seekTablePos < seekTableLen - 4) {
if (output->size - output->pos < 1) return seekTableLen - fl->seekTablePos;
if (fl->seekTablePos < seekTableLen - 4) {
BYTE sfd = 0;
sfd |= (zcs->checksumFlag) << 7;
sfd |= (fl->checksumFlag) << 7;
((BYTE*)output->dst)[output->pos] = sfd;
output->pos++;
zcs->seekTablePos++;
fl->seekTablePos++;
}
CHECK_Z(ZSTD_stwrite32(zcs, output, ZSTD_SEEKABLE_MAGICNUMBER,
CHECK_Z(ZSTD_stwrite32(fl, output, ZSTD_SEEKABLE_MAGICNUMBER,
seekTableLen - 4));
if (zcs->seekTablePos != seekTableLen) return ERROR(GENERIC);
if (fl->seekTablePos != seekTableLen) return ERROR(GENERIC);
return 0;
}
@ -314,10 +357,10 @@ size_t ZSTD_seekable_endStream(ZSTD_seekable_CStream* zcs, ZSTD_outBuffer* outpu
const size_t endFrame = ZSTD_seekable_endFrame(zcs, output);
if (ZSTD_isError(endFrame)) return endFrame;
/* return an accurate size hint */
if (endFrame) return endFrame + ZSTD_seekable_seekTableSize(zcs);
if (endFrame) return endFrame + ZSTD_seekable_seekTableSize(&zcs->framelog);
}
zcs->writingSeekTable = 1;
return ZSTD_seekable_writeSeekTable(zcs, output);
return ZSTD_seekable_writeSeekTable(&zcs->framelog, output);
}