fix : ZSTDMT_compress_advanced() correctly generates checksum

when params.fParams.checksumFlag==1.
This use case used to be impossible when only ZSTD_compress() was available

lib/compress/zstdmt_compress.c

@@ -140,7 +140,7 @@ static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* bufPool, size_t bSize)
 static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
 {
     size_t const bSize = bufPool->bufferSize;
-    DEBUGLOG(2, "ZSTDMT_getBuffer");
+    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)];
@@ -151,12 +151,12 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
             return buf;
         }
         /* size conditions not respected : scratch this buffer, create new one */
-        DEBUGLOG(2, "existing buffer does not meet size conditions => freeing");
+        DEBUGLOG(5, "existing buffer does not meet size conditions => freeing");
         ZSTD_free(buf.start, bufPool->cMem);
     }
     pthread_mutex_unlock(&bufPool->poolMutex);
     /* create new buffer */
-    DEBUGLOG(2, "create a new buffer");
+    DEBUGLOG(5, "create a new buffer");
     {   buffer_t buffer;
         void* const start = ZSTD_malloc(bSize, bufPool->cMem);
         buffer.start = start;   /* note : start can be NULL if malloc fails ! */
@@ -168,17 +168,17 @@ static buffer_t ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
 /* store buffer for later re-use, up to pool capacity */
 static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, buffer_t buf)
 {
-    DEBUGLOG(2, "ZSTDMT_releaseBuffer");
     if (buf.start == NULL) return;   /* compatible with release on NULL */
+    DEBUGLOG(5, "ZSTDMT_releaseBuffer");
     pthread_mutex_lock(&bufPool->poolMutex);
     if (bufPool->nbBuffers < bufPool->totalBuffers) {
-        bufPool->bTable[bufPool->nbBuffers++] = buf;   /* store for later re-use */
+        bufPool->bTable[bufPool->nbBuffers++] = buf;  /* stored for later use */
         pthread_mutex_unlock(&bufPool->poolMutex);
         return;
     }
     pthread_mutex_unlock(&bufPool->poolMutex);
     /* Reached bufferPool capacity (should not happen) */
-    DEBUGLOG(2, "buffer pool capacity reached => freeing ");
+    DEBUGLOG(5, "buffer pool capacity reached => freeing ");
     ZSTD_free(buf.start, bufPool->cMem);
 }
 
@@ -338,7 +338,7 @@ 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(2, "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));
 
@@ -515,7 +515,6 @@ static unsigned computeNbChunks(size_t srcSize, unsigned windowLog, unsigned nbT
 }
 
 
-/* Note : missing checksum at the end ! */
 size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
                                void* dst, size_t dstCapacity,
                          const void* src, size_t srcSize,
@@ -531,6 +530,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
     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 */
     size_t frameStartPos = 0, dstBufferPos = 0;
+    XXH64_state_t xxh64;
 
     DEBUGLOG(4, "nbChunks  : %2u   (chunkSize : %u bytes)   ", nbChunks, (U32)avgChunkSize);
     if (nbChunks==1) {   /* fallback to single-thread mode */
@@ -540,6 +540,7 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
     }
     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 */
         U32 nbJobs = nbChunks;
@@ -576,6 +577,10 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
             mtctx->jobs[u].jobCompleted_mutex = &mtctx->jobCompleted_mutex;
             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);
             DEBUG_PRINTHEX(6, mtctx->jobs[u].srcStart, 12);
             POOL_add(mtctx->factory, ZSTDMT_compressChunk, &mtctx->jobs[u]);
@@ -586,8 +591,8 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
     }   }
 
     /* collect result */
-    {   unsigned chunkID;
+    {   size_t error = 0, dstPos = 0;
-        size_t error = 0, dstPos = 0;
+        unsigned chunkID;
         for (chunkID=0; chunkID<nbChunks; chunkID++) {
             DEBUGLOG(5, "waiting for chunk %u ", chunkID);
             PTHREAD_MUTEX_LOCK(&mtctx->jobCompleted_mutex);
@@ -614,6 +619,18 @@ size_t ZSTDMT_compress_advanced(ZSTDMT_CCtx* mtctx,
                 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);
         return error ? error : dstPos;
     }
@@ -852,7 +869,7 @@ static size_t ZSTDMT_flushNextJob(ZSTDMT_CCtx* zcs, ZSTD_outBuffer* output, unsi
             zcs->jobs[wJobID].jobScanned = 1;
         }
         {   size_t const toWrite = MIN(job.cSize - job.dstFlushed, output->size - output->pos);
-            DEBUGLOG(2, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID);
+            DEBUGLOG(5, "Flushing %u bytes from job %u ", (U32)toWrite, zcs->doneJobID);
             memcpy((char*)output->dst + output->pos, (const char*)job.dstBuff.start + job.dstFlushed, toWrite);
             output->pos += toWrite;
             job.dstFlushed += toWrite;

lib/compress/zstdmt_compress.h

@@ -16,8 +16,8 @@
 
 
 /* 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.
  *        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 ZSTD_CDict* cdict,
                                            ZSTD_parameters const params,
-                                           unsigned overlapRLog);
+                                           unsigned overlapRLog);            /* overlapRLog = 9 - overlapLog */
 
 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 */

tests/fuzzer.c

@@ -278,9 +278,6 @@ static int basicUnitTests(U32 seed, double compressibility)
     }
     RDG_genBuffer(CNBuffer, CNBuffSize, compressibility, 0., seed);
 
-    /* memory tests */
-    FUZ_mallocTests(seed, compressibility);
-
     /* Basic tests */
     DISPLAYLEVEL(4, "test%3i : ZSTD_getErrorName : ", testNb++);
     {   const char* errorString = ZSTD_getErrorName(0);
@@ -479,6 +476,23 @@ static int basicUnitTests(U32 seed, double compressibility)
         }   }
         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);
     }
 

tests/zstreamtest.c

@@ -1377,13 +1377,12 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double
         /* multi-segments compression test */
         XXH64_reset(&xxhState, 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 */
                 size_t const randomSrcSize = FUZ_randomLength(&lseed, maxSampleLog);
                 size_t const srcSize = MIN(maxTestSize-totalTestSize, randomSrcSize);
                 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);
                 ZSTD_EndDirective const flush = (FUZ_rand(&lseed) & 15) ? ZSTD_e_continue : ZSTD_e_flush;
                 ZSTD_inBuffer inBuff = { srcBuffer+srcStart, srcSize, 0 };
@@ -1402,7 +1401,7 @@ static int fuzzerTests_newAPI(U32 seed, U32 nbTests, unsigned startTest, double
             {   size_t remainingToFlush = (size_t)(-1);
                 while (remainingToFlush) {
                     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);
                     outBuff.size = outBuff.pos + adjustedDstSize;
                     DISPLAYLEVEL(5, "End-flush into dst buffer of size %u \n", (U32)adjustedDstSize);