Merge pull request #947 from facebook/fix944

Fix #944
2017-12-14 10:01:52 -08:00
parent 4d0dfafa7b d23eb9a098
commit c005df136f
9 changed files with 196 additions and 103 deletions
@@ -410,7 +410,7 @@ size_t ZSTD_CCtxParam_setParameter(
        return ERROR(parameter_unsupported);
 #else
        if (CCtxParams->nbThreads <= 1) return ERROR(parameter_unsupported);
-        return ZSTDMT_CCtxParam_setMTCtxParameter(CCtxParams, ZSTDMT_p_sectionSize, value);
+        return ZSTDMT_CCtxParam_setMTCtxParameter(CCtxParams, ZSTDMT_p_jobSize, value);
 #endif

    case ZSTD_p_overlapSizeLog :
@@ -477,7 +477,7 @@ size_t ZSTD_CCtx_setParametersUsingCCtxParams(

 ZSTDLIB_API size_t ZSTD_CCtx_setPledgedSrcSize(ZSTD_CCtx* cctx, unsigned long long pledgedSrcSize)
 {
-    DEBUGLOG(4, " setting pledgedSrcSize to %u", (U32)pledgedSrcSize);
+    DEBUGLOG(4, "ZSTD_CCtx_setPledgedSrcSize to %u bytes", (U32)pledgedSrcSize);
    if (cctx->streamStage != zcss_init) return ERROR(stage_wrong);
    cctx->pledgedSrcSizePlusOne = pledgedSrcSize+1;
    return 0;
@@ -489,7 +489,7 @@ size_t ZSTD_CCtx_loadDictionary_advanced(
 {
    if (cctx->streamStage != zcss_init) return ERROR(stage_wrong);
    if (cctx->staticSize) return ERROR(memory_allocation);  /* no malloc for static CCtx */
-    DEBUGLOG(4, "load dictionary of size %u", (U32)dictSize);
+    DEBUGLOG(4, "ZSTD_CCtx_loadDictionary_advanced (size: %u)", (U32)dictSize);
    ZSTD_freeCDict(cctx->cdictLocal);  /* in case one already exists */
    if (dict==NULL || dictSize==0) {   /* no dictionary mode */
        cctx->cdictLocal = NULL;
@@ -987,15 +987,16 @@ void ZSTD_invalidateRepCodes(ZSTD_CCtx* cctx) {

 /*! ZSTD_copyCCtx_internal() :
 *  Duplicate an existing context `srcCCtx` into another one `dstCCtx`.
- *  The "context", in this case, refers to the hash and chain tables,
- *  entropy tables, and dictionary offsets.
 *  Only works during stage ZSTDcs_init (i.e. after creation, but before first call to ZSTD_compressContinue()).
- *  pledgedSrcSize=0 means "empty".
- *  @return : 0, or an error code */
+ *  The "context", in this case, refers to the hash and chain tables,
+ *  entropy tables, and dictionary references.
+ * `windowLog` value is enforced if != 0, otherwise value is copied from srcCCtx.
+ * @return : 0, or an error code */
 static size_t ZSTD_copyCCtx_internal(ZSTD_CCtx* dstCCtx,
                            const ZSTD_CCtx* srcCCtx,
+                            unsigned windowLog,
                            ZSTD_frameParameters fParams,
-                            unsigned long long pledgedSrcSize,
+                            U64 pledgedSrcSize,
                            ZSTD_buffered_policy_e zbuff)
 {
    DEBUGLOG(5, "ZSTD_copyCCtx_internal");
@@ -1005,6 +1006,7 @@ static size_t ZSTD_copyCCtx_internal(ZSTD_CCtx* dstCCtx,
    {   ZSTD_CCtx_params params = dstCCtx->requestedParams;
        /* Copy only compression parameters related to tables. */
        params.cParams = srcCCtx->appliedParams.cParams;
+        if (windowLog) params.cParams.windowLog = windowLog;
        params.fParams = fParams;
        ZSTD_resetCCtx_internal(dstCCtx, params, pledgedSrcSize,
                                ZSTDcrp_noMemset, zbuff);
@@ -1056,7 +1058,9 @@ size_t ZSTD_copyCCtx(ZSTD_CCtx* dstCCtx, const ZSTD_CCtx* srcCCtx, unsigned long
    if (pledgedSrcSize==0) pledgedSrcSize = ZSTD_CONTENTSIZE_UNKNOWN;
    fParams.contentSizeFlag = (pledgedSrcSize != ZSTD_CONTENTSIZE_UNKNOWN);

-    return ZSTD_copyCCtx_internal(dstCCtx, srcCCtx, fParams, pledgedSrcSize, zbuff);
+    return ZSTD_copyCCtx_internal(dstCCtx, srcCCtx,
+                                0 /*windowLog from srcCCtx*/, fParams, pledgedSrcSize,
+                                zbuff);
 }


@@ -2043,12 +2047,12 @@ static size_t ZSTD_compress_insertDictionary(ZSTD_CCtx* cctx,

 /*! ZSTD_compressBegin_internal() :
 * @return : 0, or an error code */
-static size_t ZSTD_compressBegin_internal(ZSTD_CCtx* cctx,
+size_t ZSTD_compressBegin_internal(ZSTD_CCtx* cctx,
                             const void* dict, size_t dictSize,
                             ZSTD_dictMode_e dictMode,
                             const ZSTD_CDict* cdict,
-                                   ZSTD_CCtx_params params, U64 pledgedSrcSize,
-                                   ZSTD_buffered_policy_e zbuff)
+                             ZSTD_CCtx_params params, U64 pledgedSrcSize,
+                             ZSTD_buffered_policy_e zbuff)
 {
    DEBUGLOG(4, "ZSTD_compressBegin_internal");
    /* params are supposed to be fully validated at this point */
@@ -2058,7 +2062,7 @@ static size_t ZSTD_compressBegin_internal(ZSTD_CCtx* cctx,
    if (cdict && cdict->dictContentSize>0) {
        cctx->requestedParams = params;
        return ZSTD_copyCCtx_internal(cctx, cdict->refContext,
-                                      params.fParams, pledgedSrcSize,
+                                      params.cParams.windowLog, params.fParams, pledgedSrcSize,
                                      zbuff);
    }

@@ -2067,17 +2071,19 @@ static size_t ZSTD_compressBegin_internal(ZSTD_CCtx* cctx,
    return ZSTD_compress_insertDictionary(cctx, dict, dictSize, dictMode);
 }

-size_t ZSTD_compressBegin_advanced_internal(
-                                    ZSTD_CCtx* cctx,
+size_t ZSTD_compressBegin_advanced_internal(ZSTD_CCtx* cctx,
                                    const void* dict, size_t dictSize,
                                    ZSTD_dictMode_e dictMode,
+                                    const ZSTD_CDict* cdict,
                                    ZSTD_CCtx_params params,
                                    unsigned long long pledgedSrcSize)
 {
    DEBUGLOG(4, "ZSTD_compressBegin_advanced_internal");
    /* compression parameters verification and optimization */
    CHECK_F( ZSTD_checkCParams(params.cParams) );
-    return ZSTD_compressBegin_internal(cctx, dict, dictSize, dictMode, NULL,
+    return ZSTD_compressBegin_internal(cctx,
+                                       dict, dictSize, dictMode,
+                                       cdict,
                                       params, pledgedSrcSize,
                                       ZSTDb_not_buffered);
 }
@@ -2090,9 +2096,10 @@ size_t ZSTD_compressBegin_advanced(ZSTD_CCtx* cctx,
 {
    ZSTD_CCtx_params const cctxParams =
            ZSTD_assignParamsToCCtxParams(cctx->requestedParams, params);
-    return ZSTD_compressBegin_advanced_internal(cctx, dict, dictSize, ZSTD_dm_auto,
-                                                cctxParams,
-                                                pledgedSrcSize);
+    return ZSTD_compressBegin_advanced_internal(cctx,
+                                            dict, dictSize, ZSTD_dm_auto,
+                                            NULL /*cdict*/,
+                                            cctxParams, pledgedSrcSize);
 }

 size_t ZSTD_compressBegin_usingDict(ZSTD_CCtx* cctx, const void* dict, size_t dictSize, int compressionLevel)
@@ -2279,7 +2286,7 @@ static size_t ZSTD_initCDict_internal(
                    ZSTD_dictMode_e dictMode,
                    ZSTD_compressionParameters cParams)
 {
-    DEBUGLOG(4, "ZSTD_initCDict_internal, mode %u", (U32)dictMode);
+    DEBUGLOG(3, "ZSTD_initCDict_internal, mode %u", (U32)dictMode);
    if ((dictLoadMethod == ZSTD_dlm_byRef) || (!dictBuffer) || (!dictSize)) {
        cdict->dictBuffer = NULL;
        cdict->dictContent = dictBuffer;
@@ -2309,7 +2316,7 @@ ZSTD_CDict* ZSTD_createCDict_advanced(const void* dictBuffer, size_t dictSize,
                                      ZSTD_dictMode_e dictMode,
                                      ZSTD_compressionParameters cParams, ZSTD_customMem customMem)
 {
-    DEBUGLOG(4, "ZSTD_createCDict_advanced, mode %u", (U32)dictMode);
+    DEBUGLOG(3, "ZSTD_createCDict_advanced, mode %u", (U32)dictMode);
    if (!customMem.customAlloc ^ !customMem.customFree) return NULL;

    {   ZSTD_CDict* const cdict = (ZSTD_CDict*)ZSTD_malloc(sizeof(ZSTD_CDict), customMem);
@@ -2513,10 +2520,10 @@ static size_t ZSTD_resetCStream_internal(ZSTD_CStream* zcs,
    assert(!((dict) && (cdict)));  /* either dict or cdict, not both */

    CHECK_F( ZSTD_compressBegin_internal(zcs,
-                                        dict, dictSize, dictMode,
-                                        cdict,
-                                        params, pledgedSrcSize,
-                                        ZSTDb_buffered) );
+                                         dict, dictSize, dictMode,
+                                         cdict,
+                                         params, pledgedSrcSize,
+                                         ZSTDb_buffered) );

    zcs->inToCompress = 0;
    zcs->inBuffPos = 0;
@@ -2540,7 +2547,7 @@ size_t ZSTD_resetCStream(ZSTD_CStream* zcs, unsigned long long pledgedSrcSize)
 }

 /*! ZSTD_initCStream_internal() :
- *  Note : not static, but hidden (not exposed). Used by zstdmt_compress.c
+ *  Note : for lib/compress only. Used by zstdmt_compress.c.
 *  Assumption 1 : params are valid
 *  Assumption 2 : either dict, or cdict, is defined, not both */
 size_t ZSTD_initCStream_internal(ZSTD_CStream* zcs,
@@ -2552,7 +2559,7 @@ size_t ZSTD_initCStream_internal(ZSTD_CStream* zcs,
    assert(!((dict) && (cdict)));  /* either dict or cdict, not both */

    if (dict && dictSize >= 8) {
-        DEBUGLOG(5, "loading dictionary of size %u", (U32)dictSize);
+        DEBUGLOG(4, "loading dictionary of size %u", (U32)dictSize);
        if (zcs->staticSize) {   /* static CCtx : never uses malloc */
            /* incompatible with internal cdict creation */
            return ERROR(memory_allocation);
@@ -2565,14 +2572,14 @@ size_t ZSTD_initCStream_internal(ZSTD_CStream* zcs,
        if (zcs->cdictLocal == NULL) return ERROR(memory_allocation);
    } else {
        if (cdict) {
-            params.cParams = ZSTD_getCParamsFromCDict(cdict);  /* cParams are enforced from cdict */
+            params.cParams = ZSTD_getCParamsFromCDict(cdict);  /* cParams are enforced from cdict; it includes windowLog */
        }
        ZSTD_freeCDict(zcs->cdictLocal);
        zcs->cdictLocal = NULL;
        zcs->cdict = cdict;
    }

-    params.compressionLevel = ZSTD_CLEVEL_CUSTOM;
+    params.compressionLevel = ZSTD_CLEVEL_CUSTOM; /* enforce usage of cParams, instead of a dynamic derivation from cLevel (but does that happen ?) */
    zcs->requestedParams = params;

    return ZSTD_resetCStream_internal(zcs, NULL, 0, ZSTD_dm_auto, zcs->cdict, params, pledgedSrcSize);
@@ -2612,10 +2619,9 @@ size_t ZSTD_initCStream_advanced(ZSTD_CStream* zcs,
                                 const void* dict, size_t dictSize,
                                 ZSTD_parameters params, unsigned long long pledgedSrcSize)
 {
-    ZSTD_CCtx_params const cctxParams =
-            ZSTD_assignParamsToCCtxParams(zcs->requestedParams, params);
+    ZSTD_CCtx_params const cctxParams = ZSTD_assignParamsToCCtxParams(zcs->requestedParams, params);
    DEBUGLOG(4, "ZSTD_initCStream_advanced: pledgedSrcSize=%u, flag=%u",
-            (U32)pledgedSrcSize, params.fParams.contentSizeFlag);
+                (U32)pledgedSrcSize, params.fParams.contentSizeFlag);
    CHECK_F( ZSTD_checkCParams(params.cParams) );
    if ((pledgedSrcSize==0) && (params.fParams.contentSizeFlag==0)) pledgedSrcSize = ZSTD_CONTENTSIZE_UNKNOWN;  /* for compatibility with older programs relying on this behavior. Users should now specify ZSTD_CONTENTSIZE_UNKNOWN. This line will be removed in the future. */
    return ZSTD_initCStream_internal(zcs, dict, dictSize, NULL /*cdict*/, cctxParams, pledgedSrcSize);
@@ -2815,7 +2821,7 @@ size_t ZSTD_compress_generic (ZSTD_CCtx* cctx,
                              ZSTD_inBuffer* input,
                              ZSTD_EndDirective endOp)
 {
-    DEBUGLOG(5, "ZSTD_compress_generic");
+    DEBUGLOG(5, "ZSTD_compress_generic, endOp=%u ", (U32)endOp);
    /* check conditions */
    if (output->pos > output->size) return ERROR(GENERIC);
    if (input->pos  > input->size)  return ERROR(GENERIC);
@@ -2862,7 +2868,6 @@ size_t ZSTD_compress_generic (ZSTD_CCtx* cctx,
 #ifdef ZSTD_MULTITHREAD
    if (cctx->appliedParams.nbThreads > 1) {
        size_t const flushMin = ZSTDMT_compressStream_generic(cctx->mtctx, output, input, endOp);
-        DEBUGLOG(5, "ZSTDMT_compressStream_generic result : %u", (U32)flushMin);
        if ( ZSTD_isError(flushMin)
          || (endOp == ZSTD_e_end && flushMin == 0) ) { /* compression completed */
            ZSTD_startNewCompression(cctx);
@@ -447,6 +447,7 @@ ZSTD_compressionParameters ZSTD_getCParamsFromCDict(const ZSTD_CDict* cdict);
 size_t ZSTD_compressBegin_advanced_internal(ZSTD_CCtx* cctx,
                                    const void* dict, size_t dictSize,
                                    ZSTD_dictMode_e dictMode,
+                                    const ZSTD_CDict* cdict,
                                    ZSTD_CCtx_params params,
                                    unsigned long long pledgedSrcSize);

@@ -310,7 +310,7 @@ static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
 typedef struct {
    buffer_t src;
    const void* srcStart;
-    size_t   dictSize;
+    size_t   prefixSize;
    size_t   srcSize;
    buffer_t dstBuff;
    size_t   cSize;
@@ -333,10 +333,10 @@ void ZSTDMT_compressChunk(void* jobDescription)
 {
    ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription;
    ZSTD_CCtx* const cctx = ZSTDMT_getCCtx(job->cctxPool);
-    const void* const src = (const char*)job->srcStart + job->dictSize;
+    const void* const src = (const char*)job->srcStart + job->prefixSize;
    buffer_t dstBuff = job->dstBuff;
-    DEBUGLOG(5, "ZSTDMT_compressChunk: job (first:%u) (last:%u) : dictSize %u, srcSize %u",
-                 job->firstChunk, job->lastChunk, (U32)job->dictSize, (U32)job->srcSize);
+    DEBUGLOG(5, "ZSTDMT_compressChunk: job (first:%u) (last:%u) : prefixSize %u, srcSize %u ",
+                 job->firstChunk, job->lastChunk, (U32)job->prefixSize, (U32)job->srcSize);

    if (cctx==NULL) {
        job->cSize = ERROR(memory_allocation);
@@ -350,28 +350,35 @@ void ZSTDMT_compressChunk(void* jobDescription)
            goto _endJob;
        }
        job->dstBuff = dstBuff;
-        DEBUGLOG(5, "ZSTDMT_compressChunk: allocated dstBuff of size %u", (U32)dstBuff.size);
+        DEBUGLOG(5, "ZSTDMT_compressChunk: received dstBuff of size %u", (U32)dstBuff.size);
    }

    if (job->cdict) {
-        size_t const initError = ZSTD_compressBegin_usingCDict_advanced(cctx, job->cdict, job->params.fParams, job->fullFrameSize);
+        size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, NULL, 0, ZSTD_dm_auto, job->cdict, job->params, job->fullFrameSize);
        DEBUGLOG(4, "ZSTDMT_compressChunk: init using CDict");
-        assert(job->firstChunk); /* should only happen for first segment */
+        assert(job->firstChunk);  /* only allowed for first job */
        if (ZSTD_isError(initError)) { job->cSize = initError; goto _endJob; }
    } else {  /* srcStart points at reloaded section */
+        U64 const pledgedSrcSize = job->firstChunk ? job->fullFrameSize : ZSTD_CONTENTSIZE_UNKNOWN;
        ZSTD_CCtx_params jobParams = job->params;   /* do not modify job->params ! copy it, modify the copy */
        size_t const forceWindowError = ZSTD_CCtxParam_setParameter(&jobParams, ZSTD_p_forceMaxWindow, !job->firstChunk);
-        U64 const pledgedSrcSize = job->firstChunk ? job->fullFrameSize : ZSTD_CONTENTSIZE_UNKNOWN;
-        /* load dictionary in "content-only" mode (no header analysis) */
-        size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, job->srcStart, job->dictSize, ZSTD_dm_rawContent, jobParams, pledgedSrcSize);
-        if (ZSTD_isError(initError) || ZSTD_isError(forceWindowError)) {
-            job->cSize = initError;
+        if (ZSTD_isError(forceWindowError)) {
+            DEBUGLOG(5, "ZSTD_CCtxParam_setParameter error : %s ", ZSTD_getErrorName(forceWindowError));
+            job->cSize = forceWindowError;
            goto _endJob;
        }
+        /* load dictionary in "content-only" mode (no header analysis) */
+        {   size_t const initError = ZSTD_compressBegin_advanced_internal(cctx, job->srcStart, job->prefixSize, ZSTD_dm_rawContent, NULL, jobParams, pledgedSrcSize);
+            DEBUGLOG(5, "ZSTD_compressBegin_advanced_internal called with windowLog = %u ", jobParams.cParams.windowLog);
+            if (ZSTD_isError(initError)) {
+                DEBUGLOG(5, "ZSTD_compressBegin_advanced_internal error : %s ", ZSTD_getErrorName(initError));
+                job->cSize = initError;
+                goto _endJob;
+        }   }
    }
-    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 job */
        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; /* save error code */ goto _endJob; }
        ZSTD_invalidateRepCodes(cctx);
    }

@@ -380,9 +387,9 @@ void ZSTDMT_compressChunk(void* jobDescription)
    job->cSize = (job->lastChunk) ?
                 ZSTD_compressEnd     (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);
-    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:
    ZSTDMT_releaseCCtx(job->cctxPool, cctx);
@@ -558,11 +565,13 @@ size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx)
 }

 /* Internal only */
-size_t ZSTDMT_CCtxParam_setMTCtxParameter(
-    ZSTD_CCtx_params* params, ZSTDMT_parameter parameter, unsigned value) {
+size_t ZSTDMT_CCtxParam_setMTCtxParameter(ZSTD_CCtx_params* params,
+                                ZSTDMT_parameter parameter, unsigned value) {
+    DEBUGLOG(4, "ZSTDMT_CCtxParam_setMTCtxParameter");
    switch(parameter)
    {
-    case ZSTDMT_p_sectionSize :
+    case ZSTDMT_p_jobSize :
+        DEBUGLOG(4, "ZSTDMT_CCtxParam_setMTCtxParameter : set jobSize to %u", value);
        if ( (value > 0)  /* value==0 => automatic job size */
           & (value < ZSTDMT_JOBSIZE_MIN) )
            value = ZSTDMT_JOBSIZE_MIN;
@@ -580,9 +589,10 @@ size_t ZSTDMT_CCtxParam_setMTCtxParameter(

 size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSTDMT_parameter parameter, unsigned value)
 {
+    DEBUGLOG(4, "ZSTDMT_setMTCtxParameter");
    switch(parameter)
    {
-    case ZSTDMT_p_sectionSize :
+    case ZSTDMT_p_jobSize :
        return ZSTDMT_CCtxParam_setMTCtxParameter(&mtctx->params, parameter, value);
    case ZSTDMT_p_overlapSectionLog :
        return ZSTDMT_CCtxParam_setMTCtxParameter(&mtctx->params, parameter, value);
@@ -618,7 +628,7 @@ static size_t ZSTDMT_compress_advanced_internal(
    size_t const overlapSize = (overlapRLog>=9) ? 0 : (size_t)1 << (params.cParams.windowLog - overlapRLog);
    unsigned nbChunks = computeNbChunks(srcSize, params.cParams.windowLog, params.nbThreads);
    size_t const proposedChunkSize = (srcSize + (nbChunks-1)) / nbChunks;
-    size_t const avgChunkSize = ((proposedChunkSize & 0x1FFFF) < 0x7FFF) ? proposedChunkSize + 0xFFFF : proposedChunkSize;   /* avoid too small last block */
+    size_t const avgChunkSize = (((proposedChunkSize-1) & 0x1FFFF) < 0x7FFF) ? proposedChunkSize + 0xFFFF : proposedChunkSize;   /* avoid too small last block */
    const char* const srcStart = (const char*)src;
    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 */
@@ -628,7 +638,8 @@ static size_t ZSTDMT_compress_advanced_internal(
    assert(mtctx->cctxPool->totalCCtx == params.nbThreads);

    DEBUGLOG(4, "ZSTDMT_compress_advanced_internal");
-    DEBUGLOG(4, "nbChunks  : %2u   (chunkSize : %u bytes)   ", nbChunks, (U32)avgChunkSize);
+    DEBUGLOG(4, "nbChunks  : %2u (raw chunkSize : %u bytes; fixed chunkSize: %u)   ",
+                nbChunks, (U32)proposedChunkSize, (U32)avgChunkSize);
    if (nbChunks==1) {   /* fallback to single-thread mode */
        ZSTD_CCtx* const cctx = mtctx->cctxPool->cctx[0];
        if (cdict) return ZSTD_compress_usingCDict_advanced(cctx, dst, dstCapacity, src, srcSize, cdict, jobParams.fParams);
@@ -657,7 +668,7 @@ static size_t ZSTDMT_compress_advanced_internal(

            mtctx->jobs[u].src = g_nullBuffer;
            mtctx->jobs[u].srcStart = srcStart + frameStartPos - dictSize;
-            mtctx->jobs[u].dictSize = dictSize;
+            mtctx->jobs[u].prefixSize = dictSize;
            mtctx->jobs[u].srcSize = chunkSize;
            mtctx->jobs[u].cdict = (u==0) ? cdict : NULL;
            mtctx->jobs[u].fullFrameSize = srcSize;
@@ -815,7 +826,7 @@ size_t ZSTDMT_initCStream_internal(
    zcs->targetSectionSize = params.jobSize ? params.jobSize : (size_t)1 << (params.cParams.windowLog + 2);
    if (zcs->targetSectionSize < ZSTDMT_JOBSIZE_MIN) zcs->targetSectionSize = ZSTDMT_JOBSIZE_MIN;
    if (zcs->targetSectionSize < zcs->targetDictSize) zcs->targetSectionSize = zcs->targetDictSize;  /* job size must be >= overlap size */
-    DEBUGLOG(4, "Job Size : %u KB", (U32)(zcs->targetSectionSize>>10));
+    DEBUGLOG(4, "Job Size : %u KB (note : set to %u)", (U32)(zcs->targetSectionSize>>10), params.jobSize);
    zcs->inBuffSize = zcs->targetDictSize + zcs->targetSectionSize;
    DEBUGLOG(4, "inBuff Size : %u KB", (U32)(zcs->inBuffSize>>10));
    ZSTDMT_setBufferSize(zcs->bufPool, MAX(zcs->inBuffSize, ZSTD_compressBound(zcs->targetSectionSize)) );
@@ -888,7 +899,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
    zcs->jobs[jobID].src = zcs->inBuff.buffer;
    zcs->jobs[jobID].srcStart = zcs->inBuff.buffer.start;
    zcs->jobs[jobID].srcSize = srcSize;
-    zcs->jobs[jobID].dictSize = zcs->dictSize;
+    zcs->jobs[jobID].prefixSize = zcs->dictSize;
    assert(zcs->inBuff.filled >= srcSize + zcs->dictSize);
    zcs->jobs[jobID].params = zcs->params;
    /* do not calculate checksum within sections, but write it in header for first section */
@@ -953,6 +964,7 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* zcs, size_t srcSize, unsi
 static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsigned blockToFlush)
 {
    unsigned const wJobID = zcs->doneJobID & zcs->jobIDMask;
+    DEBUGLOG(5, "ZSTDMT_flushNextJob");
    if (zcs->doneJobID == zcs->nextJobID) return 0;   /* all flushed ! */
    ZSTD_PTHREAD_MUTEX_LOCK(&zcs->jobCompleted_mutex);
    while (zcs->jobs[wJobID].jobCompleted==0) {
@@ -965,7 +977,8 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
    {   ZSTDMT_jobDescription job = zcs->jobs[wJobID];
        if (!job.jobScanned) {
            if (ZSTD_isError(job.cSize)) {
-                DEBUGLOG(5, "compression error detected ");
+                DEBUGLOG(5, "job %u : compression error detected : %s",
+                            zcs->doneJobID, ZSTD_getErrorName(job.cSize));
                ZSTDMT_waitForAllJobsCompleted(zcs);
                ZSTDMT_releaseAllJobResources(zcs);
                return job.cSize;
@@ -1014,7 +1027,7 @@ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx,
 {
    size_t const newJobThreshold = mtctx->dictSize + mtctx->targetSectionSize;
    unsigned forwardInputProgress = 0;
-    DEBUGLOG(5, "ZSTDMT_compressStream_generic");
+    DEBUGLOG(5, "ZSTDMT_compressStream_generic ");
    assert(output->pos <= output->size);
    assert(input->pos  <= input->size);

@@ -1097,9 +1110,11 @@ size_t ZSTDMT_compressStream(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, ZSTD_inBu
 static size_t ZSTDMT_flushStream_internal(ZSTDMT_CCtx* mtctx, ZSTD_outBuffer* output, unsigned endFrame)
 {
    size_t const srcSize = mtctx->inBuff.filled - mtctx->dictSize;
+    DEBUGLOG(5, "ZSTDMT_flushStream_internal");

    if ( ((srcSize > 0) || (endFrame && !mtctx->frameEnded))
       && (mtctx->nextJobID <= mtctx->doneJobID + mtctx->jobIDMask) ) {
+           DEBUGLOG(5, "ZSTDMT_flushStream_internal : create a new job");
        CHECK_F( ZSTDMT_createCompressionJob(mtctx, srcSize, endFrame) );
    }

@@ -84,13 +84,13 @@ ZSTDLIB_API size_t ZSTDMT_initCStream_usingCDict(ZSTDMT_CCtx* mtctx,
 /* ZSTDMT_parameter :
 * List of parameters that can be set using ZSTDMT_setMTCtxParameter() */
 typedef enum {
-    ZSTDMT_p_sectionSize,        /* size of input "section". Each section is compressed in parallel. 0 means default, which is dynamically determined within compression functions */
-    ZSTDMT_p_overlapSectionLog   /* Log of overlapped section; 0 == no overlap, 6(default) == use 1/8th of window, >=9 == use full window */
+    ZSTDMT_p_jobSize,           /* Each job is compressed in parallel. By default, this value is dynamically determined depending on compression parameters. Can be set explicitly here. */
+    ZSTDMT_p_overlapSectionLog  /* Each job may reload a part of previous job to enhance compressionr ratio; 0 == no overlap, 6(default) == use 1/8th of window, >=9 == use full window */
 } ZSTDMT_parameter;

 /* ZSTDMT_setMTCtxParameter() :
 * allow setting individual parameters, one at a time, among a list of enums defined in ZSTDMT_parameter.
- * The function must be called typically after ZSTD_createCCtx().
+ * The function must be called typically after ZSTD_createCCtx() but __before ZSTDMT_init*() !__
 * Parameters not explicitly reset by ZSTDMT_init*() remain the same in consecutive compression sessions.
 * @return : 0, or an error code (which can be tested using ZSTD_isError()) */
 ZSTDLIB_API size_t ZSTDMT_setMTCtxParameter(ZSTDMT_CCtx* mtctx, ZSTDMT_parameter parameter, unsigned value);
@@ -114,7 +114,7 @@ size_t ZSTDMT_CCtxParam_setMTCtxParameter(ZSTD_CCtx_params* params, ZSTDMT_param

 /* ZSTDMT_CCtxParam_setNbThreads()
 * Set nbThreads, and clamp it correctly,
- * but also reset jobSize and overlapLog */ 
+ * but also reset jobSize and overlapLog */
 size_t ZSTDMT_CCtxParam_setNbThreads(ZSTD_CCtx_params* params, unsigned nbThreads);

 /*! ZSTDMT_initCStream_internal() :
@@ -2451,14 +2451,16 @@ size_t ZSTD_decompressStream(ZSTD_DStream* zds, ZSTD_outBuffer* output, ZSTD_inB
                        return ZSTD_decompressLegacyStream(zds->legacyContext, legacyVersion, output, input);
                    }
 #endif
-                    return hSize; /* error */
+                    return hSize;   /* error */
                }
                if (hSize != 0) {   /* need more input */
                    size_t const toLoad = hSize - zds->lhSize;   /* if hSize!=0, hSize > zds->lhSize */
-                    if (toLoad > (size_t)(iend-ip)) {   /* not enough input to load full header */
-                        if (iend-ip > 0) {
-                            memcpy(zds->headerBuffer + zds->lhSize, ip, iend-ip);
-                            zds->lhSize += iend-ip;
+                    size_t const remainingInput = (size_t)(iend-ip);
+                    assert(iend >= ip);
+                    if (toLoad > remainingInput) {   /* not enough input to load full header */
+                        if (remainingInput > 0) {
+                            memcpy(zds->headerBuffer + zds->lhSize, ip, remainingInput);
+                            zds->lhSize += remainingInput;
                        }
                        input->pos = input->size;
                        return (MAX(ZSTD_frameHeaderSize_min, hSize) - zds->lhSize) + ZSTD_blockHeaderSize;   /* remaining header bytes + next block header */
@@ -2473,8 +2475,10 @@ size_t ZSTD_decompressStream(ZSTD_DStream* zds, ZSTD_outBuffer* output, ZSTD_inB
                && (U64)(size_t)(oend-op) >= zds->fParams.frameContentSize) {
                size_t const cSize = ZSTD_findFrameCompressedSize(istart, iend-istart);
                if (cSize <= (size_t)(iend-istart)) {
+                    /* shortcut : using single-pass mode */
                    size_t const decompressedSize = ZSTD_decompress_usingDDict(zds, op, oend-op, istart, cSize, zds->ddict);
                    if (ZSTD_isError(decompressedSize)) return decompressedSize;
+                    DEBUGLOG(4, "shortcut to single-pass ZSTD_decompress_usingDDict()")
                    ip = istart + cSize;
                    op += decompressedSize;
                    zds->expected = 0;
@@ -2497,8 +2501,9 @@ size_t ZSTD_decompressStream(ZSTD_DStream* zds, ZSTD_outBuffer* output, ZSTD_inB
            }

            /* control buffer memory usage */
-            DEBUGLOG(4, "Control max buffer memory usage (max %u KB)",
-                        (U32)(zds->maxWindowSize >> 10));
+            DEBUGLOG(4, "Control max memory usage (%u KB <= max %u KB)",
+                        (U32)(zds->fParams.windowSize >>10),
+                        (U32)(zds->maxWindowSize >> 10) );
            zds->fParams.windowSize = MAX(zds->fParams.windowSize, 1U << ZSTD_WINDOWLOG_ABSOLUTEMIN);
            if (zds->fParams.windowSize > zds->maxWindowSize) return ERROR(frameParameter_windowTooLarge);

@@ -1016,7 +1016,8 @@ typedef enum {
                              * More threads improve speed, but also increase memory usage.
                              * Can only receive a value > 1 if ZSTD_MULTITHREAD is enabled.
                              * Special: value 0 means "do not change nbThreads" */
-    ZSTD_p_jobSize,          /* Size of a compression job. Each compression job is completed in parallel.
+    ZSTD_p_jobSize,          /* Size of a compression job. This value is only enforced in streaming (non-blocking) mode.
+                              * Each compression job is completed in parallel, so indirectly controls the nb of active threads.
                              * 0 means default, which is dynamically determined based on compression parameters.
                              * Job size must be a minimum of overlapSize, or 1 KB, whichever is largest
                              * The minimum size is automatically and transparently enforced */
@@ -1144,13 +1145,19 @@ typedef enum {
 *  - Compression parameters cannot be changed once compression is started.
 *  - outpot->pos must be <= dstCapacity, input->pos must be <= srcSize
 *  - outpot->pos and input->pos will be updated. They are guaranteed to remain below their respective limit.
- *  - @return provides the minimum amount of data still to flush from internal buffers
+ *  - In single-thread mode (default), function is blocking : it completed its job before returning to caller.
+ *  - In multi-thread mode, function is non-blocking : it just acquires a copy of input, and distribute job to internal worker threads,
+ *                                                     and then immediately returns, just indicating that there is some data remaining to be flushed.
+ *                                                     The function nonetheless guarantees forward progress : it will return only after it reads or write at least 1+ byte.
+ *  - Exception : in multi-threading mode, if the first call requests a ZSTD_e_end directive, it is blocking : it will complete compression before giving back control to caller.
+ *  - @return provides the minimum amount of data remaining to be flushed from internal buffers
 *            or an error code, which can be tested using ZSTD_isError().
- *            if @return != 0, flush is not fully completed, there is some data left within internal buffers.
- *  - after a ZSTD_e_end directive, if internal buffer is not fully flushed,
+ *            if @return != 0, flush is not fully completed, there is still some data left within internal buffers.
+ *            This is useful to determine if a ZSTD_e_flush or ZSTD_e_end directive is completed.
+ *  - after a ZSTD_e_end directive, if internal buffer is not fully flushed (@return != 0),
 *            only ZSTD_e_end or ZSTD_e_flush operations are allowed.
- *            It is necessary to fully flush internal buffers
- *            before starting a new compression job, or changing compression parameters.
+ *            Before starting a new compression job, or changing compression parameters,
+ *            it is required to fully flush internal buffers.
 */
 ZSTDLIB_API size_t ZSTD_compress_generic (ZSTD_CCtx* cctx,
                                          ZSTD_outBuffer* output,