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

Merge pull request #756 from facebook/memOpt

Browse Source 
Memory optimisations for ZSTDMT
This commit is contained in:
Yann Collet 2017-07-12 18:32:37 -07:00 committed by GitHub
commit d6799635a5
10 changed files with 426 additions and 212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
NEWS
View File

@ -1,6 +1,7 @@
v1.3.1 v1.3.1
perf: substantially decreased memory usage in Multi-threading mode, thanks to reports by Tino Reichardt
build: fix Visual compilation for non x86/x64 targets, reported by Greg Slazinski (#718) build: fix Visual compilation for non x86/x64 targets, reported by Greg Slazinski (#718)
API exp : breaking change : ZSTD_getframeHeader() API exp : breaking change : ZSTD_getframeHeader() provides more information
v1.3.0 v1.3.0
cli : new : `--list` command, by Paul Cruz cli : new : `--list` command, by Paul Cruz

20
lib/common/pool.c
View File

@ -92,7 +92,7 @@ POOL_ctx *POOL_create(size_t numThreads, size_t queueSize) {
* and full queues. * and full queues.
*/ */
ctx->queueSize = queueSize + 1; ctx->queueSize = queueSize + 1;
ctx->queue = (POOL_job *)malloc(ctx->queueSize * sizeof(POOL_job)); ctx->queue = (POOL_job*) malloc(ctx->queueSize * sizeof(POOL_job));
ctx->queueHead = 0; ctx->queueHead = 0;
ctx->queueTail = 0; ctx->queueTail = 0;
pthread_mutex_init(&ctx->queueMutex, NULL); pthread_mutex_init(&ctx->queueMutex, NULL);
@ -100,7 +100,7 @@ POOL_ctx *POOL_create(size_t numThreads, size_t queueSize) {
pthread_cond_init(&ctx->queuePopCond, NULL); pthread_cond_init(&ctx->queuePopCond, NULL);
ctx->shutdown = 0; ctx->shutdown = 0;
/* Allocate space for the thread handles */ /* Allocate space for the thread handles */
ctx->threads = (pthread_t *)malloc(numThreads * sizeof(pthread_t)); ctx->threads = (pthread_t*)malloc(numThreads * sizeof(pthread_t));
ctx->numThreads = 0; ctx->numThreads = 0;
/* Check for errors */ /* Check for errors */
if (!ctx->threads || !ctx->queue) { POOL_free(ctx); return NULL; } if (!ctx->threads || !ctx->queue) { POOL_free(ctx); return NULL; }
@ -153,8 +153,8 @@ size_t POOL_sizeof(POOL_ctx *ctx) {
+ ctx->numThreads * sizeof(pthread_t); + ctx->numThreads * sizeof(pthread_t);
} }
void POOL_add(void *ctxVoid, POOL_function function, void *opaque) { void POOL_add(void* ctxVoid, POOL_function function, void *opaque) {
POOL_ctx *ctx = (POOL_ctx *)ctxVoid; POOL_ctx* const ctx = (POOL_ctx*)ctxVoid;
if (!ctx) { return; } if (!ctx) { return; }
pthread_mutex_lock(&ctx->queueMutex); pthread_mutex_lock(&ctx->queueMutex);
@ -183,22 +183,22 @@ struct POOL_ctx_s {
int data; int data;
}; };
POOL_ctx *POOL_create(size_t numThreads, size_t queueSize) { POOL_ctx* POOL_create(size_t numThreads, size_t queueSize) {
(void)numThreads; (void)numThreads;
(void)queueSize; (void)queueSize;
return (POOL_ctx *)malloc(sizeof(POOL_ctx)); return (POOL_ctx*)malloc(sizeof(POOL_ctx));
} }
void POOL_free(POOL_ctx *ctx) { void POOL_free(POOL_ctx* ctx) {
if (ctx) free(ctx); free(ctx);
} }
void POOL_add(void *ctx, POOL_function function, void *opaque) { void POOL_add(void* ctx, POOL_function function, void* opaque) {
(void)ctx; (void)ctx;
function(opaque); function(opaque);
} }
size_t POOL_sizeof(POOL_ctx *ctx) { size_t POOL_sizeof(POOL_ctx* ctx) {
if (ctx==NULL) return 0; /* supports sizeof NULL */ if (ctx==NULL) return 0; /* supports sizeof NULL */
return sizeof(*ctx); return sizeof(*ctx);
} }

10
lib/common/pool.h
View File

@ -19,11 +19,11 @@ extern "C" {
typedef struct POOL_ctx_s POOL_ctx; typedef struct POOL_ctx_s POOL_ctx;
/*! POOL_create() : /*! POOL_create() :
Create a thread pool with at most `numThreads` threads. * Create a thread pool with at most `numThreads` threads.
`numThreads` must be at least 1. * `numThreads` must be at least 1.
The maximum number of queued jobs before blocking is `queueSize`. * The maximum number of queued jobs before blocking is `queueSize`.
`queueSize` must be at least 1. * `queueSize` must be at least 1.
@return : The POOL_ctx pointer on success else NULL. * @return : POOL_ctx pointer on success, else NULL.
*/ */
POOL_ctx *POOL_create(size_t numThreads, size_t queueSize); POOL_ctx *POOL_create(size_t numThreads, size_t queueSize);

2
lib/compress/zstd_compress.c
View File

@ -377,7 +377,7 @@ size_t ZSTD_CCtx_setParameter(ZSTD_CCtx* cctx, ZSTD_cParameter param, unsigned v
return ERROR(compressionParameter_unsupported); return ERROR(compressionParameter_unsupported);
ZSTDMT_freeCCtx(cctx->mtctx); ZSTDMT_freeCCtx(cctx->mtctx);
cctx->nbThreads = 1; cctx->nbThreads = 1;
cctx->mtctx = ZSTDMT_createCCtx(value); cctx->mtctx = ZSTDMT_createCCtx_advanced(value, cctx->customMem);
if (cctx->mtctx == NULL) return ERROR(memory_allocation); if (cctx->mtctx == NULL) return ERROR(memory_allocation);
} }
cctx->nbThreads = value; cctx->nbThreads = value;

320
lib/compress/zstdmt_compress.c
View File

@ -73,6 +73,7 @@ static unsigned long long GetCurrentClockTimeMicroseconds(void)
/* ===== Buffer Pool ===== */ /* ===== Buffer Pool ===== */
/* a single Buffer Pool can be invoked from multiple threads in parallel */
typedef struct buffer_s { typedef struct buffer_s {
void* start; void* start;
@ -82,6 +83,8 @@ typedef struct buffer_s {
static const buffer_t g_nullBuffer = { NULL, 0 }; static const buffer_t g_nullBuffer = { NULL, 0 };
typedef struct ZSTDMT_bufferPool_s { typedef struct ZSTDMT_bufferPool_s {
pthread_mutex_t poolMutex;
size_t bufferSize;
unsigned totalBuffers; unsigned totalBuffers;
unsigned nbBuffers; unsigned nbBuffers;
ZSTD_customMem cMem; ZSTD_customMem cMem;
@ -90,10 +93,12 @@ typedef struct ZSTDMT_bufferPool_s {
static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_customMem cMem) static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_customMem cMem)
{ {
unsigned const maxNbBuffers = 2*nbThreads + 2; unsigned const maxNbBuffers = 2*nbThreads + 3;
ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc( ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc(
sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem); sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem);
if (bufPool==NULL) return NULL; if (bufPool==NULL) return NULL;
pthread_mutex_init(&bufPool->poolMutex, NULL);
bufPool->bufferSize = 64 KB;
bufPool->totalBuffers = maxNbBuffers; bufPool->totalBuffers = maxNbBuffers;
bufPool->nbBuffers = 0; bufPool->nbBuffers = 0;
bufPool->cMem = cMem; bufPool->cMem = cMem;
@ -106,6 +111,7 @@ static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool)
if (!bufPool) return; /* compatibility with free on NULL */ if (!bufPool) return; /* compatibility with free on NULL */
for (u=0; u<bufPool->totalBuffers; u++) for (u=0; u<bufPool->totalBuffers; u++)
ZSTD_free(bufPool->bTable[u].start, bufPool->cMem); ZSTD_free(bufPool->bTable[u].start, bufPool->cMem);
pthread_mutex_destroy(&bufPool->poolMutex);
ZSTD_free(bufPool, bufPool->cMem); ZSTD_free(bufPool, bufPool->cMem);
} }
@ -116,65 +122,85 @@ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)
+ (bufPool->totalBuffers - 1) * sizeof(buffer_t); + (bufPool->totalBuffers - 1) * sizeof(buffer_t);
unsigned u; unsigned u;
size_t totalBufferSize = 0; size_t totalBufferSize = 0;
pthread_mutex_lock(&bufPool->poolMutex);
for (u=0; u<bufPool->totalBuffers; u++) for (u=0; u<bufPool->totalBuffers; u++)
totalBufferSize += bufPool->bTable[u].size; totalBufferSize += bufPool->bTable[u].size;
pthread_mutex_unlock(&bufPool->poolMutex);
return poolSize + totalBufferSize; return poolSize + totalBufferSize;
} }
/** ZSTDMT_getBuffer() : static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* bufPool, size_t bSize)
* assumption : invocation from main thread only ! */
static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* pool, size_t bSize)
{ {
if (pool->nbBuffers) { /* try to use an existing buffer */ bufPool->bufferSize = bSize;
buffer_t const buf = pool->bTable[--(pool->nbBuffers)]; }
/** ZSTDMT_getBuffer() :
* assumption : bufPool must be valid */
static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
{
size_t const bSize = bufPool->bufferSize;
DEBUGLOG(5, "ZSTDMT_getBuffer");
pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers) { /* try to use an existing buffer */
buffer_t const buf = bufPool->bTable[--(bufPool->nbBuffers)];
size_t const availBufferSize = buf.size; size_t const availBufferSize = buf.size;
if ((availBufferSize >= bSize) & (availBufferSize <= 10*bSize)) if ((availBufferSize >= bSize) & (availBufferSize <= 10*bSize)) {
/* large enough, but not too much */ /* large enough, but not too much */
pthread_mutex_unlock(&bufPool->poolMutex);
return buf; return buf;
}
/* size conditions not respected : scratch this buffer, create new one */ /* size conditions not respected : scratch this buffer, create new one */
ZSTD_free(buf.start, pool->cMem); DEBUGLOG(5, "existing buffer does not meet size conditions => freeing");
ZSTD_free(buf.start, bufPool->cMem);
} }
pthread_mutex_unlock(&bufPool->poolMutex);
/* create new buffer */ /* create new buffer */
DEBUGLOG(5, "create a new buffer");
{ buffer_t buffer; { buffer_t buffer;
void* const start = ZSTD_malloc(bSize, pool->cMem); void* const start = ZSTD_malloc(bSize, bufPool->cMem);
if (start==NULL) bSize = 0;
buffer.start = start; /* note : start can be NULL if malloc fails ! */ buffer.start = start; /* note : start can be NULL if malloc fails ! */
buffer.size = bSize; buffer.size = (start==NULL) ? 0 : bSize;
return buffer; return buffer;
} }
} }
/* store buffer for later re-use, up to pool capacity */ /* store buffer for later re-use, up to pool capacity */
static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* pool, buffer_t buf) static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf)
{ {
if (buf.start == NULL) return; /* release on NULL */ if (buf.start == NULL) return; /* compatible with release on NULL */
if (pool->nbBuffers < pool->totalBuffers) { DEBUGLOG(5, "ZSTDMT_releaseBuffer");
pool->bTable[pool->nbBuffers++] = buf; /* store for later re-use */ pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers < bufPool->totalBuffers) {
bufPool->bTable[bufPool->nbBuffers++] = buf; /* stored for later use */
pthread_mutex_unlock(&bufPool->poolMutex);
return; return;
} }
pthread_mutex_unlock(&bufPool->poolMutex);
/* Reached bufferPool capacity (should not happen) */ /* Reached bufferPool capacity (should not happen) */
ZSTD_free(buf.start, pool->cMem); DEBUGLOG(5, "buffer pool capacity reached => freeing ");
ZSTD_free(buf.start, bufPool->cMem);
} }
/* ===== CCtx Pool ===== */ /* ===== CCtx Pool ===== */
/* a single CCtx Pool can be invoked from multiple threads in parallel */
typedef struct { typedef struct {
pthread_mutex_t poolMutex;
unsigned totalCCtx; unsigned totalCCtx;
unsigned availCCtx; unsigned availCCtx;
ZSTD_customMem cMem; ZSTD_customMem cMem;
ZSTD_CCtx* cctx[1]; /* variable size */ ZSTD_CCtx* cctx[1]; /* variable size */
} ZSTDMT_CCtxPool; } ZSTDMT_CCtxPool;
/* assumption : CCtxPool invocation only from main thread */
/* note : all CCtx borrowed from the pool should be released back to the pool _before_ freeing the pool */ /* note : all CCtx borrowed from the pool should be released back to the pool _before_ freeing the pool */
static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool) static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool)
{ {
unsigned u; unsigned u;
for (u=0; u<pool->totalCCtx; u++) for (u=0; u<pool->totalCCtx; u++)
ZSTD_freeCCtx(pool->cctx[u]); /* note : compatible with free on NULL */ ZSTD_freeCCtx(pool->cctx[u]); /* note : compatible with free on NULL */
pthread_mutex_destroy(&pool->poolMutex);
ZSTD_free(pool, pool->cMem); ZSTD_free(pool, pool->cMem);
} }
@ -186,6 +212,7 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads,
ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_calloc( ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_calloc(
sizeof(ZSTDMT_CCtxPool) + (nbThreads-1)*sizeof(ZSTD_CCtx*), cMem); sizeof(ZSTDMT_CCtxPool) + (nbThreads-1)*sizeof(ZSTD_CCtx*), cMem);
if (!cctxPool) return NULL; if (!cctxPool) return NULL;
pthread_mutex_init(&cctxPool->poolMutex, NULL);
cctxPool->cMem = cMem; cctxPool->cMem = cMem;
cctxPool->totalCCtx = nbThreads; cctxPool->totalCCtx = nbThreads;
cctxPool->availCCtx = 1; /* at least one cctx for single-thread mode */ cctxPool->availCCtx = 1; /* at least one cctx for single-thread mode */
@ -198,50 +225,57 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads,
/* only works during initialization phase, not during compression */ /* only works during initialization phase, not during compression */
static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool) static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool)
{ {
unsigned const nbThreads = cctxPool->totalCCtx; pthread_mutex_lock(&cctxPool->poolMutex);
size_t const poolSize = sizeof(*cctxPool) { unsigned const nbThreads = cctxPool->totalCCtx;
+ (nbThreads-1)*sizeof(ZSTD_CCtx*); size_t const poolSize = sizeof(*cctxPool)
unsigned u; + (nbThreads-1)*sizeof(ZSTD_CCtx*);
size_t totalCCtxSize = 0; unsigned u;
for (u=0; u<nbThreads; u++) size_t totalCCtxSize = 0;
totalCCtxSize += ZSTD_sizeof_CCtx(cctxPool->cctx[u]); for (u=0; u<nbThreads; u++) {
totalCCtxSize += ZSTD_sizeof_CCtx(cctxPool->cctx[u]);
return poolSize + totalCCtxSize; }
pthread_mutex_unlock(&cctxPool->poolMutex);
return poolSize + totalCCtxSize;
}
} }
static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* pool) static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool)
{ {
if (pool->availCCtx) { DEBUGLOG(5, "ZSTDMT_getCCtx");
pool->availCCtx--; pthread_mutex_lock(&cctxPool->poolMutex);
return pool->cctx[pool->availCCtx]; if (cctxPool->availCCtx) {
} cctxPool->availCCtx--;
return ZSTD_createCCtx(); /* note : can be NULL, when creation fails ! */ { ZSTD_CCtx* const cctx = cctxPool->cctx[cctxPool->availCCtx];
pthread_mutex_unlock(&cctxPool->poolMutex);
return cctx;
} }
pthread_mutex_unlock(&cctxPool->poolMutex);
DEBUGLOG(5, "create one more CCtx");
return ZSTD_createCCtx_advanced(cctxPool->cMem); /* note : can be NULL, when creation fails ! */
} }
static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx) static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
{ {
if (cctx==NULL) return; /* compatibility with release on NULL */ if (cctx==NULL) return; /* compatibility with release on NULL */
pthread_mutex_lock(&pool->poolMutex);
if (pool->availCCtx < pool->totalCCtx) if (pool->availCCtx < pool->totalCCtx)
pool->cctx[pool->availCCtx++] = cctx; pool->cctx[pool->availCCtx++] = cctx;
else else {
/* pool overflow : should not happen, since totalCCtx==nbThreads */ /* pool overflow : should not happen, since totalCCtx==nbThreads */
DEBUGLOG(5, "CCtx pool overflow : free cctx");
ZSTD_freeCCtx(cctx); ZSTD_freeCCtx(cctx);
}
pthread_mutex_unlock(&pool->poolMutex);
} }
/* ===== Thread worker ===== */ /* ===== Thread worker ===== */
typedef struct { typedef struct {
buffer_t buffer;
size_t filled;
} inBuff_t;
typedef struct {
ZSTD_CCtx* cctx;
buffer_t src; buffer_t src;
const void* srcStart; const void* srcStart;
size_t srcSize;
size_t dictSize; size_t dictSize;
size_t srcSize;
buffer_t dstBuff; buffer_t dstBuff;
size_t cSize; size_t cSize;
size_t dstFlushed; size_t dstFlushed;
@ -253,6 +287,8 @@ typedef struct {
pthread_cond_t* jobCompleted_cond; pthread_cond_t* jobCompleted_cond;
ZSTD_parameters params; ZSTD_parameters params;
const ZSTD_CDict* cdict; const ZSTD_CDict* cdict;
ZSTDMT_CCtxPool* cctxPool;
ZSTDMT_bufferPool* bufPool;
unsigned long long fullFrameSize; unsigned long long fullFrameSize;
} ZSTDMT_jobDescription; } ZSTDMT_jobDescription;
@ -260,37 +296,56 @@ typedef struct {
void ZSTDMT_compressChunk(void* jobDescription) void ZSTDMT_compressChunk(void* jobDescription)
{ {
ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription;
ZSTD_CCtx* cctx = ZSTDMT_getCCtx(job->cctxPool);
const void* const src = (const char*)job->srcStart + job->dictSize; const void* const src = (const char*)job->srcStart + job->dictSize;
buffer_t const dstBuff = job->dstBuff; buffer_t dstBuff = job->dstBuff;
DEBUGLOG(5, "job (first:%u) (last:%u) : dictSize %u, srcSize %u", DEBUGLOG(5, "job (first:%u) (last:%u) : dictSize %u, srcSize %u",
job->firstChunk, job->lastChunk, (U32)job->dictSize, (U32)job->srcSize); job->firstChunk, job->lastChunk, (U32)job->dictSize, (U32)job->srcSize);
if (cctx==NULL) {
job->cSize = ERROR(memory_allocation);
goto _endJob;
}
if (dstBuff.start == NULL) {
dstBuff = ZSTDMT_getBuffer(job->bufPool);
if (dstBuff.start==NULL) {
job->cSize = ERROR(memory_allocation);
goto _endJob;
}
job->dstBuff = dstBuff;
}
if (job->cdict) { /* should only happen for first segment */ if (job->cdict) { /* should only happen for first segment */
size_t const initError = ZSTD_compressBegin_usingCDict_advanced(job->cctx, job->cdict, job->params.fParams, job->fullFrameSize); size_t const initError = ZSTD_compressBegin_usingCDict_advanced(cctx, job->cdict, job->params.fParams, job->fullFrameSize);
DEBUGLOG(5, "using CDict"); DEBUGLOG(5, "using CDict");
if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; } if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; }
} else { /* srcStart points at reloaded section */ } else { /* srcStart points at reloaded section */
if (!job->firstChunk) job->params.fParams.contentSizeFlag = 0; /* ensure no srcSize control */ if (!job->firstChunk) job->params.fParams.contentSizeFlag = 0; /* ensure no srcSize control */
{ size_t const dictModeError = ZSTD_setCCtxParameter(job->cctx, ZSTD_p_forceRawDict, 1); /* Force loading dictionary in "content-only" mode (no header analysis) */ { size_t const dictModeError = ZSTD_setCCtxParameter(cctx, ZSTD_p_forceRawDict, 1); /* Force loading dictionary in "content-only" mode (no header analysis) */
size_t const initError = ZSTD_compressBegin_advanced(job->cctx, job->srcStart, job->dictSize, job->params, job->fullFrameSize); size_t const initError = ZSTD_compressBegin_advanced(cctx, job->srcStart, job->dictSize, job->params, job->fullFrameSize);
if (ZSTD_isError(initError) || ZSTD_isError(dictModeError)) { job->cSize = initError; goto _endJob; } if (ZSTD_isError(initError) || ZSTD_isError(dictModeError)) { job->cSize = initError; goto _endJob; }
ZSTD_setCCtxParameter(job->cctx, ZSTD_p_forceWindow, 1); ZSTD_setCCtxParameter(cctx, ZSTD_p_forceWindow, 1);
} } } }
if (!job->firstChunk) { /* flush and overwrite frame header when it's not first segment */ if (!job->firstChunk) { /* flush and overwrite frame header when it's not first segment */
size_t const hSize = ZSTD_compressContinue(job->cctx, dstBuff.start, dstBuff.size, src, 0); size_t const hSize = ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, 0);
if (ZSTD_isError(hSize)) { job->cSize = hSize; goto _endJob; } if (ZSTD_isError(hSize)) { job->cSize = hSize; goto _endJob; }
ZSTD_invalidateRepCodes(job->cctx); ZSTD_invalidateRepCodes(cctx);
} }
DEBUGLOG(5, "Compressing : "); DEBUGLOG(5, "Compressing : ");
DEBUG_PRINTHEX(4, job->srcStart, 12); DEBUG_PRINTHEX(4, job->srcStart, 12);
job->cSize = (job->lastChunk) ? job->cSize = (job->lastChunk) ?
ZSTD_compressEnd (job->cctx, dstBuff.start, dstBuff.size, src, job->srcSize) : ZSTD_compressEnd (cctx, dstBuff.start, dstBuff.size, src, job->srcSize) :
ZSTD_compressContinue(job->cctx, dstBuff.start, dstBuff.size, src, job->srcSize); ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, job->srcSize);
DEBUGLOG(5, "compressed %u bytes into %u bytes (first:%u) (last:%u)", DEBUGLOG(5, "compressed %u bytes into %u bytes (first:%u) (last:%u)",
(unsigned)job->srcSize, (unsigned)job->cSize, job->firstChunk, job->lastChunk); (unsigned)job->srcSize, (unsigned)job->cSize, job->firstChunk, job->lastChunk);
DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize)); DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize));
_endJob: _endJob:
ZSTDMT_releaseCCtx(job->cctxPool, cctx);
ZSTDMT_releaseBuffer(job->bufPool, job->src);
job->src = g_nullBuffer; job->srcStart = NULL;
PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex); PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex);
job->jobCompleted = 1; job->jobCompleted = 1;
job->jobScanned = 0; job->jobScanned = 0;
@ -303,15 +358,19 @@ _endJob:
/* ===== Multi-threaded compression ===== */ /* ===== Multi-threaded compression ===== */
/* ------------------------------------------ */ /* ------------------------------------------ */
typedef struct {
buffer_t buffer;
size_t filled;
} inBuff_t;
struct ZSTDMT_CCtx_s { struct ZSTDMT_CCtx_s {
POOL_ctx* factory; POOL_ctx* factory;
ZSTDMT_jobDescription* jobs; ZSTDMT_jobDescription* jobs;
ZSTDMT_bufferPool* buffPool; ZSTDMT_bufferPool* bufPool;
ZSTDMT_CCtxPool* cctxPool; ZSTDMT_CCtxPool* cctxPool;
pthread_mutex_t jobCompleted_mutex; pthread_mutex_t jobCompleted_mutex;
pthread_cond_t jobCompleted_cond; pthread_cond_t jobCompleted_cond;
size_t targetSectionSize; size_t targetSectionSize;
size_t marginSize;
size_t inBuffSize; size_t inBuffSize;
size_t dictSize; size_t dictSize;
size_t targetDictSize; size_t targetDictSize;
@ -362,9 +421,9 @@ ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem)
mtctx->factory = POOL_create(nbThreads, 1); mtctx->factory = POOL_create(nbThreads, 1);
mtctx->jobs = ZSTDMT_allocJobsTable(&nbJobs, cMem); mtctx->jobs = ZSTDMT_allocJobsTable(&nbJobs, cMem);
mtctx->jobIDMask = nbJobs - 1; mtctx->jobIDMask = nbJobs - 1;
mtctx->buffPool = ZSTDMT_createBufferPool(nbThreads, cMem); mtctx->bufPool = ZSTDMT_createBufferPool(nbThreads, cMem);
mtctx->cctxPool = ZSTDMT_createCCtxPool(nbThreads, cMem); mtctx->cctxPool = ZSTDMT_createCCtxPool(nbThreads, cMem);
if (!mtctx->factory | !mtctx->jobs | !mtctx->buffPool | !mtctx->cctxPool) { if (!mtctx->factory | !mtctx->jobs | !mtctx->bufPool | !mtctx->cctxPool) {
ZSTDMT_freeCCtx(mtctx); ZSTDMT_freeCCtx(mtctx);
return NULL; return NULL;
} }
@ -386,15 +445,13 @@ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx)
unsigned jobID; unsigned jobID;
DEBUGLOG(3, "ZSTDMT_releaseAllJobResources"); DEBUGLOG(3, "ZSTDMT_releaseAllJobResources");
for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) { for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) {
ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[jobID].dstBuff); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff);
mtctx->jobs[jobID].dstBuff = g_nullBuffer; mtctx->jobs[jobID].dstBuff = g_nullBuffer;
ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[jobID].src); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].src);
mtctx->jobs[jobID].src = g_nullBuffer; mtctx->jobs[jobID].src = g_nullBuffer;
ZSTDMT_releaseCCtx(mtctx->cctxPool, mtctx->jobs[jobID].cctx);
mtctx->jobs[jobID].cctx = NULL;
} }
memset(mtctx->jobs, 0, (mtctx->jobIDMask+1)*sizeof(ZSTDMT_jobDescription)); memset(mtctx->jobs, 0, (mtctx->jobIDMask+1)*sizeof(ZSTDMT_jobDescription));
ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->inBuff.buffer); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer);
mtctx->inBuff.buffer = g_nullBuffer; mtctx->inBuff.buffer = g_nullBuffer;
mtctx->allJobsCompleted = 1; mtctx->allJobsCompleted = 1;
} }
@ -404,7 +461,7 @@ size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx)
if (mtctx==NULL) return 0; /* compatible with free on NULL */ if (mtctx==NULL) return 0; /* compatible with free on NULL */
POOL_free(mtctx->factory); POOL_free(mtctx->factory);
if (!mtctx->allJobsCompleted) ZSTDMT_releaseAllJobResources(mtctx); /* stop workers first */ if (!mtctx->allJobsCompleted) ZSTDMT_releaseAllJobResources(mtctx); /* stop workers first */
ZSTDMT_freeBufferPool(mtctx->buffPool); /* release job resources into pools first */ ZSTDMT_freeBufferPool(mtctx->bufPool); /* release job resources into pools first */
ZSTD_free(mtctx->jobs, mtctx->cMem); ZSTD_free(mtctx->jobs, mtctx->cMem);
ZSTDMT_freeCCtxPool(mtctx->cctxPool); ZSTDMT_freeCCtxPool(mtctx->cctxPool);
ZSTD_freeCDict(mtctx->cdictLocal); ZSTD_freeCDict(mtctx->cdictLocal);
@ -418,11 +475,11 @@ size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx)
{ {
if (mtctx == NULL) return 0; /* supports sizeof NULL */ if (mtctx == NULL) return 0; /* supports sizeof NULL */
return sizeof(*mtctx) return sizeof(*mtctx)
+ POOL_sizeof(mtctx->factory) + POOL_sizeof(mtctx->factory)
+ ZSTDMT_sizeof_bufferPool(mtctx->buffPool) + ZSTDMT_sizeof_bufferPool(mtctx->bufPool)
+ (mtctx->jobIDMask+1) * sizeof(ZSTDMT_jobDescription) + (mtctx->jobIDMask+1) * sizeof(ZSTDMT_jobDescription)
+ ZSTDMT_sizeof_CCtxPool(mtctx->cctxPool) + ZSTDMT_sizeof_CCtxPool(mtctx->cctxPool)
+ ZSTD_sizeof_CDict(mtctx->cdictLocal); + ZSTD_sizeof_CDict(mtctx->cdictLocal);
} }
size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSDTMT_parameter parameter, unsigned value) size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSDTMT_parameter parameter, unsigned value)
@ -459,11 +516,11 @@ static unsigned computeNbChunks(size_t srcSize, unsigned windowLog, unsigned nbT
size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx, size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
void* dst, size_t dstCapacity, void* dst, size_t dstCapacity,
const void* src, size_t srcSize, const void* src, size_t srcSize,
const ZSTD_CDict* cdict, const ZSTD_CDict* cdict,
ZSTD_parameters const params, ZSTD_parameters const params,
unsigned overlapRLog) unsigned overlapRLog)
{ {
size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog); size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog);
unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, mtctx->nbThreads); unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, mtctx->nbThreads);
@ -473,6 +530,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
size_t remainingSrcSize = srcSize; size_t remainingSrcSize = srcSize;
unsigned const compressWithinDst = (dstCapacity >= ZSTD_compressBound(srcSize)) ? nbChunks : (unsigned)(dstCapacity / ZSTD_compressBound(avgChunkSize)); /* presumes avgChunkSize >= 256 KB, which should be the case */ unsigned const compressWithinDst = (dstCapacity >= ZSTD_compressBound(srcSize)) ? nbChunks : (unsigned)(dstCapacity / ZSTD_compressBound(avgChunkSize)); /* presumes avgChunkSize >= 256 KB, which should be the case */
size_t frameStartPos = 0, dstBufferPos = 0; size_t frameStartPos = 0, dstBufferPos = 0;
XXH64_state_t xxh64;
DEBUGLOG(4, "nbChunks : %2u (chunkSize : %u bytes) ", nbChunks, (U32)avgChunkSize); DEBUGLOG(4, "nbChunks : %2u (chunkSize : %u bytes) ", nbChunks, (U32)avgChunkSize);
if (nbChunks==1) { /* fallback to single-thread mode */ if (nbChunks==1) { /* fallback to single-thread mode */
@ -481,6 +539,8 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
return ZSTD_compress_advanced(cctx, dst, dstCapacity, src, srcSize, NULL, 0, params); return ZSTD_compress_advanced(cctx, dst, dstCapacity, src, srcSize, NULL, 0, params);
} }
assert(avgChunkSize >= 256 KB); /* condition for ZSTD_compressBound(A) + ZSTD_compressBound(B) <= ZSTD_compressBound(A+B), which is useful to avoid allocating extra buffers */ assert(avgChunkSize >= 256 KB); /* condition for ZSTD_compressBound(A) + ZSTD_compressBound(B) <= ZSTD_compressBound(A+B), which is useful to avoid allocating extra buffers */
ZSTDMT_setBufferSize(mtctx->bufPool, ZSTD_compressBound(avgChunkSize) );
XXH64_reset(&xxh64, 0);
if (nbChunks > mtctx->jobIDMask+1) { /* enlarge job table */ if (nbChunks > mtctx->jobIDMask+1) { /* enlarge job table */
U32 nbJobs = nbChunks; U32 nbJobs = nbChunks;
@ -496,17 +556,10 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
size_t const chunkSize = MIN(remainingSrcSize, avgChunkSize); size_t const chunkSize = MIN(remainingSrcSize, avgChunkSize);
size_t const dstBufferCapacity = ZSTD_compressBound(chunkSize); size_t const dstBufferCapacity = ZSTD_compressBound(chunkSize);
buffer_t const dstAsBuffer = { (char*)dst + dstBufferPos, dstBufferCapacity }; buffer_t const dstAsBuffer = { (char*)dst + dstBufferPos, dstBufferCapacity };
buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : ZSTDMT_getBuffer(mtctx->buffPool, dstBufferCapacity); buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : g_nullBuffer;
ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(mtctx->cctxPool);
size_t dictSize = u ? overlapSize : 0; size_t dictSize = u ? overlapSize : 0;
if ((cctx==NULL) || (dstBuffer.start==NULL)) { mtctx->jobs[u].src = g_nullBuffer;
mtctx->jobs[u].cSize = ERROR(memory_allocation); /* job result */
mtctx->jobs[u].jobCompleted = 1;
nbChunks = u+1; /* only wait and free u jobs, instead of initially expected nbChunks ones */
break; /* let's wait for previous jobs to complete, but don't start new ones */
}
mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize; mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize;
mtctx->jobs[u].dictSize = dictSize; mtctx->jobs[u].dictSize = dictSize;
mtctx->jobs[u].srcSize = chunkSize; mtctx->jobs[u].srcSize = chunkSize;
@ -516,13 +569,18 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
/* do not calculate checksum within sections, but write it in header for first section */ /* do not calculate checksum within sections, but write it in header for first section */
if (u!=0) mtctx->jobs[u].params.fParams.checksumFlag = 0; if (u!=0) mtctx->jobs[u].params.fParams.checksumFlag = 0;
mtctx->jobs[u].dstBuff = dstBuffer; mtctx->jobs[u].dstBuff = dstBuffer;
mtctx->jobs[u].cctx = cctx; mtctx->jobs[u].cctxPool = mtctx->cctxPool;
mtctx->jobs[u].bufPool = mtctx->bufPool;
mtctx->jobs[u].firstChunk = (u==0); mtctx->jobs[u].firstChunk = (u==0);
mtctx->jobs[u].lastChunk = (u==nbChunks-1); mtctx->jobs[u].lastChunk = (u==nbChunks-1);
mtctx->jobs[u].jobCompleted = 0; mtctx->jobs[u].jobCompleted = 0;
mtctx->jobs[u].jobCompleted_mutex = &mtctx->jobCompleted_mutex; mtctx->jobs[u].jobCompleted_mutex = &mtctx->jobCompleted_mutex;
mtctx->jobs[u].jobCompleted_cond = &mtctx->jobCompleted_cond; mtctx->jobs[u].jobCompleted_cond = &mtctx->jobCompleted_cond;
if (params.fParams.checksumFlag) {
XXH64_update(&xxh64, srcStart + frameStartPos, chunkSize);
}
DEBUGLOG(5, "posting job %u (%u bytes)", u, (U32)chunkSize); DEBUGLOG(5, "posting job %u (%u bytes)", u, (U32)chunkSize);
DEBUG_PRINTHEX(6, mtctx->jobs[u].srcStart, 12); DEBUG_PRINTHEX(6, mtctx->jobs[u].srcStart, 12);
POOL_add(mtctx->factory, ZSTDMT_compressChunk, &mtctx->jobs[u]); POOL_add(mtctx->factory, ZSTDMT_compressChunk, &mtctx->jobs[u]);
@ -533,8 +591,8 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
} } } }
/* collect result */ /* collect result */
{ unsigned chunkID; { size_t error = 0, dstPos = 0;
size_t error = 0, dstPos = 0; unsigned chunkID;
for (chunkID=0; chunkID<nbChunks; chunkID++) { for (chunkID=0; chunkID<nbChunks; chunkID++) {
DEBUGLOG(5, "waiting for chunk %u ", chunkID); DEBUGLOG(5, "waiting for chunk %u ", chunkID);
PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex); PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex);
@ -545,8 +603,6 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
pthread_mutex_unlock(&mtctx->jobCompleted_mutex); pthread_mutex_unlock(&mtctx->jobCompleted_mutex);
DEBUGLOG(5, "ready to write chunk %u ", chunkID); DEBUGLOG(5, "ready to write chunk %u ", chunkID);
ZSTDMT_releaseCCtx(mtctx->cctxPool, mtctx->jobs[chunkID].cctx);
mtctx->jobs[chunkID].cctx = NULL;
mtctx->jobs[chunkID].srcStart = NULL; mtctx->jobs[chunkID].srcStart = NULL;
{ size_t const cSize = mtctx->jobs[chunkID].cSize; { size_t const cSize = mtctx->jobs[chunkID].cSize;
if (ZSTD_isError(cSize)) error = cSize; if (ZSTD_isError(cSize)) error = cSize;
@ -556,13 +612,25 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
memmove((char*)dst + dstPos, mtctx->jobs[chunkID].dstBuff.start, cSize); /* may overlap when chunk compressed within dst */ memmove((char*)dst + dstPos, mtctx->jobs[chunkID].dstBuff.start, cSize); /* may overlap when chunk compressed within dst */
if (chunkID >= compressWithinDst) { /* chunk compressed into its own buffer, which must be released */ if (chunkID >= compressWithinDst) { /* chunk compressed into its own buffer, which must be released */
DEBUGLOG(5, "releasing buffer %u>=%u", chunkID, compressWithinDst); DEBUGLOG(5, "releasing buffer %u>=%u", chunkID, compressWithinDst);
ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->jobs[chunkID].dstBuff); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[chunkID].dstBuff);
} }
mtctx->jobs[chunkID].dstBuff = g_nullBuffer; mtctx->jobs[chunkID].dstBuff = g_nullBuffer;
} }
dstPos += cSize ; dstPos += cSize ;
} }
} } /* for (chunkID=0; chunkID<nbChunks; chunkID++) */
DEBUGLOG(4, "checksumFlag : %u ", params.fParams.checksumFlag);
if (params.fParams.checksumFlag) {
U32 const checksum = (U32)XXH64_digest(&xxh64);
if (dstPos + 4 > dstCapacity) {
error = ERROR(dstSize_tooSmall);
} else {
DEBUGLOG(4, "writing checksum : %08X \n", checksum);
MEM_writeLE32((char*)dst + dstPos, checksum);
dstPos += 4;
} }
if (!error) DEBUGLOG(4, "compressed size : %u ", (U32)dstPos); if (!error) DEBUGLOG(4, "compressed size : %u ", (U32)dstPos);
return error ? error : dstPos; return error ? error : dstPos;
} }
@ -615,8 +683,8 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs,
if (zcs->nbThreads==1) { if (zcs->nbThreads==1) {
DEBUGLOG(4, "single thread mode"); DEBUGLOG(4, "single thread mode");
return ZSTD_initCStream_internal(zcs->cctxPool->cctx[0], return ZSTD_initCStream_internal(zcs->cctxPool->cctx[0],
dict, dictSize, cdict, dict, dictSize, cdict,
params, pledgedSrcSize); params, pledgedSrcSize);
} }
if (zcs->allJobsCompleted == 0) { /* previous compression not correctly finished */ if (zcs->allJobsCompleted == 0) { /* previous compression not correctly finished */
@ -649,11 +717,9 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs,
zcs->targetSectionSize = MAX(ZSTDMT_SECTION_SIZE_MIN, zcs->targetSectionSize); zcs->targetSectionSize = MAX(ZSTDMT_SECTION_SIZE_MIN, zcs->targetSectionSize);
zcs->targetSectionSize = MAX(zcs->targetDictSize, zcs->targetSectionSize); zcs->targetSectionSize = MAX(zcs->targetDictSize, zcs->targetSectionSize);
DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10)); DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10));
zcs->marginSize = zcs->targetSectionSize >> 2; zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize;
zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize + zcs->marginSize; ZSTDMT_setBufferSize(zcs->bufPool, MAX(zcs->inBuffSize, ZSTD_compressBound(zcs->targetSectionSize)) );
zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->buffPool, zcs->inBuffSize); zcs->inBuff.buffer = g_nullBuffer;
if (zcs->inBuff.buffer.start == NULL) return ERROR(memory_allocation);
zcs->inBuff.filled = 0;
zcs->dictSize = 0; zcs->dictSize = 0;
zcs->doneJobID = 0; zcs->doneJobID = 0;
zcs->nextJobID = 0; zcs->nextJobID = 0;
@ -664,8 +730,9 @@ size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs,
} }
size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx, size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx,
const void* dict, size_t dictSize, const void* dict, size_t dictSize,
ZSTD_parameters params, unsigned long long pledgedSrcSize) ZSTD_parameters params,
unsigned long long pledgedSrcSize)
{ {
DEBUGLOG(5, "ZSTDMT_initCStream_advanced"); DEBUGLOG(5, "ZSTDMT_initCStream_advanced");
return ZSTDMT_initCStream_internal(mtctx, dict, dictSize, NULL, params, pledgedSrcSize); return ZSTDMT_initCStream_internal(mtctx, dict, dictSize, NULL, params, pledgedSrcSize);
@ -701,19 +768,8 @@ size_t ZSTDMT_initCStream(ZSTDMT_CCtx* zcs, int compressionLevel) {
static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsigned endFrame) static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsigned endFrame)
{ {
size_t const dstBufferCapacity = ZSTD_compressBound(srcSize);
buffer_t const dstBuffer = ZSTDMT_getBuffer(zcs->buffPool, dstBufferCapacity);
ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(zcs->cctxPool);
unsigned const jobID = zcs->nextJobID & zcs->jobIDMask; unsigned const jobID = zcs->nextJobID & zcs->jobIDMask;
if ((cctx==NULL) || (dstBuffer.start==NULL)) {
zcs->jobs[jobID].jobCompleted = 1;
zcs->nextJobID++;
ZSTDMT_waitForAllJobsCompleted(zcs);
ZSTDMT_releaseAllJobResources(zcs);
return ERROR(memory_allocation);
}
DEBUGLOG(4, "preparing job %u to compress %u bytes with %u preload ", DEBUGLOG(4, "preparing job %u to compress %u bytes with %u preload ",
zcs->nextJobID, (U32)srcSize, (U32)zcs->dictSize); zcs->nextJobID, (U32)srcSize, (U32)zcs->dictSize);
zcs->jobs[jobID].src = zcs->inBuff.buffer; zcs->jobs[jobID].src = zcs->inBuff.buffer;
@ -726,8 +782,9 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
if (zcs->nextJobID) zcs->jobs[jobID].params.fParams.checksumFlag = 0; if (zcs->nextJobID) zcs->jobs[jobID].params.fParams.checksumFlag = 0;
zcs->jobs[jobID].cdict = zcs->nextJobID==0 ? zcs->cdict : NULL; zcs->jobs[jobID].cdict = zcs->nextJobID==0 ? zcs->cdict : NULL;
zcs->jobs[jobID].fullFrameSize = zcs->frameContentSize; zcs->jobs[jobID].fullFrameSize = zcs->frameContentSize;
zcs->jobs[jobID].dstBuff = dstBuffer; zcs->jobs[jobID].dstBuff = g_nullBuffer;
zcs->jobs[jobID].cctx = cctx; zcs->jobs[jobID].cctxPool = zcs->cctxPool;
zcs->jobs[jobID].bufPool = zcs->bufPool;
zcs->jobs[jobID].firstChunk = (zcs->nextJobID==0); zcs->jobs[jobID].firstChunk = (zcs->nextJobID==0);
zcs->jobs[jobID].lastChunk = endFrame; zcs->jobs[jobID].lastChunk = endFrame;
zcs->jobs[jobID].jobCompleted = 0; zcs->jobs[jobID].jobCompleted = 0;
@ -735,11 +792,13 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
zcs->jobs[jobID].jobCompleted_mutex = &zcs->jobCompleted_mutex; zcs->jobs[jobID].jobCompleted_mutex = &zcs->jobCompleted_mutex;
zcs->jobs[jobID].jobCompleted_cond = &zcs->jobCompleted_cond; zcs->jobs[jobID].jobCompleted_cond = &zcs->jobCompleted_cond;
if (zcs->params.fParams.checksumFlag)
XXH64_update(&zcs->xxhState, (const char*)zcs->inBuff.buffer.start + zcs->dictSize, srcSize);
/* get a new buffer for next input */ /* get a new buffer for next input */
if (!endFrame) { if (!endFrame) {
size_t const newDictSize = MIN(srcSize + zcs->dictSize, zcs->targetDictSize); size_t const newDictSize = MIN(srcSize + zcs->dictSize, zcs->targetDictSize);
DEBUGLOG(5, "ZSTDMT_createCompressionJob::endFrame = %u", endFrame); zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->bufPool);
zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->buffPool, zcs->inBuffSize);
if (zcs->inBuff.buffer.start == NULL) { /* not enough memory to allocate next input buffer */ if (zcs->inBuff.buffer.start == NULL) { /* not enough memory to allocate next input buffer */
zcs->jobs[jobID].jobCompleted = 1; zcs->jobs[jobID].jobCompleted = 1;
zcs->nextJobID++; zcs->nextJobID++;
@ -747,18 +806,12 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
ZSTDMT_releaseAllJobResources(zcs); ZSTDMT_releaseAllJobResources(zcs);
return ERROR(memory_allocation); return ERROR(memory_allocation);
} }
DEBUGLOG(5, "inBuff currently filled to %u", (U32)zcs->inBuff.filled);
zcs->inBuff.filled -= srcSize + zcs->dictSize - newDictSize; zcs->inBuff.filled -= srcSize + zcs->dictSize - newDictSize;
DEBUGLOG(5, "new job : inBuff filled to %u, with %u dict and %u src",
(U32)zcs->inBuff.filled, (U32)newDictSize,
(U32)(zcs->inBuff.filled - newDictSize));
memmove(zcs->inBuff.buffer.start, memmove(zcs->inBuff.buffer.start,
(const char*)zcs->jobs[jobID].srcStart + zcs->dictSize + srcSize - newDictSize, (const char*)zcs->jobs[jobID].srcStart + zcs->dictSize + srcSize - newDictSize,
zcs->inBuff.filled); zcs->inBuff.filled);
DEBUGLOG(5, "new inBuff pre-filled");
zcs->dictSize = newDictSize; zcs->dictSize = newDictSize;
} else { /* if (endFrame==1) */ } else { /* if (endFrame==1) */
DEBUGLOG(5, "ZSTDMT_createCompressionJob::endFrame = %u", endFrame);
zcs->inBuff.buffer = g_nullBuffer; zcs->inBuff.buffer = g_nullBuffer;
zcs->inBuff.filled = 0; zcs->inBuff.filled = 0;
zcs->dictSize = 0; zcs->dictSize = 0;
@ -804,11 +857,8 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
ZSTDMT_releaseAllJobResources(zcs); ZSTDMT_releaseAllJobResources(zcs);
return job.cSize; return job.cSize;
} }
ZSTDMT_releaseCCtx(zcs->cctxPool, job.cctx);
zcs->jobs[wJobID].cctx = NULL;
DEBUGLOG(5, "zcs->params.fParams.checksumFlag : %u ", zcs->params.fParams.checksumFlag); DEBUGLOG(5, "zcs->params.fParams.checksumFlag : %u ", zcs->params.fParams.checksumFlag);
if (zcs->params.fParams.checksumFlag) { if (zcs->params.fParams.checksumFlag) {
XXH64_update(&zcs->xxhState, (const char*)job.srcStart + job.dictSize, job.srcSize);
if (zcs->frameEnded && (zcs->doneJobID+1 == zcs->nextJobID)) { /* write checksum at end of last section */ if (zcs->frameEnded && (zcs->doneJobID+1 == zcs->nextJobID)) { /* write checksum at end of last section */
U32 const checksum = (U32)XXH64_digest(&zcs->xxhState); U32 const checksum = (U32)XXH64_digest(&zcs->xxhState);
DEBUGLOG(5, "writing checksum : %08X \n", checksum); DEBUGLOG(5, "writing checksum : %08X \n", checksum);
@ -816,9 +866,6 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
job.cSize += 4; job.cSize += 4;
zcs->jobs[wJobID].cSize += 4; zcs->jobs[wJobID].cSize += 4;
} } } }
ZSTDMT_releaseBuffer(zcs->buffPool, job.src);
zcs->jobs[wJobID].srcStart = NULL;
zcs->jobs[wJobID].src = g_nullBuffer;
zcs->jobs[wJobID].jobScanned = 1; zcs->jobs[wJobID].jobScanned = 1;
} }
{ size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos); { size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos);
@ -828,7 +875,7 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
job.dstFlushed += toWrite; job.dstFlushed += toWrite;
} }
if (job.dstFlushed == job.cSize) { /* output buffer fully flushed => move to next one */ if (job.dstFlushed == job.cSize) { /* output buffer fully flushed => move to next one */
ZSTDMT_releaseBuffer(zcs->buffPool, job.dstBuff); ZSTDMT_releaseBuffer(zcs->bufPool, job.dstBuff);
zcs->jobs[wJobID].dstBuff = g_nullBuffer; zcs->jobs[wJobID].dstBuff = g_nullBuffer;
zcs->jobs[wJobID].jobCompleted = 0; zcs->jobs[wJobID].jobCompleted = 0;
zcs->doneJobID++; zcs->doneJobID++;
@ -852,18 +899,18 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx,
ZSTD_inBuffer* input, ZSTD_inBuffer* input,
ZSTD_EndDirective endOp) ZSTD_EndDirective endOp)
{ {
size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize + mtctx->marginSize; size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize;
assert(output->pos <= output->size); assert(output->pos <= output->size);
assert(input->pos <= input->size); assert(input->pos <= input->size);
if ((mtctx->frameEnded) && (endOp==ZSTD_e_continue)) { if ((mtctx->frameEnded) && (endOp==ZSTD_e_continue)) {
/* current frame being ended. Only flush/end are allowed. Or start new frame with init */ /* current frame being ended. Only flush/end are allowed. Or start new frame with init */
return ERROR(stage_wrong); return ERROR(stage_wrong);
} }
if (mtctx->nbThreads==1) { if (mtctx->nbThreads==1) { /* delegate to single-thread (synchronous) */
return ZSTD_compressStream_generic(mtctx->cctxPool->cctx[0], output, input, endOp); return ZSTD_compressStream_generic(mtctx->cctxPool->cctx[0], output, input, endOp);
} }
/* single-pass shortcut (note : this is blocking-mode) */ /* single-pass shortcut (note : this is synchronous-mode) */
if ( (mtctx->nextJobID==0) /* just started */ if ( (mtctx->nextJobID==0) /* just started */
&& (mtctx->inBuff.filled==0) /* nothing buffered */ && (mtctx->inBuff.filled==0) /* nothing buffered */
&& (endOp==ZSTD_e_end) /* end order */ && (endOp==ZSTD_e_end) /* end order */
@ -875,20 +922,25 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx,
if (ZSTD_isError(cSize)) return cSize; if (ZSTD_isError(cSize)) return cSize;
input->pos = input->size; input->pos = input->size;
output->pos += cSize; output->pos += cSize;
ZSTDMT_releaseBuffer(mtctx->buffPool, mtctx->inBuff.buffer); /* was allocated in initStream */ ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer); /* was allocated in initStream */
mtctx->allJobsCompleted = 1; mtctx->allJobsCompleted = 1;
mtctx->frameEnded = 1; mtctx->frameEnded = 1;
return 0; return 0;
} }
/* fill input buffer */ /* fill input buffer */
if ((input->src) && (mtctx->inBuff.buffer.start)) { /* support NULL input */ if (input->size > input->pos) { /* support NULL input */
size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled); if (mtctx->inBuff.buffer.start == NULL) {
DEBUGLOG(2, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad); mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool);
memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad); if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation);
input->pos += toLoad; mtctx->inBuff.filled = 0;
mtctx->inBuff.filled += toLoad; }
} { size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled);
DEBUGLOG(5, "inBuff:%08X; inBuffSize=%u; ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad);
memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad);
input->pos += toLoad;
mtctx->inBuff.filled += toLoad;
} }
if ( (mtctx->inBuff.filled >= newJobThreshold) /* filled enough : let's compress */ if ( (mtctx->inBuff.filled >= newJobThreshold) /* filled enough : let's compress */
&& (mtctx->nextJobID <= mtctx->doneJobID + mtctx->jobIDMask) ) { /* avoid overwriting job round buffer */ && (mtctx->nextJobID <= mtctx->doneJobID + mtctx->jobIDMask) ) { /* avoid overwriting job round buffer */

6
lib/compress/zstdmt_compress.h
View File

@ -16,8 +16,8 @@
/* Note : This is an internal API. /* Note : This is an internal API.
* Some methods are still exposed (ZSTDLIB_API), because for some time, * Some methods are still exposed (ZSTDLIB_API),
* it used to be the only way to invoke MT compression. * because it used to be the only way to invoke MT compression.
* Now, it's recommended to use ZSTD_compress_generic() instead. * Now, it's recommended to use ZSTD_compress_generic() instead.
* These methods will stop being exposed in a future version */ * These methods will stop being exposed in a future version */
@ -68,7 +68,7 @@ ZSTDLIB_API size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
const void* src, size_t srcSize, const void* src, size_t srcSize,
const ZSTD_CDict* cdict, const ZSTD_CDict* cdict,
ZSTD_parameters const params, ZSTD_parameters const params,
unsigned overlapRLog); unsigned overlapRLog); /* overlapRLog = 9 - overlapLog */
ZSTDLIB_API size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx, ZSTDLIB_API size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx,
const void* dict, size_t dictSize, /* dict can be released after init, a local copy is preserved within zcs */ const void* dict, size_t dictSize, /* dict can be released after init, a local copy is preserved within zcs */

12
tests/Makefile
View File

@ -79,7 +79,7 @@ all32: fullbench32 fuzzer32 zstreamtest32
allnothread: fullbench fuzzer paramgrill datagen decodecorpus allnothread: fullbench fuzzer paramgrill datagen decodecorpus
dll: fuzzer-dll zstreamtest-dll dll: fuzzer-dll zstreamtest-dll
zstd: zstd:
$(MAKE) -C $(PRGDIR) $@ $(MAKE) -C $(PRGDIR) $@
@ -108,11 +108,11 @@ fullbench-dll: $(PRGDIR)/datagen.c fullbench.c
$(MAKE) -C $(ZSTDDIR) libzstd $(MAKE) -C $(ZSTDDIR) libzstd
$(CC) $(FLAGS) $^ -o $@$(EXT) -DZSTD_DLL_IMPORT=1 $(ZSTDDIR)/dll/libzstd.dll $(CC) $(FLAGS) $^ -o $@$(EXT) -DZSTD_DLL_IMPORT=1 $(ZSTDDIR)/dll/libzstd.dll
fuzzer : $(ZSTD_FILES) $(ZDICT_FILES) $(PRGDIR)/datagen.c fuzzer.c fuzzer : CPPFLAGS += $(MULTITHREAD_CPP)
$(CC) $(FLAGS) $^ -o $@$(EXT) fuzzer : LDFLAGS += $(MULTITHREAD_LD)
fuzzer32: CFLAGS += -m32
fuzzer32 : $(ZSTD_FILES) $(ZDICT_FILES) $(PRGDIR)/datagen.c fuzzer.c fuzzer fuzzer32 : $(ZSTD_FILES) $(ZDICT_FILES) $(PRGDIR)/datagen.c fuzzer.c
$(CC) -m32 $(FLAGS) $^ -o $@$(EXT) $(CC) $(FLAGS) $^ -o $@$(EXT)
fuzzer-dll : LDFLAGS+= -L$(ZSTDDIR) -lzstd fuzzer-dll : LDFLAGS+= -L$(ZSTDDIR) -lzstd
fuzzer-dll : $(ZSTDDIR)/common/xxhash.c $(PRGDIR)/datagen.c fuzzer.c fuzzer-dll : $(ZSTDDIR)/common/xxhash.c $(PRGDIR)/datagen.c fuzzer.c

208
tests/fuzzer.c
View File

@ -51,14 +51,14 @@ static const U32 nbTestsDefault = 30000;
/*-************************************ /*-************************************
* Display Macros * Display Macros
**************************************/ **************************************/
#define DISPLAY(...) fprintf(stderr, __VA_ARGS__) #define DISPLAY(...) fprintf(stdout, __VA_ARGS__)
#define DISPLAYLEVEL(l, ...) if (g_displayLevel>=l) { DISPLAY(__VA_ARGS__); } #define DISPLAYLEVEL(l, ...) if (g_displayLevel>=l) { DISPLAY(__VA_ARGS__); }
static U32 g_displayLevel = 2; static U32 g_displayLevel = 2;
#define DISPLAYUPDATE(l, ...) if (g_displayLevel>=l) { \ #define DISPLAYUPDATE(l, ...) if (g_displayLevel>=l) { \
if ((FUZ_clockSpan(g_displayClock) > g_refreshRate) || (g_displayLevel>=4)) \ if ((FUZ_clockSpan(g_displayClock) > g_refreshRate) || (g_displayLevel>=4)) \
{ g_displayClock = clock(); DISPLAY(__VA_ARGS__); \ { g_displayClock = clock(); DISPLAY(__VA_ARGS__); \
if (g_displayLevel>=4) fflush(stderr); } } if (g_displayLevel>=4) fflush(stdout); } }
static const clock_t g_refreshRate = CLOCKS_PER_SEC / 6; static const clock_t g_refreshRate = CLOCKS_PER_SEC / 6;
static clock_t g_displayClock = 0; static clock_t g_displayClock = 0;
@ -97,7 +97,161 @@ static unsigned FUZ_highbit32(U32 v32)
/*============================================= /*=============================================
* Basic Unit tests * Memory Tests
=============================================*/
#if defined(__APPLE__) && defined(__MACH__)
#include <malloc/malloc.h> /* malloc_size */
typedef struct {
unsigned long long totalMalloc;
size_t currentMalloc;
size_t peakMalloc;
unsigned nbMalloc;
unsigned nbFree;
} mallocCounter_t;
static const mallocCounter_t INIT_MALLOC_COUNTER = { 0, 0, 0, 0, 0 };
static void* FUZ_mallocDebug(void* counter, size_t size)
{
mallocCounter_t* const mcPtr = (mallocCounter_t*)counter;
void* const ptr = malloc(size);
if (ptr==NULL) return NULL;
DISPLAYLEVEL(4, "allocating %u KB => effectively %u KB \n",
(U32)(size >> 10), (U32)(malloc_size(ptr) >> 10)); /* OS-X specific */
mcPtr->totalMalloc += size;
mcPtr->currentMalloc += size;
if (mcPtr->currentMalloc > mcPtr->peakMalloc)
mcPtr->peakMalloc = mcPtr->currentMalloc;
mcPtr->nbMalloc += 1;
return ptr;
}
static void FUZ_freeDebug(void* counter, void* address)
{
mallocCounter_t* const mcPtr = (mallocCounter_t*)counter;
DISPLAYLEVEL(4, "freeing %u KB \n", (U32)(malloc_size(address) >> 10));
mcPtr->nbFree += 1;
mcPtr->currentMalloc -= malloc_size(address); /* OS-X specific */
free(address);
}
static void FUZ_displayMallocStats(mallocCounter_t count)
{
DISPLAYLEVEL(3, "peak:%u KB, nbMallocs:%u, total:%u KB \n",
(U32)(count.peakMalloc >> 10),
count.nbMalloc,
(U32)(count.totalMalloc >> 10));
}
#define CHECK_Z(f) { \
size_t const err = f; \
if (ZSTD_isError(err)) { \
DISPLAY("Error => %s : %s ", \
#f, ZSTD_getErrorName(err)); \
exit(1); \
} }
static int FUZ_mallocTests(unsigned seed, double compressibility)
{
size_t const inSize = 64 MB + 16 MB + 4 MB + 1 MB + 256 KB + 64 KB; /* 85.3 MB */
size_t const outSize = ZSTD_compressBound(inSize);
void* const inBuffer = malloc(inSize);
void* const outBuffer = malloc(outSize);
/* test only played in verbose mode, as they are long */
if (g_displayLevel<3) return 0;
/* Create compressible noise */
if (!inBuffer || !outBuffer) {
DISPLAY("Not enough memory, aborting\n");
exit(1);
}
RDG_genBuffer(inBuffer, inSize, compressibility, 0. /*auto*/, seed);
/* simple compression tests */
{ int compressionLevel;
for (compressionLevel=1; compressionLevel<=6; compressionLevel++) {
mallocCounter_t malcount = INIT_MALLOC_COUNTER;
ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount };
ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem);
CHECK_Z( ZSTD_compressCCtx(cctx, outBuffer, outSize, inBuffer, inSize, compressionLevel) );
ZSTD_freeCCtx(cctx);
DISPLAYLEVEL(3, "compressCCtx level %i : ", compressionLevel);
FUZ_displayMallocStats(malcount);
} }
/* streaming compression tests */
{ int compressionLevel;
for (compressionLevel=1; compressionLevel<=6; compressionLevel++) {
mallocCounter_t malcount = INIT_MALLOC_COUNTER;
ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount };
ZSTD_CCtx* const cstream = ZSTD_createCStream_advanced(cMem);
ZSTD_outBuffer out = { outBuffer, outSize, 0 };
ZSTD_inBuffer in = { inBuffer, inSize, 0 };
CHECK_Z( ZSTD_initCStream(cstream, compressionLevel) );
CHECK_Z( ZSTD_compressStream(cstream, &out, &in) );
CHECK_Z( ZSTD_endStream(cstream, &out) );
ZSTD_freeCStream(cstream);
DISPLAYLEVEL(3, "compressStream level %i : ", compressionLevel);
FUZ_displayMallocStats(malcount);
} }
/* advanced MT API test */
{ U32 nbThreads;
for (nbThreads=1; nbThreads<=4; nbThreads++) {
int compressionLevel;
for (compressionLevel=1; compressionLevel<=6; compressionLevel++) {
mallocCounter_t malcount = INIT_MALLOC_COUNTER;
ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount };
ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem);
ZSTD_outBuffer out = { outBuffer, outSize, 0 };
ZSTD_inBuffer in = { inBuffer, inSize, 0 };
CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel) );
CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads) );
while ( ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end) ) {}
ZSTD_freeCCtx(cctx);
DISPLAYLEVEL(3, "compress_generic,-T%u,end level %i : ",
nbThreads, compressionLevel);
FUZ_displayMallocStats(malcount);
} } }
/* advanced MT streaming API test */
{ U32 nbThreads;
for (nbThreads=1; nbThreads<=4; nbThreads++) {
int compressionLevel;
for (compressionLevel=1; compressionLevel<=6; compressionLevel++) {
mallocCounter_t malcount = INIT_MALLOC_COUNTER;
ZSTD_customMem const cMem = { FUZ_mallocDebug, FUZ_freeDebug, &malcount };
ZSTD_CCtx* const cctx = ZSTD_createCCtx_advanced(cMem);
ZSTD_outBuffer out = { outBuffer, outSize, 0 };
ZSTD_inBuffer in = { inBuffer, inSize, 0 };
CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_compressionLevel, (U32)compressionLevel) );
CHECK_Z( ZSTD_CCtx_setParameter(cctx, ZSTD_p_nbThreads, nbThreads) );
CHECK_Z( ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_continue) );
while ( ZSTD_compress_generic(cctx, &out, &in, ZSTD_e_end) ) {}
ZSTD_freeCCtx(cctx);
DISPLAYLEVEL(3, "compress_generic,-T%u,continue level %i : ",
nbThreads, compressionLevel);
FUZ_displayMallocStats(malcount);
} } }
return 0;
}
#else
static int FUZ_mallocTests(unsigned seed, double compressibility)
{
(void)seed; (void)compressibility;
return 0;
}
#endif
/*=============================================
* Unit tests
=============================================*/ =============================================*/
#define CHECK_V(var, fn) size_t const var = fn; if (ZSTD_isError(var)) goto _output_error #define CHECK_V(var, fn) size_t const var = fn; if (ZSTD_isError(var)) goto _output_error
@ -108,7 +262,8 @@ static int basicUnitTests(U32 seed, double compressibility)
{ {
size_t const CNBuffSize = 5 MB; size_t const CNBuffSize = 5 MB;
void* const CNBuffer = malloc(CNBuffSize); void* const CNBuffer = malloc(CNBuffSize);
void* const compressedBuffer = malloc(ZSTD_compressBound(CNBuffSize)); size_t const compressedBufferSize = ZSTD_compressBound(CNBuffSize);
void* const compressedBuffer = malloc(compressedBufferSize);
void* const decodedBuffer = malloc(CNBuffSize); void* const decodedBuffer = malloc(CNBuffSize);
ZSTD_DCtx* dctx = ZSTD_createDCtx(); ZSTD_DCtx* dctx = ZSTD_createDCtx();
int testResult = 0; int testResult = 0;
@ -139,7 +294,7 @@ static int basicUnitTests(U32 seed, double compressibility)
{ ZSTD_CCtx* cctx = ZSTD_createCCtx(); { ZSTD_CCtx* cctx = ZSTD_createCCtx();
if (cctx==NULL) goto _output_error; if (cctx==NULL) goto _output_error;
CHECKPLUS(r, ZSTD_compressCCtx(cctx, CHECKPLUS(r, ZSTD_compressCCtx(cctx,
compressedBuffer, ZSTD_compressBound(CNBuffSize), compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, 1), CNBuffer, CNBuffSize, 1),
cSize=r ); cSize=r );
DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100); DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100);
@ -226,7 +381,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(4, "test%3i : simple compression test with static CCtx : ", testNb++); DISPLAYLEVEL(4, "test%3i : simple compression test with static CCtx : ", testNb++);
CHECKPLUS(r, ZSTD_compressCCtx(staticCCtx, CHECKPLUS(r, ZSTD_compressCCtx(staticCCtx,
compressedBuffer, ZSTD_compressBound(CNBuffSize), compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, STATIC_CCTX_LEVEL), CNBuffer, CNBuffSize, STATIC_CCTX_LEVEL),
cSize=r ); cSize=r );
DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n",
@ -295,7 +450,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(4, "test%3i : compress %u bytes with 2 threads : ", testNb++, (U32)CNBuffSize); DISPLAYLEVEL(4, "test%3i : compress %u bytes with 2 threads : ", testNb++, (U32)CNBuffSize);
CHECKPLUS(r, ZSTDMT_compressCCtx(mtctx, CHECKPLUS(r, ZSTDMT_compressCCtx(mtctx,
compressedBuffer, ZSTD_compressBound(CNBuffSize), compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, CNBuffer, CNBuffSize,
1), 1),
cSize=r ); cSize=r );
@ -321,6 +476,23 @@ static int basicUnitTests(U32 seed, double compressibility)
} } } }
DISPLAYLEVEL(4, "OK \n"); DISPLAYLEVEL(4, "OK \n");
DISPLAYLEVEL(4, "test%3i : compress -T2 with checksum : ", testNb++);
{ ZSTD_parameters params = ZSTD_getParams(1, CNBuffSize, 0);
params.fParams.checksumFlag = 1;
params.fParams.contentSizeFlag = 1;
CHECKPLUS(r, ZSTDMT_compress_advanced(mtctx,
compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize,
NULL, params, 3 /*overlapRLog*/),
cSize=r );
}
DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100);
DISPLAYLEVEL(4, "test%3i : decompress %u bytes : ", testNb++, (U32)CNBuffSize);
{ size_t const r = ZSTD_decompress(decodedBuffer, CNBuffSize, compressedBuffer, cSize);
if (r != CNBuffSize) goto _output_error; }
DISPLAYLEVEL(4, "OK \n");
ZSTDMT_freeCCtx(mtctx); ZSTDMT_freeCCtx(mtctx);
} }
@ -382,7 +554,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(4, "test%3i : compress with flat dictionary : ", testNb++); DISPLAYLEVEL(4, "test%3i : compress with flat dictionary : ", testNb++);
cSize = 0; cSize = 0;
CHECKPLUS(r, ZSTD_compressEnd(ctxOrig, compressedBuffer, ZSTD_compressBound(CNBuffSize), CHECKPLUS(r, ZSTD_compressEnd(ctxOrig, compressedBuffer, compressedBufferSize,
(const char*)CNBuffer + dictSize, CNBuffSize - dictSize), (const char*)CNBuffer + dictSize, CNBuffSize - dictSize),
cSize += r); cSize += r);
DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100); DISPLAYLEVEL(4, "OK (%u bytes : %.2f%%)\n", (U32)cSize, (double)cSize/CNBuffSize*100);
@ -398,7 +570,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(4, "test%3i : compress with duplicated context : ", testNb++); DISPLAYLEVEL(4, "test%3i : compress with duplicated context : ", testNb++);
{ size_t const cSizeOrig = cSize; { size_t const cSizeOrig = cSize;
cSize = 0; cSize = 0;
CHECKPLUS(r, ZSTD_compressEnd(ctxDuplicated, compressedBuffer, ZSTD_compressBound(CNBuffSize), CHECKPLUS(r, ZSTD_compressEnd(ctxDuplicated, compressedBuffer, compressedBufferSize,
(const char*)CNBuffer + dictSize, CNBuffSize - dictSize), (const char*)CNBuffer + dictSize, CNBuffSize - dictSize),
cSize += r); cSize += r);
if (cSize != cSizeOrig) goto _output_error; /* should be identical ==> same size */ if (cSize != cSizeOrig) goto _output_error; /* should be identical ==> same size */
@ -483,7 +655,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(4, "OK : %u \n", dictID); DISPLAYLEVEL(4, "OK : %u \n", dictID);
DISPLAYLEVEL(4, "test%3i : compress with dictionary : ", testNb++); DISPLAYLEVEL(4, "test%3i : compress with dictionary : ", testNb++);
cSize = ZSTD_compress_usingDict(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), cSize = ZSTD_compress_usingDict(cctx, compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, CNBuffer, CNBuffSize,
dictBuffer, dictSize, 4); dictBuffer, dictSize, 4);
if (ZSTD_isError(cSize)) goto _output_error; if (ZSTD_isError(cSize)) goto _output_error;
@ -521,7 +693,7 @@ static int basicUnitTests(U32 seed, double compressibility)
1 /* byReference */, ZSTD_dm_auto, 1 /* byReference */, ZSTD_dm_auto,
cParams, ZSTD_defaultCMem); cParams, ZSTD_defaultCMem);
DISPLAYLEVEL(4, "(size : %u) : ", (U32)ZSTD_sizeof_CDict(cdict)); DISPLAYLEVEL(4, "(size : %u) : ", (U32)ZSTD_sizeof_CDict(cdict));
cSize = ZSTD_compress_usingCDict(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), cSize = ZSTD_compress_usingCDict(cctx, compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, cdict); CNBuffer, CNBuffSize, cdict);
ZSTD_freeCDict(cdict); ZSTD_freeCDict(cdict);
if (ZSTD_isError(cSize)) goto _output_error; if (ZSTD_isError(cSize)) goto _output_error;
@ -556,7 +728,7 @@ static int basicUnitTests(U32 seed, double compressibility)
goto _output_error; goto _output_error;
} }
cSize = ZSTD_compress_usingCDict(cctx, cSize = ZSTD_compress_usingCDict(cctx,
compressedBuffer, ZSTD_compressBound(CNBuffSize), compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, cdict); CNBuffer, CNBuffSize, cdict);
if (ZSTD_isError(cSize)) { if (ZSTD_isError(cSize)) {
DISPLAY("ZSTD_compress_usingCDict failed "); DISPLAY("ZSTD_compress_usingCDict failed ");
@ -570,7 +742,7 @@ static int basicUnitTests(U32 seed, double compressibility)
{ ZSTD_frameParameters const fParams = { 0 /* frameSize */, 1 /* checksum */, 1 /* noDictID*/ }; { ZSTD_frameParameters const fParams = { 0 /* frameSize */, 1 /* checksum */, 1 /* noDictID*/ };
ZSTD_compressionParameters const cParams = ZSTD_getCParams(1, CNBuffSize, dictSize); ZSTD_compressionParameters const cParams = ZSTD_getCParams(1, CNBuffSize, dictSize);
ZSTD_CDict* const cdict = ZSTD_createCDict_advanced(dictBuffer, dictSize, 1 /*byRef*/, ZSTD_dm_auto, cParams, ZSTD_defaultCMem); ZSTD_CDict* const cdict = ZSTD_createCDict_advanced(dictBuffer, dictSize, 1 /*byRef*/, ZSTD_dm_auto, cParams, ZSTD_defaultCMem);
cSize = ZSTD_compress_usingCDict_advanced(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), cSize = ZSTD_compress_usingCDict_advanced(cctx, compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, cdict, fParams); CNBuffer, CNBuffSize, cdict, fParams);
ZSTD_freeCDict(cdict); ZSTD_freeCDict(cdict);
if (ZSTD_isError(cSize)) goto _output_error; if (ZSTD_isError(cSize)) goto _output_error;
@ -594,7 +766,7 @@ static int basicUnitTests(U32 seed, double compressibility)
DISPLAYLEVEL(4, "test%3i : ZSTD_compress_advanced, no dictID : ", testNb++); DISPLAYLEVEL(4, "test%3i : ZSTD_compress_advanced, no dictID : ", testNb++);
{ ZSTD_parameters p = ZSTD_getParams(3, CNBuffSize, dictSize); { ZSTD_parameters p = ZSTD_getParams(3, CNBuffSize, dictSize);
p.fParams.noDictIDFlag = 1; p.fParams.noDictIDFlag = 1;
cSize = ZSTD_compress_advanced(cctx, compressedBuffer, ZSTD_compressBound(CNBuffSize), cSize = ZSTD_compress_advanced(cctx, compressedBuffer, compressedBufferSize,
CNBuffer, CNBuffSize, CNBuffer, CNBuffSize,
dictBuffer, dictSize, p); dictBuffer, dictSize, p);
if (ZSTD_isError(cSize)) goto _output_error; if (ZSTD_isError(cSize)) goto _output_error;
@ -902,6 +1074,7 @@ static size_t FUZ_randomLength(U32* seed, U32 maxLog)
goto _output_error; \ goto _output_error; \
} } } }
#undef CHECK_Z
#define CHECK_Z(f) { \ #define CHECK_Z(f) { \
size_t const err = f; \ size_t const err = f; \
if (ZSTD_isError(err)) { \ if (ZSTD_isError(err)) { \
@ -1245,6 +1418,7 @@ int main(int argc, const char** argv)
U32 mainPause = 0; U32 mainPause = 0;
U32 maxDuration = 0; U32 maxDuration = 0;
int bigTests = 1; int bigTests = 1;
U32 memTestsOnly = 0;
const char* const programName = argv[0]; const char* const programName = argv[0];
/* Check command line */ /* Check command line */
@ -1255,6 +1429,7 @@ int main(int argc, const char** argv)
/* Handle commands. Aggregated commands are allowed */ /* Handle commands. Aggregated commands are allowed */
if (argument[0]=='-') { if (argument[0]=='-') {
if (!strcmp(argument, "--memtest")) { memTestsOnly=1; continue; }
if (!strcmp(argument, "--no-big-tests")) { bigTests=0; continue; } if (!strcmp(argument, "--no-big-tests")) { bigTests=0; continue; }
argument++; argument++;
@ -1326,6 +1501,11 @@ int main(int argc, const char** argv)
DISPLAY("Seed = %u\n", seed); DISPLAY("Seed = %u\n", seed);
if (proba!=FUZ_compressibility_default) DISPLAY("Compressibility : %u%%\n", proba); if (proba!=FUZ_compressibility_default) DISPLAY("Compressibility : %u%%\n", proba);
if (memTestsOnly) {
g_displayLevel = MAX(3, g_displayLevel);
return FUZ_mallocTests(seed, ((double)proba) / 100);
}
if (nbTests < testNb) nbTests = testNb; if (nbTests < testNb) nbTests = testNb;
if (testNb==0) if (testNb==0)

31
tests/playTests.sh
View File

@ -7,17 +7,17 @@ die() {
roundTripTest() { roundTripTest() {
if [ -n "$3" ]; then if [ -n "$3" ]; then
local_c="$3" cLevel="$3"
local_p="$2" proba="$2"
else else
local_c="$2" cLevel="$2"
local_p="" proba=""
fi fi
rm -f tmp1 tmp2 rm -f tmp1 tmp2
$ECHO "roundTripTest: ./datagen $1 $local_p | $ZSTD -v$local_c | $ZSTD -d" $ECHO "roundTripTest: ./datagen $1 $proba | $ZSTD -v$cLevel | $ZSTD -d"
./datagen $1 $local_p | $MD5SUM > tmp1 ./datagen $1 $proba | $MD5SUM > tmp1
./datagen $1 $local_p | $ZSTD --ultra -v$local_c | $ZSTD -d | $MD5SUM > tmp2 ./datagen $1 $proba | $ZSTD --ultra -v$cLevel | $ZSTD -d | $MD5SUM > tmp2
$DIFF -q tmp1 tmp2 $DIFF -q tmp1 tmp2
} }
@ -625,16 +625,15 @@ roundTripTest -g35000000 -P75 10
roundTripTest -g35000000 -P75 11 roundTripTest -g35000000 -P75 11
roundTripTest -g35000000 -P75 12 roundTripTest -g35000000 -P75 12
roundTripTest -g18000000 -P80 13 roundTripTest -g18000013 -P80 13
roundTripTest -g18000000 -P80 14 roundTripTest -g18000014 -P80 14
roundTripTest -g18000000 -P80 15 roundTripTest -g18000015 -P80 15
roundTripTest -g18000000 -P80 16 roundTripTest -g18000016 -P80 16
roundTripTest -g18000000 -P80 17 roundTripTest -g18000017 -P80 17
roundTripTest -g18000018 -P94 18
roundTripTest -g18000019 -P94 19
roundTripTest -g50000000 -P94 18 roundTripTest -g68000020 -P99 20
roundTripTest -g50000000 -P94 19
roundTripTest -g99000000 -P99 20
roundTripTest -g6000000000 -P99 1 roundTripTest -g6000000000 -P99 1
fileRoundTripTest -g4193M -P99 1 fileRoundTripTest -g4193M -P99 1

26
tests/zstreamtest.c
View File

@ -95,19 +95,6 @@ unsigned int FUZ_rand(unsigned int* seedPtr)
return rand32 >> 5; return rand32 >> 5;
} }
static void* allocFunction(void* opaque, size_t size)
{
void* address = malloc(size);
(void)opaque;
return address;
}
static void freeFunction(void* opaque, void* address)
{
(void)opaque;
free(address);
}
/*====================================================== /*======================================================
* Basic Unit tests * Basic Unit tests
@ -1390,13 +1377,12 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double
/* multi-segments compression test */ /* multi-segments compression test */
XXH64_reset(&xxhState, 0); XXH64_reset(&xxhState, 0);
{ ZSTD_outBuffer outBuff = { cBuffer, cBufferSize, 0 } ; { ZSTD_outBuffer outBuff = { cBuffer, cBufferSize, 0 } ;
U32 n; for (cSize=0, totalTestSize=0 ; (totalTestSize < maxTestSize) ; ) {
for (n=0, cSize=0, totalTestSize=0 ; totalTestSize < maxTestSize ; n++) {
/* compress random chunks into randomly sized dst buffers */ /* compress random chunks into randomly sized dst buffers */
size_t const randomSrcSize = FUZ_randomLength(&lseed, maxSampleLog); size_t const randomSrcSize = FUZ_randomLength(&lseed, maxSampleLog);
size_t const srcSize = MIN(maxTestSize-totalTestSize, randomSrcSize); size_t const srcSize = MIN(maxTestSize-totalTestSize, randomSrcSize);
size_t const srcStart = FUZ_rand(&lseed) % (srcBufferSize - srcSize); size_t const srcStart = FUZ_rand(&lseed) % (srcBufferSize - srcSize);
size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog); size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog+1);
size_t const dstBuffSize = MIN(cBufferSize - cSize, randomDstSize); size_t const dstBuffSize = MIN(cBufferSize - cSize, randomDstSize);
ZSTD_EndDirective const flush = (FUZ_rand(&lseed) & 15) ? ZSTD_e_continue : ZSTD_e_flush; ZSTD_EndDirective const flush = (FUZ_rand(&lseed) & 15) ? ZSTD_e_continue : ZSTD_e_flush;
ZSTD_inBuffer inBuff = { srcBuffer+srcStart, srcSize, 0 }; ZSTD_inBuffer inBuff = { srcBuffer+srcStart, srcSize, 0 };
@ -1415,7 +1401,7 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double
{ size_t remainingToFlush = (size_t)(-1); { size_t remainingToFlush = (size_t)(-1);
while (remainingToFlush) { while (remainingToFlush) {
ZSTD_inBuffer inBuff = { NULL, 0, 0 }; ZSTD_inBuffer inBuff = { NULL, 0, 0 };
size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog); size_t const randomDstSize = FUZ_randomLength(&lseed, maxSampleLog+1);
size_t const adjustedDstSize = MIN(cBufferSize - cSize, randomDstSize); size_t const adjustedDstSize = MIN(cBufferSize - cSize, randomDstSize);
outBuff.size = outBuff.pos + adjustedDstSize; outBuff.size = outBuff.pos + adjustedDstSize;
DISPLAYLEVEL(5, "End-flush into dst buffer of size %u \n", (U32)adjustedDstSize); DISPLAYLEVEL(5, "End-flush into dst buffer of size %u \n", (U32)adjustedDstSize);
@ -1543,7 +1529,6 @@ int main(int argc, const char** argv)
int bigTests = (sizeof(size_t) == 8); int bigTests = (sizeof(size_t) == 8);
e_api selected_api = simple_api; e_api selected_api = simple_api;
const char* const programName = argv[0]; const char* const programName = argv[0];
ZSTD_customMem const customMem = { allocFunction, freeFunction, NULL };
ZSTD_customMem const customNULL = ZSTD_defaultCMem; ZSTD_customMem const customNULL = ZSTD_defaultCMem;
/* Check command line */ /* Check command line */
@ -1657,10 +1642,7 @@ int main(int argc, const char** argv)
if (testNb==0) { if (testNb==0) {
result = basicUnitTests(0, ((double)proba) / 100, customNULL); /* constant seed for predictability */ result = basicUnitTests(0, ((double)proba) / 100, customNULL); /* constant seed for predictability */
if (!result) { }
DISPLAYLEVEL(3, "Unit tests using customMem :\n")
result = basicUnitTests(0, ((double)proba) / 100, customMem); /* use custom memory allocation functions */
} }
if (!result) { if (!result) {
switch(selected_api) switch(selected_api)