contrib/zstd/lib/compress/zstdmt_compress.c

   1 /*
   2  * Copyright (c) 2016-present, Yann Collet, Facebook, Inc.
   3  * All rights reserved.
   4  *
   5  * This source code is licensed under both the BSD-style license (found in the
   6  * LICENSE file in the root directory of this source tree) and the GPLv2 (found
   7  * in the COPYING file in the root directory of this source tree).
   8  */
   9
  10
  11 /* ======   Tuning parameters   ====== */
  12 #define ZSTDMT_NBTHREADS_MAX 256
  13 #define ZSTDMT_OVERLAPLOG_DEFAULT 6
  14
  15
  16 /* ======   Compiler specifics   ====== */
  17 #if defined(_MSC_VER)
  18 #  pragma warning(disable : 4204)   /* disable: C4204: non-constant aggregate initializer */
  19 #endif
  20
  21
  22 /* ======   Dependencies   ====== */
  23 #include <string.h>      /* memcpy, memset */
  24 #include "pool.h"        /* threadpool */
  25 #include "threading.h"   /* mutex */
  26 #include "zstd_internal.h"  /* MIN, ERROR, ZSTD_*, ZSTD_highbit32 */
  27 #include "zstdmt_compress.h"
  28
  29
  30 /* ======   Debug   ====== */
  31 #if defined(ZSTD_DEBUG) && (ZSTD_DEBUG>=2)
  32
  33 #  include <stdio.h>
  34 #  include <unistd.h>
  35 #  include <sys/times.h>
  36 #  define DEBUGLOGRAW(l, ...) if (l<=ZSTD_DEBUG) { fprintf(stderr, __VA_ARGS__); }
  37
  38 #  define DEBUG_PRINTHEX(l,p,n) {            \
  39     unsigned debug_u;                        \
  40     for (debug_u=0; debug_u<(n); debug_u++)  \
  41         DEBUGLOGRAW(l, "%02X ", ((const unsigned char*)(p))[debug_u]); \
  42     DEBUGLOGRAW(l, " \n");                   \
  43 }
  44
  45 static unsigned long long GetCurrentClockTimeMicroseconds(void)
  46 {
  47    static clock_t _ticksPerSecond = 0;
  48    if (_ticksPerSecond <= 0) _ticksPerSecond = sysconf(_SC_CLK_TCK);
  49
  50    { struct tms junk; clock_t newTicks = (clock_t) times(&junk);
  51      return ((((unsigned long long)newTicks)*(1000000))/_ticksPerSecond); }
  52 }
  53
  54 #define MUTEX_WAIT_TIME_DLEVEL 6
  55 #define PTHREAD_MUTEX_LOCK(mutex) {               \
  56     if (ZSTD_DEBUG>=MUTEX_WAIT_TIME_DLEVEL) {   \
  57         unsigned long long const beforeTime = GetCurrentClockTimeMicroseconds(); \
  58         pthread_mutex_lock(mutex);                \
  59         {   unsigned long long const afterTime = GetCurrentClockTimeMicroseconds(); \
  60             unsigned long long const elapsedTime = (afterTime-beforeTime); \
  61             if (elapsedTime > 1000) {  /* or whatever threshold you like; I'm using 1 millisecond here */ \
  62                 DEBUGLOG(MUTEX_WAIT_TIME_DLEVEL, "Thread took %llu microseconds to acquire mutex %s \n", \
  63                    elapsedTime, #mutex);          \
  64         }   }                                     \
  65     } else pthread_mutex_lock(mutex);             \
  66 }
  67
  68 #else
  69
  70 #  define PTHREAD_MUTEX_LOCK(m) pthread_mutex_lock(m)
  71 #  define DEBUG_PRINTHEX(l,p,n) {}
  72
  73 #endif
  74
  75
  76 /* =====   Buffer Pool   ===== */
  77 /* a single Buffer Pool can be invoked from multiple threads in parallel */
  78
  79 typedef struct buffer_s {
  80     void* start;
  81     size_t size;
  82 } buffer_t;
  83
  84 static const buffer_t g_nullBuffer = { NULL, 0 };
  85
  86 typedef struct ZSTDMT_bufferPool_s {
  87     pthread_mutex_t poolMutex;
  88     size_t bufferSize;
  89     unsigned totalBuffers;
  90     unsigned nbBuffers;
  91     ZSTD_customMem cMem;
  92     buffer_t bTable[1];   /* variable size */
  93 } ZSTDMT_bufferPool;
  94
  95 static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned nbThreads, ZSTD_customMem cMem)
  96 {
  97     unsigned const maxNbBuffers = 2*nbThreads + 3;
  98     ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_calloc(
  99         sizeof(ZSTDMT_bufferPool) + (maxNbBuffers-1) * sizeof(buffer_t), cMem);
 100     if (bufPool==NULL) return NULL;
 101     if (pthread_mutex_init(&bufPool->poolMutex, NULL)) {
 102         ZSTD_free(bufPool, cMem);
 103         return NULL;
 104     }
 105     bufPool->bufferSize = 64 KB;
 106     bufPool->totalBuffers = maxNbBuffers;
 107     bufPool->nbBuffers = 0;
 108     bufPool->cMem = cMem;
 109     return bufPool;
 110 }
 111
 112 static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool)
 113 {
 114     unsigned u;
 115     if (!bufPool) return;   /* compatibility with free on NULL */
 116     for (u=0; u<bufPool->totalBuffers; u++)
 117         ZSTD_free(bufPool->bTable[u].start, bufPool->cMem);
 118     pthread_mutex_destroy(&bufPool->poolMutex);
 119     ZSTD_free(bufPool, bufPool->cMem);
 120 }
 121
 122 /* only works at initialization, not during compression */
 123 static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)
 124 {
 125     size_t const poolSize = sizeof(*bufPool)
 126                             + (bufPool->totalBuffers - 1) * sizeof(buffer_t);
 127     unsigned u;
 128     size_t totalBufferSize = 0;
 129     pthread_mutex_lock(&bufPool->poolMutex);
 130     for (u=0; u<bufPool->totalBuffers; u++)
 131         totalBufferSize += bufPool->bTable[u].size;
 132     pthread_mutex_unlock(&bufPool->poolMutex);
 133
 134     return poolSize + totalBufferSize;
 135 }
 136
 137 static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* bufPool, size_t bSize)
 138 {
 139     bufPool->bufferSize = bSize;
 140 }
 141
 142 /** ZSTDMT_getBuffer() :
 143  *  assumption : bufPool must be valid */
 144 static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
 145 {
 146     size_t const bSize = bufPool->bufferSize;
 147     DEBUGLOG(5, "ZSTDMT_getBuffer");
 148     pthread_mutex_lock(&bufPool->poolMutex);
 149     if (bufPool->nbBuffers) {   /* try to use an existing buffer */
 150         buffer_t const buf = bufPool->bTable[--(bufPool->nbBuffers)];
 151         size_t const availBufferSize = buf.size;
 152         if ((availBufferSize >= bSize) & (availBufferSize <= 10*bSize)) {
 153             /* large enough, but not too much */
 154             pthread_mutex_unlock(&bufPool->poolMutex);
 155             return buf;
 156         }
 157         /* size conditions not respected : scratch this buffer, create new one */
 158         DEBUGLOG(5, "existing buffer does not meet size conditions => freeing");
 159         ZSTD_free(buf.start, bufPool->cMem);
 160     }
 161     pthread_mutex_unlock(&bufPool->poolMutex);
 162     /* create new buffer */
 163     DEBUGLOG(5, "create a new buffer");
 164     {   buffer_t buffer;
 165         void* const start = ZSTD_malloc(bSize, bufPool->cMem);
 166         buffer.start = start;   /* note : start can be NULL if malloc fails ! */
 167         buffer.size = (start==NULL) ? 0 : bSize;
 168         return buffer;
 169     }
 170 }
 171
 172 /* store buffer for later re-use, up to pool capacity */
 173 static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf)
 174 {
 175     if (buf.start == NULL) return;   /* compatible with release on NULL */
 176     DEBUGLOG(5, "ZSTDMT_releaseBuffer");
 177     pthread_mutex_lock(&bufPool->poolMutex);
 178     if (bufPool->nbBuffers < bufPool->totalBuffers) {
 179         bufPool->bTable[bufPool->nbBuffers++] = buf;  /* stored for later use */
 180         pthread_mutex_unlock(&bufPool->poolMutex);
 181         return;
 182     }
 183     pthread_mutex_unlock(&bufPool->poolMutex);
 184     /* Reached bufferPool capacity (should not happen) */
 185     DEBUGLOG(5, "buffer pool capacity reached => freeing ");
 186     ZSTD_free(buf.start, bufPool->cMem);
 187 }
 188
 189
 190 /* =====   CCtx Pool   ===== */
 191 /* a single CCtx Pool can be invoked from multiple threads in parallel */
 192
 193 typedef struct {
 194     pthread_mutex_t poolMutex;
 195     unsigned totalCCtx;
 196     unsigned availCCtx;
 197     ZSTD_customMem cMem;
 198     ZSTD_CCtx* cctx[1];   /* variable size */
 199 } ZSTDMT_CCtxPool;
 200
 201 /* note : all CCtx borrowed from the pool should be released back to the pool _before_ freeing the pool */
 202 static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool)
 203 {
 204     unsigned u;
 205     for (u=0; u<pool->totalCCtx; u++)
 206         ZSTD_freeCCtx(pool->cctx[u]);  /* note : compatible with free on NULL */
 207     pthread_mutex_destroy(&pool->poolMutex);
 208     ZSTD_free(pool, pool->cMem);
 209 }
 210
 211 /* ZSTDMT_createCCtxPool() :
 212  * implies nbThreads >= 1 , checked by caller ZSTDMT_createCCtx() */
 213 static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(unsigned nbThreads,
 214                                               ZSTD_customMem cMem)
 215 {
 216     ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_calloc(
 217         sizeof(ZSTDMT_CCtxPool) + (nbThreads-1)*sizeof(ZSTD_CCtx*), cMem);
 218     if (!cctxPool) return NULL;
 219     if (pthread_mutex_init(&cctxPool->poolMutex, NULL)) {
 220         ZSTD_free(cctxPool, cMem);
 221         return NULL;
 222     }
 223     cctxPool->cMem = cMem;
 224     cctxPool->totalCCtx = nbThreads;
 225     cctxPool->availCCtx = 1;   /* at least one cctx for single-thread mode */
 226     cctxPool->cctx[0] = ZSTD_createCCtx_advanced(cMem);
 227     if (!cctxPool->cctx[0]) { ZSTDMT_freeCCtxPool(cctxPool); return NULL; }
 228     DEBUGLOG(3, "cctxPool created, with %u threads", nbThreads);
 229     return cctxPool;
 230 }
 231
 232 /* only works during initialization phase, not during compression */
 233 static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool)
 234 {
 235     pthread_mutex_lock(&cctxPool->poolMutex);
 236     {   unsigned const nbThreads = cctxPool->totalCCtx;
 237         size_t const poolSize = sizeof(*cctxPool)
 238                                 + (nbThreads-1)*sizeof(ZSTD_CCtx*);
 239         unsigned u;
 240         size_t totalCCtxSize = 0;
 241         for (u=0; u<nbThreads; u++) {
 242             totalCCtxSize += ZSTD_sizeof_CCtx(cctxPool->cctx[u]);
 243         }
 244         pthread_mutex_unlock(&cctxPool->poolMutex);
 245         return poolSize + totalCCtxSize;
 246     }
 247 }
 248
 249 static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool)
 250 {
 251     DEBUGLOG(5, "ZSTDMT_getCCtx");
 252     pthread_mutex_lock(&cctxPool->poolMutex);
 253     if (cctxPool->availCCtx) {
 254         cctxPool->availCCtx--;
 255         {   ZSTD_CCtx* const cctx = cctxPool->cctx[cctxPool->availCCtx];
 256             pthread_mutex_unlock(&cctxPool->poolMutex);
 257             return cctx;
 258     }   }
 259     pthread_mutex_unlock(&cctxPool->poolMutex);
 260     DEBUGLOG(5, "create one more CCtx");
 261     return ZSTD_createCCtx_advanced(cctxPool->cMem);   /* note : can be NULL, when creation fails ! */
 262 }
 263
 264 static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
 265 {
 266     if (cctx==NULL) return;   /* compatibility with release on NULL */
 267     pthread_mutex_lock(&pool->poolMutex);
 268     if (pool->availCCtx < pool->totalCCtx)
 269         pool->cctx[pool->availCCtx++] = cctx;
 270     else {
 271         /* pool overflow : should not happen, since totalCCtx==nbThreads */
 272         DEBUGLOG(5, "CCtx pool overflow : free cctx");
 273         ZSTD_freeCCtx(cctx);
 274     }
 275     pthread_mutex_unlock(&pool->poolMutex);
 276 }
 277
 278
 279 /* =====   Thread worker   ===== */
 280
 281 typedef struct {
 282     buffer_t src;
 283     const void* srcStart;
 284     size_t   dictSize;
 285     size_t   srcSize;
 286     buffer_t dstBuff;
 287     size_t   cSize;
 288     size_t   dstFlushed;
 289     unsigned firstChunk;
 290     unsigned lastChunk;
 291     unsigned jobCompleted;
 292     unsigned jobScanned;
 293     pthread_mutex_t* jobCompleted_mutex;
 294     pthread_cond_t* jobCompleted_cond;
 295     ZSTD_parameters params;
 296     const ZSTD_CDict* cdict;
 297     ZSTDMT_CCtxPool* cctxPool;
 298     ZSTDMT_bufferPool* bufPool;
 299     unsigned long long fullFrameSize;
 300 } ZSTDMT_jobDescription;
 301
 302 /* ZSTDMT_compressChunk() : POOL_function type */
 303 void ZSTDMT_compressChunk(void* jobDescription)
 304 {
 305     ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription;
 306     ZSTD_CCtx* cctx = ZSTDMT_getCCtx(job->cctxPool);
 307     const void* const src = (const char*)job->srcStart + job->dictSize;
 308     buffer_t dstBuff = job->dstBuff;
 309     DEBUGLOG(5, "job (first:%u) (last:%u) : dictSize %u, srcSize %u",
 310                  job->firstChunk, job->lastChunk, (U32)job->dictSize, (U32)job->srcSize);
 311
 312     if (cctx==NULL) {
 313         job->cSize = ERROR(memory_allocation);
 314         goto _endJob;
 315     }
 316
 317     if (dstBuff.start == NULL) {
 318         dstBuff = ZSTDMT_getBuffer(job->bufPool);
 319         if (dstBuff.start==NULL) {
 320             job->cSize = ERROR(memory_allocation);
 321             goto _endJob;
 322         }
 323         job->dstBuff = dstBuff;
 324     }
 325
 326     if (job->cdict) {  /* should only happen for first segment */
 327         size_t const initError = ZSTD_compressBegin_usingCDict_advanced(cctx, job->cdict, job->params.fParams, job->fullFrameSize);
 328         DEBUGLOG(5, "using CDict");
 329         if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; }
 330     } else {  /* srcStart points at reloaded section */
 331         if (!job->firstChunk) job->params.fParams.contentSizeFlag = 0;  /* ensure no srcSize control */
 332         {   size_t const dictModeError = ZSTD_setCCtxParameter(cctx, ZSTD_p_forceRawDict, 1);  /* Force loading dictionary in "content-only" mode (no header analysis) */
 333             size_t const initError = ZSTD_compressBegin_advanced(cctx, job->srcStart, job->dictSize, job->params, job->fullFrameSize);
 334             if (ZSTD_isError(initError) || ZSTD_isError(dictModeError)) { job->cSize = initError; goto _endJob; }
 335             ZSTD_setCCtxParameter(cctx, ZSTD_p_forceWindow, 1);
 336     }   }
 337     if (!job->firstChunk) {  /* flush and overwrite frame header when it's not first segment */
 338         size_t const hSize = ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, 0);
 339         if (ZSTD_isError(hSize)) { job->cSize = hSize; goto _endJob; }
 340         ZSTD_invalidateRepCodes(cctx);
 341     }
 342
 343     DEBUGLOG(5, "Compressing : ");
 344     DEBUG_PRINTHEX(4, job->srcStart, 12);
 345     job->cSize = (job->lastChunk) ?
 346                  ZSTD_compressEnd     (cctx, dstBuff.start, dstBuff.size, src, job->srcSize) :
 347                  ZSTD_compressContinue(cctx, dstBuff.start, dstBuff.size, src, job->srcSize);
 348     DEBUGLOG(5, "compressed %u bytes into %u bytes   (first:%u) (last:%u)",
 349                 (unsigned)job->srcSize, (unsigned)job->cSize, job->firstChunk, job->lastChunk);
 350     DEBUGLOG(5, "dstBuff.size : %u ; => %s", (U32)dstBuff.size, ZSTD_getErrorName(job->cSize));
 351
 352 _endJob:
 353     ZSTDMT_releaseCCtx(job->cctxPool, cctx);
 354     ZSTDMT_releaseBuffer(job->bufPool, job->src);
 355     job->src = g_nullBuffer; job->srcStart = NULL;
 356     PTHREAD_MUTEX_LOCK(job->jobCompleted_mutex);
 357     job->jobCompleted = 1;
 358     job->jobScanned = 0;
 359     pthread_cond_signal(job->jobCompleted_cond);
 360     pthread_mutex_unlock(job->jobCompleted_mutex);
 361 }
 362
 363
 364 /* ------------------------------------------ */
 365 /* =====   Multi-threaded compression   ===== */
 366 /* ------------------------------------------ */
 367
 368 typedef struct {
 369     buffer_t buffer;
 370     size_t filled;
 371 } inBuff_t;
 372
 373 struct ZSTDMT_CCtx_s {
 374     POOL_ctx* factory;
 375     ZSTDMT_jobDescription* jobs;
 376     ZSTDMT_bufferPool* bufPool;
 377     ZSTDMT_CCtxPool* cctxPool;
 378     pthread_mutex_t jobCompleted_mutex;
 379     pthread_cond_t jobCompleted_cond;
 380     size_t targetSectionSize;
 381     size_t inBuffSize;
 382     size_t dictSize;
 383     size_t targetDictSize;
 384     inBuff_t inBuff;
 385     ZSTD_parameters params;
 386     XXH64_state_t xxhState;
 387     unsigned nbThreads;
 388     unsigned jobIDMask;
 389     unsigned doneJobID;
 390     unsigned nextJobID;
 391     unsigned frameEnded;
 392     unsigned allJobsCompleted;
 393     unsigned overlapLog;
 394     unsigned long long frameContentSize;
 395     size_t sectionSize;
 396     ZSTD_customMem cMem;
 397     ZSTD_CDict* cdictLocal;
 398     const ZSTD_CDict* cdict;
 399 };
 400
 401 static ZSTDMT_jobDescription* ZSTDMT_allocJobsTable(U32* nbJobsPtr, ZSTD_customMem cMem)
 402 {
 403     U32 const nbJobsLog2 = ZSTD_highbit32(*nbJobsPtr) + 1;
 404     U32 const nbJobs = 1 << nbJobsLog2;
 405     *nbJobsPtr = nbJobs;
 406     return (ZSTDMT_jobDescription*) ZSTD_calloc(
 407                             nbJobs * sizeof(ZSTDMT_jobDescription), cMem);
 408 }
 409
 410 ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbThreads, ZSTD_customMem cMem)
 411 {
 412     ZSTDMT_CCtx* mtctx;
 413     U32 nbJobs = nbThreads + 2;
 414     DEBUGLOG(3, "ZSTDMT_createCCtx_advanced");
 415
 416     if (nbThreads < 1) return NULL;
 417     nbThreads = MIN(nbThreads , ZSTDMT_NBTHREADS_MAX);
 418     if ((cMem.customAlloc!=NULL) ^ (cMem.customFree!=NULL))
 419         /* invalid custom allocator */
 420         return NULL;
 421
 422     mtctx = (ZSTDMT_CCtx*) ZSTD_calloc(sizeof(ZSTDMT_CCtx), cMem);
 423     if (!mtctx) return NULL;
 424     mtctx->cMem = cMem;
 425     mtctx->nbThreads = nbThreads;
 426     mtctx->allJobsCompleted = 1;
 427     mtctx->sectionSize = 0;
 428     mtctx->overlapLog = ZSTDMT_OVERLAPLOG_DEFAULT;
 429     mtctx->factory = POOL_create(nbThreads, 0);
 430     mtctx->jobs = ZSTDMT_allocJobsTable(&nbJobs, cMem);
 431     mtctx->jobIDMask = nbJobs - 1;
 432     mtctx->bufPool = ZSTDMT_createBufferPool(nbThreads, cMem);
 433     mtctx->cctxPool = ZSTDMT_createCCtxPool(nbThreads, cMem);
 434     if (!mtctx->factory | !mtctx->jobs | !mtctx->bufPool | !mtctx->cctxPool) {
 435         ZSTDMT_freeCCtx(mtctx);
 436         return NULL;
 437     }
 438     if (pthread_mutex_init(&mtctx->jobCompleted_mutex, NULL)) {
 439         ZSTDMT_freeCCtx(mtctx);
 440         return NULL;
 441     }
 442     if (pthread_cond_init(&mtctx->jobCompleted_cond, NULL)) {
 443         ZSTDMT_freeCCtx(mtctx);
 444         return NULL;
 445     }
 446     DEBUGLOG(3, "mt_cctx created, for %u threads", nbThreads);
 447     return mtctx;
 448 }
 449
 450 ZSTDMT_CCtx* ZSTDMT_createCCtx(unsigned nbThreads)
 451 {
 452     return ZSTDMT_createCCtx_advanced(nbThreads, ZSTD_defaultCMem);
 453 }
 454
 455 /* ZSTDMT_releaseAllJobResources() :
 456  * note : ensure all workers are killed first ! */
 457 static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx)
 458 {
 459     unsigned jobID;
 460     DEBUGLOG(3, "ZSTDMT_releaseAllJobResources");
 461     for (jobID=0; jobID <= mtctx->jobIDMask; jobID++) {
 462         ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff);
 463         mtctx->jobs[jobID].dstBuff = g_nullBuffer;
 464         ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].src);
 465         mtctx->jobs[jobID].src = g_nullBuffer;
 466     }
 467     memset(mtctx->jobs, 0, (mtctx->jobIDMask+1)*sizeof(ZSTDMT_jobDescription));
 468     ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer);
 469     mtctx->inBuff.buffer = g_nullBuffer;
 470     mtctx->allJobsCompleted = 1;
 471 }
 472
 473 size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx)
 474 {
 475     if (mtctx==NULL) return 0;   /* compatible with free on NULL */
 476     POOL_free(mtctx->factory);
 477     if (!mtctx->allJobsCompleted) ZSTDMT_releaseAllJobResources(mtctx); /* stop workers first */
 478     ZSTDMT_freeBufferPool(mtctx->bufPool);  /* release job resources into pools first */
 479     ZSTD_free(mtctx->jobs, mtctx->cMem);
 480     ZSTDMT_freeCCtxPool(mtctx->cctxPool);
 481     ZSTD_freeCDict(mtctx->cdictLocal);
 482     pthread_mutex_destroy(&mtctx->jobCompleted_mutex);
 483     pthread_cond_destroy(&mtctx->jobCompleted_cond);
 484     ZSTD_free(mtctx, mtctx->cMem);
 485     return 0;
 486 }
 487
 488 size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx)
 489 {
 490     if (mtctx == NULL) return 0;   /* supports sizeof NULL */
 491     return sizeof(*mtctx)
 492             + POOL_sizeof(mtctx->factory)
 493             + ZSTDMT_sizeof_bufferPool(mtctx->bufPool)
 494             + (mtctx->jobIDMask+1) * sizeof(ZSTDMT_jobDescription)
 495             + ZSTDMT_sizeof_CCtxPool(mtctx->cctxPool)
 496             + ZSTD_sizeof_CDict(mtctx->cdictLocal);
 497 }
 498
 499 size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSDTMT_parameter parameter, unsigned value)
 500 {
 501     switch(parameter)
 502     {
 503     case ZSTDMT_p_sectionSize :
 504         mtctx->sectionSize = value;
 505         return 0;
 506     case ZSTDMT_p_overlapSectionLog :
 507         DEBUGLOG(5, "ZSTDMT_p_overlapSectionLog : %u", value);
 508         mtctx->overlapLog = (value >= 9) ? 9 : value;
 509         return 0;
 510     default :
 511         return ERROR(parameter_unsupported);
 512     }
 513 }
 514
 515
 516 /* ------------------------------------------ */
 517 /* =====   Multi-threaded compression   ===== */
 518 /* ------------------------------------------ */
 519
 520 static unsigned computeNbChunks(size_t srcSize, unsigned windowLog, unsigned nbThreads) {
 521     size_t const chunkSizeTarget = (size_t)1 << (windowLog + 2);
 522     size_t const chunkMaxSize = chunkSizeTarget << 2;
 523     size_t const passSizeMax = chunkMaxSize * nbThreads;
 524     unsigned const multiplier = (unsigned)(srcSize / passSizeMax) + 1;
 525     unsigned const nbChunksLarge = multiplier * nbThreads;
 526     unsigned const nbChunksMax = (unsigned)(srcSize / chunkSizeTarget) + 1;
 527     unsigned const nbChunksSmall = MIN(nbChunksMax, nbThreads);
 528     return (multiplier>1) ? nbChunksLarge : nbChunksSmall;
 529 }
 530
 531
 532 size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
 533                                void* dst, size_t dstCapacity,
 534                          const void* src, size_t srcSize,
 535                          const ZSTD_CDict* cdict,
 536                                ZSTD_parameters const params,
 537                                unsigned overlapLog)
 538 {
 539     unsigned const overlapRLog = (overlapLog>9) ? 0 : 9-overlapLog;
 540     size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog);
 541     unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, mtctx->nbThreads);
 542     size_t const proposedChunkSize = (srcSize + (nbChunks-1)) / nbChunks;
 543     size_t const avgChunkSize = ((proposedChunkSize & 0x1FFFF) < 0x7FFF) ? proposedChunkSize + 0xFFFF : proposedChunkSize;   /* avoid too small last block */
 544     const char* const srcStart = (const char*)src;
 545     size_t remainingSrcSize = srcSize;
 546     unsigned const compressWithinDst = (dstCapacity >= ZSTD_compressBound(srcSize)) ? nbChunks : (unsigned)(dstCapacity / ZSTD_compressBound(avgChunkSize));  /* presumes avgChunkSize >= 256 KB, which should be the case */
 547     size_t frameStartPos = 0, dstBufferPos = 0;
 548     XXH64_state_t xxh64;
 549
 550     DEBUGLOG(4, "nbChunks  : %2u   (chunkSize : %u bytes)   ", nbChunks, (U32)avgChunkSize);
 551     if (nbChunks==1) {   /* fallback to single-thread mode */
 552         ZSTD_CCtx* const cctx = mtctx->cctxPool->cctx[0];
 553         if (cdict) return ZSTD_compress_usingCDict_advanced(cctx, dst, dstCapacity, src, srcSize, cdict, params.fParams);
 554         return ZSTD_compress_advanced(cctx, dst, dstCapacity, src, srcSize, NULL, 0, params);
 555     }
 556     assert(avgChunkSize >= 256 KB);  /* condition for ZSTD_compressBound(A) + ZSTD_compressBound(B) <= ZSTD_compressBound(A+B), which is required for compressWithinDst */
 557     ZSTDMT_setBufferSize(mtctx->bufPool, ZSTD_compressBound(avgChunkSize) );
 558     XXH64_reset(&xxh64, 0);
 559
 560     if (nbChunks > mtctx->jobIDMask+1) {  /* enlarge job table */
 561         U32 nbJobs = nbChunks;
 562         ZSTD_free(mtctx->jobs, mtctx->cMem);
 563         mtctx->jobIDMask = 0;
 564         mtctx->jobs = ZSTDMT_allocJobsTable(&nbJobs, mtctx->cMem);
 565         if (mtctx->jobs==NULL) return ERROR(memory_allocation);
 566         mtctx->jobIDMask = nbJobs - 1;
 567     }
 568
 569     {   unsigned u;
 570         for (u=0; u<nbChunks; u++) {
 571             size_t const chunkSize = MIN(remainingSrcSize, avgChunkSize);
 572             size_t const dstBufferCapacity = ZSTD_compressBound(chunkSize);
 573             buffer_t const dstAsBuffer = { (char*)dst + dstBufferPos, dstBufferCapacity };
 574             buffer_t const dstBuffer = u < compressWithinDst ? dstAsBuffer : g_nullBuffer;
 575             size_t dictSize = u ? overlapSize : 0;
 576
 577             mtctx->jobs[u].src = g_nullBuffer;
 578             mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize;
 579             mtctx->jobs[u].dictSize = dictSize;
 580             mtctx->jobs[u].srcSize = chunkSize;
 581             mtctx->jobs[u].cdict = mtctx->nextJobID==0 ? cdict : NULL;
 582             mtctx->jobs[u].fullFrameSize = srcSize;
 583             mtctx->jobs[u].params = params;
 584             /* do not calculate checksum within sections, but write it in header for first section */
 585             if (u!=0) mtctx->jobs[u].params.fParams.checksumFlag = 0;
 586             mtctx->jobs[u].dstBuff = dstBuffer;
 587             mtctx->jobs[u].cctxPool = mtctx->cctxPool;
 588             mtctx->jobs[u].bufPool = mtctx->bufPool;
 589             mtctx->jobs[u].firstChunk = (u==0);
 590             mtctx->jobs[u].lastChunk = (u==nbChunks-1);
 591             mtctx->jobs[u].jobCompleted = 0;
 592             mtctx->jobs[u].jobCompleted_mutex = &mtctx->jobCompleted_mutex;
 593             mtctx->jobs[u].jobCompleted_cond = &mtctx->jobCompleted_cond;
 594
 595             if (params.fParams.checksumFlag) {
 596                 XXH64_update(&xxh64, srcStart + frameStartPos, chunkSize);
 597             }
 598
 599             DEBUGLOG(5, "posting job %u   (%u bytes)", u, (U32)chunkSize);
 600             DEBUG_PRINTHEX(6, mtctx->jobs[u].srcStart, 12);
 601             POOL_add(mtctx->factory, ZSTDMT_compressChunk, &mtctx->jobs[u]);
 602
 603             frameStartPos += chunkSize;
 604             dstBufferPos += dstBufferCapacity;
 605             remainingSrcSize -= chunkSize;
 606     }   }
 607
 608     /* collect result */
 609     {   size_t error = 0, dstPos = 0;
 610         unsigned chunkID;
 611         for (chunkID=0; chunkID<nbChunks; chunkID++) {
 612             DEBUGLOG(5, "waiting for chunk %u ", chunkID);
 613             PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex);
 614             while (mtctx->jobs[chunkID].jobCompleted==0) {
 615                 DEBUGLOG(5, "waiting for jobCompleted signal from chunk %u", chunkID);
 616                 pthread_cond_wait(&mtctx->jobCompleted_cond, &mtctx->jobCompleted_mutex);
 617             }
 618             pthread_mutex_unlock(&mtctx->jobCompleted_mutex);
 619             DEBUGLOG(5, "ready to write chunk %u ", chunkID);
 620
 621             mtctx->jobs[chunkID].srcStart = NULL;
 622             {   size_t const cSize = mtctx->jobs[chunkID].cSize;
 623                 if (ZSTD_isError(cSize)) error = cSize;
 624                 if ((!error) && (dstPos + cSize > dstCapacity)) error = ERROR(dstSize_tooSmall);
 625                 if (chunkID) {   /* note : chunk 0 is written directly at dst, which is correct position */
 626                     if (!error)
 627                         memmove((char*)dst + dstPos, mtctx->jobs[chunkID].dstBuff.start, cSize);  /* may overlap when chunk compressed within dst */
 628                     if (chunkID >= compressWithinDst) {  /* chunk compressed into its own buffer, which must be released */
 629                         DEBUGLOG(5, "releasing buffer %u>=%u", chunkID, compressWithinDst);
 630                         ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[chunkID].dstBuff);
 631                     }
 632                     mtctx->jobs[chunkID].dstBuff = g_nullBuffer;
 633                 }
 634                 dstPos += cSize ;
 635             }
 636         }  /* for (chunkID=0; chunkID<nbChunks; chunkID++) */
 637
 638         DEBUGLOG(4, "checksumFlag : %u ", params.fParams.checksumFlag);
 639         if (params.fParams.checksumFlag) {
 640             U32 const checksum = (U32)XXH64_digest(&xxh64);
 641             if (dstPos + 4 > dstCapacity) {
 642                 error = ERROR(dstSize_tooSmall);
 643             } else {
 644                 DEBUGLOG(4, "writing checksum : %08X \n", checksum);
 645                 MEM_writeLE32((char*)dst + dstPos, checksum);
 646                 dstPos += 4;
 647         }   }
 648
 649         if (!error) DEBUGLOG(4, "compressed size : %u  ", (U32)dstPos);
 650         return error ? error : dstPos;
 651     }
 652 }
 653
 654
 655 size_t ZSTDMT_compressCCtx(ZSTDMT_CCtx* mtctx,
 656                            void* dst, size_t dstCapacity,
 657                      const void* src, size_t srcSize,
 658                            int compressionLevel)
 659 {
 660     U32 const overlapLog = (compressionLevel >= ZSTD_maxCLevel()) ? 9 : ZSTDMT_OVERLAPLOG_DEFAULT;
 661     ZSTD_parameters params = ZSTD_getParams(compressionLevel, srcSize, 0);
 662     params.fParams.contentSizeFlag = 1;
 663     return ZSTDMT_compress_advanced(mtctx, dst, dstCapacity, src, srcSize, NULL, params, overlapLog);
 664 }
 665
 666
 667 /* ====================================== */
 668 /* =======      Streaming API     ======= */
 669 /* ====================================== */
 670
 671 static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* zcs)
 672 {
 673     DEBUGLOG(4, "ZSTDMT_waitForAllJobsCompleted");
 674     while (zcs->doneJobID < zcs->nextJobID) {
 675         unsigned const jobID = zcs->doneJobID & zcs->jobIDMask;
 676         PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
 677         while (zcs->jobs[jobID].jobCompleted==0) {
 678             DEBUGLOG(5, "waiting for jobCompleted signal from chunk %u", zcs->doneJobID);   /* we want to block when waiting for data to flush */
 679             pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex);
 680         }
 681         pthread_mutex_unlock(&zcs->jobCompleted_mutex);
 682         zcs->doneJobID++;
 683     }
 684 }
 685
 686
 687 /** ZSTDMT_initCStream_internal() :
 688  *  internal usage only */
 689 size_t ZSTDMT_initCStream_internal(ZSTDMT_CCtx* zcs,
 690                     const void* dict, size_t dictSize, const ZSTD_CDict* cdict,
 691                     ZSTD_parameters params, unsigned long long pledgedSrcSize)
 692 {
 693     DEBUGLOG(4, "ZSTDMT_initCStream_internal");
 694     /* params are supposed to be fully validated at this point */
 695     assert(!ZSTD_isError(ZSTD_checkCParams(params.cParams)));
 696     assert(!((dict) && (cdict)));  /* either dict or cdict, not both */
 697
 698     if (zcs->nbThreads==1) {
 699         DEBUGLOG(4, "single thread mode");
 700         return ZSTD_initCStream_internal(zcs->cctxPool->cctx[0],
 701                                         dict, dictSize, cdict,
 702                                         params, pledgedSrcSize);
 703     }
 704
 705     if (zcs->allJobsCompleted == 0) {   /* previous compression not correctly finished */
 706         ZSTDMT_waitForAllJobsCompleted(zcs);
 707         ZSTDMT_releaseAllJobResources(zcs);
 708         zcs->allJobsCompleted = 1;
 709     }
 710
 711     zcs->params = params;
 712     zcs->frameContentSize = pledgedSrcSize;
 713     if (dict) {
 714         DEBUGLOG(4,"cdictLocal: %08X", (U32)(size_t)zcs->cdictLocal);
 715         ZSTD_freeCDict(zcs->cdictLocal);
 716         zcs->cdictLocal = ZSTD_createCDict_advanced(dict, dictSize,
 717                                                     0 /* byRef */, ZSTD_dm_auto,   /* note : a loadPrefix becomes an internal CDict */
 718                                                     params.cParams, zcs->cMem);
 719         zcs->cdict = zcs->cdictLocal;
 720         if (zcs->cdictLocal == NULL) return ERROR(memory_allocation);
 721     } else {
 722         DEBUGLOG(4,"cdictLocal: %08X", (U32)(size_t)zcs->cdictLocal);
 723         ZSTD_freeCDict(zcs->cdictLocal);
 724         zcs->cdictLocal = NULL;
 725         zcs->cdict = cdict;
 726     }
 727
 728     zcs->targetDictSize = (zcs->overlapLog==0) ? 0 : (size_t)1 << (zcs->params.cParams.windowLog - (9 - zcs->overlapLog));
 729     DEBUGLOG(4, "overlapLog : %u ", zcs->overlapLog);
 730     DEBUGLOG(4, "overlap Size : %u KB", (U32)(zcs->targetDictSize>>10));
 731     zcs->targetSectionSize = zcs->sectionSize ? zcs->sectionSize : (size_t)1 << (zcs->params.cParams.windowLog + 2);
 732     zcs->targetSectionSize = MAX(ZSTDMT_SECTION_SIZE_MIN, zcs->targetSectionSize);
 733     zcs->targetSectionSize = MAX(zcs->targetDictSize, zcs->targetSectionSize);
 734     DEBUGLOG(4, "Section Size : %u KB", (U32)(zcs->targetSectionSize>>10));
 735     zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize;
 736     ZSTDMT_setBufferSize(zcs->bufPool, MAX(zcs->inBuffSize, ZSTD_compressBound(zcs->targetSectionSize)) );
 737     zcs->inBuff.buffer = g_nullBuffer;
 738     zcs->dictSize = 0;
 739     zcs->doneJobID = 0;
 740     zcs->nextJobID = 0;
 741     zcs->frameEnded = 0;
 742     zcs->allJobsCompleted = 0;
 743     if (params.fParams.checksumFlag) XXH64_reset(&zcs->xxhState, 0);
 744     return 0;
 745 }
 746
 747 size_t ZSTDMT_initCStream_advanced(ZSTDMT_CCtx* mtctx,
 748                              const void* dict, size_t dictSize,
 749                                    ZSTD_parameters params,
 750                                    unsigned long long pledgedSrcSize)
 751 {
 752     DEBUGLOG(5, "ZSTDMT_initCStream_advanced");
 753     return ZSTDMT_initCStream_internal(mtctx, dict, dictSize, NULL, params, pledgedSrcSize);
 754 }
 755
 756 size_t ZSTDMT_initCStream_usingCDict(ZSTDMT_CCtx* mtctx,
 757                                const ZSTD_CDict* cdict,
 758                                      ZSTD_frameParameters fParams,
 759                                      unsigned long long pledgedSrcSize)
 760 {
 761     ZSTD_parameters params = ZSTD_getParamsFromCDict(cdict);
 762     if (cdict==NULL) return ERROR(dictionary_wrong);   /* method incompatible with NULL cdict */
 763     params.fParams = fParams;
 764     return ZSTDMT_initCStream_internal(mtctx, NULL, 0 /*dictSize*/, cdict,
 765                                         params, pledgedSrcSize);
 766 }
 767
 768
 769 /* ZSTDMT_resetCStream() :
 770  * pledgedSrcSize is optional and can be zero == unknown */
 771 size_t ZSTDMT_resetCStream(ZSTDMT_CCtx* zcs, unsigned long long pledgedSrcSize)
 772 {
 773     if (zcs->nbThreads==1)
 774         return ZSTD_resetCStream(zcs->cctxPool->cctx[0], pledgedSrcSize);
 775     return ZSTDMT_initCStream_internal(zcs, NULL, 0, 0, zcs->params, pledgedSrcSize);
 776 }
 777
 778 size_t ZSTDMT_initCStream(ZSTDMT_CCtx* zcs, int compressionLevel) {
 779     ZSTD_parameters const params = ZSTD_getParams(compressionLevel, 0, 0);
 780     return ZSTDMT_initCStream_internal(zcs, NULL, 0, NULL, params, 0);
 781 }
 782
 783
 784 static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsigned endFrame)
 785 {
 786     unsigned const jobID = zcs->nextJobID & zcs->jobIDMask;
 787
 788     DEBUGLOG(4, "preparing job %u to compress %u bytes with %u preload ",
 789                 zcs->nextJobID, (U32)srcSize, (U32)zcs->dictSize);
 790     zcs->jobs[jobID].src = zcs->inBuff.buffer;
 791     zcs->jobs[jobID].srcStart = zcs->inBuff.buffer.start;
 792     zcs->jobs[jobID].srcSize = srcSize;
 793     zcs->jobs[jobID].dictSize = zcs->dictSize;
 794     assert(zcs->inBuff.filled >= srcSize + zcs->dictSize);
 795     zcs->jobs[jobID].params = zcs->params;
 796     /* do not calculate checksum within sections, but write it in header for first section */
 797     if (zcs->nextJobID) zcs->jobs[jobID].params.fParams.checksumFlag = 0;
 798     zcs->jobs[jobID].cdict = zcs->nextJobID==0 ? zcs->cdict : NULL;
 799     zcs->jobs[jobID].fullFrameSize = zcs->frameContentSize;
 800     zcs->jobs[jobID].dstBuff = g_nullBuffer;
 801     zcs->jobs[jobID].cctxPool = zcs->cctxPool;
 802     zcs->jobs[jobID].bufPool = zcs->bufPool;
 803     zcs->jobs[jobID].firstChunk = (zcs->nextJobID==0);
 804     zcs->jobs[jobID].lastChunk = endFrame;
 805     zcs->jobs[jobID].jobCompleted = 0;
 806     zcs->jobs[jobID].dstFlushed = 0;
 807     zcs->jobs[jobID].jobCompleted_mutex = &zcs->jobCompleted_mutex;
 808     zcs->jobs[jobID].jobCompleted_cond = &zcs->jobCompleted_cond;
 809
 810     if (zcs->params.fParams.checksumFlag)
 811         XXH64_update(&zcs->xxhState, (const char*)zcs->inBuff.buffer.start + zcs->dictSize, srcSize);
 812
 813     /* get a new buffer for next input */
 814     if (!endFrame) {
 815         size_t const newDictSize = MIN(srcSize + zcs->dictSize, zcs->targetDictSize);
 816         zcs->inBuff.buffer = ZSTDMT_getBuffer(zcs->bufPool);
 817         if (zcs->inBuff.buffer.start == NULL) {   /* not enough memory to allocate next input buffer */
 818             zcs->jobs[jobID].jobCompleted = 1;
 819             zcs->nextJobID++;
 820             ZSTDMT_waitForAllJobsCompleted(zcs);
 821             ZSTDMT_releaseAllJobResources(zcs);
 822             return ERROR(memory_allocation);
 823         }
 824         zcs->inBuff.filled -= srcSize + zcs->dictSize - newDictSize;
 825         memmove(zcs->inBuff.buffer.start,
 826             (const char*)zcs->jobs[jobID].srcStart + zcs->dictSize + srcSize - newDictSize,
 827             zcs->inBuff.filled);
 828         zcs->dictSize = newDictSize;
 829     } else {   /* if (endFrame==1) */
 830         zcs->inBuff.buffer = g_nullBuffer;
 831         zcs->inBuff.filled = 0;
 832         zcs->dictSize = 0;
 833         zcs->frameEnded = 1;
 834         if (zcs->nextJobID == 0) {
 835             /* single chunk exception : checksum is calculated directly within worker thread */
 836             zcs->params.fParams.checksumFlag = 0;
 837     }   }
 838
 839     DEBUGLOG(4, "posting job %u : %u bytes  (end:%u) (note : doneJob = %u=>%u)",
 840                 zcs->nextJobID,
 841                 (U32)zcs->jobs[jobID].srcSize,
 842                 zcs->jobs[jobID].lastChunk,
 843                 zcs->doneJobID,
 844                 zcs->doneJobID & zcs->jobIDMask);
 845     POOL_add(zcs->factory, ZSTDMT_compressChunk, &zcs->jobs[jobID]);   /* this call is blocking when thread worker pool is exhausted */
 846     zcs->nextJobID++;
 847     return 0;
 848 }
 849
 850
 851 /* ZSTDMT_flushNextJob() :
 852  * output : will be updated with amount of data flushed .
 853  * blockToFlush : if >0, the function will block and wait if there is no data available to flush .
 854  * @return : amount of data remaining within internal buffer, 1 if unknown but > 0, 0 if no more, or an error code */
 855 static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsigned blockToFlush)
 856 {
 857     unsigned const wJobID = zcs->doneJobID & zcs->jobIDMask;
 858     if (zcs->doneJobID == zcs->nextJobID) return 0;   /* all flushed ! */
 859     PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
 860     while (zcs->jobs[wJobID].jobCompleted==0) {
 861         DEBUGLOG(5, "waiting for jobCompleted signal from job %u", zcs->doneJobID);
 862         if (!blockToFlush) { pthread_mutex_unlock(&zcs->jobCompleted_mutex); return 0; }  /* nothing ready to be flushed => skip */
 863         pthread_cond_wait(&zcs->jobCompleted_cond, &zcs->jobCompleted_mutex);  /* block when nothing available to flush */
 864     }
 865     pthread_mutex_unlock(&zcs->jobCompleted_mutex);
 866     /* compression job completed : output can be flushed */
 867     {   ZSTDMT_jobDescription job = zcs->jobs[wJobID];
 868         if (!job.jobScanned) {
 869             if (ZSTD_isError(job.cSize)) {
 870                 DEBUGLOG(5, "compression error detected ");
 871                 ZSTDMT_waitForAllJobsCompleted(zcs);
 872                 ZSTDMT_releaseAllJobResources(zcs);
 873                 return job.cSize;
 874             }
 875             DEBUGLOG(5, "zcs->params.fParams.checksumFlag : %u ", zcs->params.fParams.checksumFlag);
 876             if (zcs->params.fParams.checksumFlag) {
 877                 if (zcs->frameEnded && (zcs->doneJobID+1 == zcs->nextJobID)) {  /* write checksum at end of last section */
 878                     U32 const checksum = (U32)XXH64_digest(&zcs->xxhState);
 879                     DEBUGLOG(5, "writing checksum : %08X \n", checksum);
 880                     MEM_writeLE32((char*)job.dstBuff.start + job.cSize, checksum);
 881                     job.cSize += 4;
 882                     zcs->jobs[wJobID].cSize += 4;
 883             }   }
 884             zcs->jobs[wJobID].jobScanned = 1;
 885         }
 886         {   size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos);
 887             DEBUGLOG(5, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID);
 888             memcpy((char*)output->dst + output->pos, (const char*)job.dstBuff.start + job.dstFlushed, toWrite);
 889             output->pos += toWrite;
 890             job.dstFlushed += toWrite;
 891         }
 892         if (job.dstFlushed == job.cSize) {   /* output buffer fully flushed => move to next one */
 893             ZSTDMT_releaseBuffer(zcs->bufPool, job.dstBuff);
 894             zcs->jobs[wJobID].dstBuff = g_nullBuffer;
 895             zcs->jobs[wJobID].jobCompleted = 0;
 896             zcs->doneJobID++;
 897         } else {
 898             zcs->jobs[wJobID].dstFlushed = job.dstFlushed;
 899         }
 900         /* return value : how many bytes left in buffer ; fake it to 1 if unknown but >0 */
 901         if (job.cSize > job.dstFlushed) return (job.cSize - job.dstFlushed);
 902         if (zcs->doneJobID < zcs->nextJobID) return 1;   /* still some buffer to flush */
 903         zcs->allJobsCompleted = zcs->frameEnded;   /* frame completed and entirely flushed */
 904         return 0;   /* everything flushed */
 905 }   }
 906
 907
 908 /** ZSTDMT_compressStream_generic() :
 909  *  internal use only
 910  *  assumption : output and input are valid (pos <= size)
 911  * @return : minimum amount of data remaining to flush, 0 if none */
 912 size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx,
 913                                      ZSTD_outBuffer* output,
 914                                      ZSTD_inBuffer* input,
 915                                      ZSTD_EndDirective endOp)
 916 {
 917     size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize;
 918     assert(output->pos <= output->size);
 919     assert(input->pos  <= input->size);
 920     if ((mtctx->frameEnded) && (endOp==ZSTD_e_continue)) {
 921         /* current frame being ended. Only flush/end are allowed. Or start new frame with init */
 922         return ERROR(stage_wrong);
 923     }
 924     if (mtctx->nbThreads==1) {  /* delegate to single-thread (synchronous) */
 925         return ZSTD_compressStream_generic(mtctx->cctxPool->cctx[0], output, input, endOp);
 926     }
 927
 928     /* single-pass shortcut (note : this is synchronous-mode) */
 929     if ( (mtctx->nextJobID==0)      /* just started */
 930       && (mtctx->inBuff.filled==0)  /* nothing buffered */
 931       && (endOp==ZSTD_e_end)        /* end order */
 932       && (output->size - output->pos >= ZSTD_compressBound(input->size - input->pos)) ) { /* enough room */
 933         size_t const cSize = ZSTDMT_compress_advanced(mtctx,
 934                 (char*)output->dst + output->pos, output->size - output->pos,
 935                 (const char*)input->src + input->pos, input->size - input->pos,
 936                 mtctx->cdict, mtctx->params, mtctx->overlapLog);
 937         if (ZSTD_isError(cSize)) return cSize;
 938         input->pos = input->size;
 939         output->pos += cSize;
 940         ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->inBuff.buffer);  /* was allocated in initStream */
 941         mtctx->allJobsCompleted = 1;
 942         mtctx->frameEnded = 1;
 943         return 0;
 944     }
 945
 946     /* fill input buffer */
 947     if (input->size > input->pos) {   /* support NULL input */
 948         if (mtctx->inBuff.buffer.start == NULL) {
 949             mtctx->inBuff.buffer = ZSTDMT_getBuffer(mtctx->bufPool);
 950             if (mtctx->inBuff.buffer.start == NULL) return ERROR(memory_allocation);
 951             mtctx->inBuff.filled = 0;
 952         }
 953         {   size_t const toLoad = MIN(input->size - input->pos, mtctx->inBuffSize - mtctx->inBuff.filled);
 954             DEBUGLOG(5, "inBuff:%08X;  inBuffSize=%u;  ToCopy=%u", (U32)(size_t)mtctx->inBuff.buffer.start, (U32)mtctx->inBuffSize, (U32)toLoad);
 955             memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, (const char*)input->src + input->pos, toLoad);
 956             input->pos += toLoad;
 957             mtctx->inBuff.filled += toLoad;
 958     }   }
 959
 960     if ( (mtctx->inBuff.filled >= newJobThreshold)  /* filled enough : let's compress */
 961       && (mtctx->nextJobID <= mtctx->doneJobID + mtctx->jobIDMask) ) {   /* avoid overwriting job round buffer */
 962         CHECK_F( ZSTDMT_createCompressionJob(mtctx, mtctx->targetSectionSize, 0 /* endFrame */) );
 963     }
 964
 965     /* check for potential compressed data ready to be flushed */
 966     CHECK_F( ZSTDMT_flushNextJob(mtctx, output, (mtctx->inBuff.filled == mtctx->inBuffSize) /* blockToFlush */) ); /* block if it wasn't possible to create new job due to saturation */
 967
 968     if (input->pos < input->size)  /* input not consumed : do not flush yet */
 969         endOp = ZSTD_e_continue;
 970
 971     switch(endOp)
 972     {
 973         case ZSTD_e_flush:
 974             return ZSTDMT_flushStream(mtctx, output);
 975         case ZSTD_e_end:
 976             return ZSTDMT_endStream(mtctx, output);
 977         case ZSTD_e_continue:
 978             return 1;
 979         default:
 980             return ERROR(GENERIC);   /* invalid endDirective */
 981     }
 982 }
 983
 984
 985 size_t ZSTDMT_compressStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, ZSTD_inBuffer* input)
 986 {
 987     CHECK_F( ZSTDMT_compressStream_generic(zcs, output, input, ZSTD_e_continue) );
 988
 989     /* recommended next input size : fill current input buffer */
 990     return zcs->inBuffSize - zcs->inBuff.filled;   /* note : could be zero when input buffer is fully filled and no more availability to create new job */
 991 }
 992
 993
 994 static size_t ZSTDMT_flushStream_internal(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsigned endFrame)
 995 {
 996     size_t const srcSize = zcs->inBuff.filled - zcs->dictSize;
 997
 998     if ( ((srcSize > 0) || (endFrame && !zcs->frameEnded))
 999        && (zcs->nextJobID <= zcs->doneJobID + zcs->jobIDMask) ) {
1000         CHECK_F( ZSTDMT_createCompressionJob(zcs, srcSize, endFrame) );
1001     }
1002
1003     /* check if there is any data available to flush */
1004     return ZSTDMT_flushNextJob(zcs, output, 1 /* blockToFlush */);
1005 }
1006
1007
1008 size_t ZSTDMT_flushStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output)
1009 {
1010     DEBUGLOG(5, "ZSTDMT_flushStream");
1011     if (zcs->nbThreads==1)
1012         return ZSTD_flushStream(zcs->cctxPool->cctx[0], output);
1013     return ZSTDMT_flushStream_internal(zcs, output, 0 /* endFrame */);
1014 }
1015
1016 size_t ZSTDMT_endStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output)
1017 {
1018     DEBUGLOG(4, "ZSTDMT_endStream");
1019     if (zcs->nbThreads==1)
1020         return ZSTD_endStream(zcs->cctxPool->cctx[0], output);
1021     return ZSTDMT_flushStream_internal(zcs, output, 1 /* endFrame */);
1022 }