/* * Copyright (c) Meta Platforms, Inc. and affiliates. * All rights reserved. * * This source code is licensed under both the BSD-style license (found in the * LICENSE file in the root directory of this source tree) and the GPLv2 (found * in the COPYING file in the root directory of this source tree). * You may select, at your option, one of the above-listed licenses. */ /* ====== Compiler specifics ====== */ #if defined(_MSC_VER) # pragma warning(disable : 4204) /* disable: C4204: non-constant aggregate initializer */ #endif /* ====== Dependencies ====== */ #include "../common/allocations.h" /* ZSTD_customMalloc, ZSTD_customCalloc, ZSTD_customFree */ #include "../common/zstd_deps.h" /* ZSTD_memmove, ZSTD_memset, INT_MAX, UINT_MAX */ #include "../common/mem.h" /* MEM_STATIC */ #include "../common/pool.h" /* threadpool */ #include "../common/threading.h" /* mutex */ #include "zstd_compress_internal.h" /* MIN, ERROR, ZSTD_*, ZSTD_highbit32 */ #include "zstd_ldm.h" #include "zstdmt_compress.h" /* The Rust MT parameter-update policy receives only this scalar snapshot. * Keep the private ZSTDMT_CCtx and ZSTD_CCtx_params layouts in this * translation unit. */ typedef struct { int compressionLevel; int cctxSrcSizeHint; U64 srcSizeHint; size_t dictSize; int mode; int enableLdm; U32 ldmDefaultWindowLog; ZSTD_compressionParameters overrides; int useRowMatchFinder; U32 exclusionMask; U32 savedWindowLog; } ZSTDMT_RustCParamsUpdateProjection; typedef char ZSTDMT_rust_cparams_update_projection_layout[ (offsetof(ZSTDMT_RustCParamsUpdateProjection, compressionLevel) == 0 && offsetof(ZSTDMT_RustCParamsUpdateProjection, cctxSrcSizeHint) == sizeof(int) && offsetof(ZSTDMT_RustCParamsUpdateProjection, srcSizeHint) == 2 * sizeof(int) && offsetof(ZSTDMT_RustCParamsUpdateProjection, dictSize) == 16 && offsetof(ZSTDMT_RustCParamsUpdateProjection, mode) == (sizeof(void*) == 8 ? 24 : 20) && offsetof(ZSTDMT_RustCParamsUpdateProjection, enableLdm) == (sizeof(void*) == 8 ? 28 : 24) && offsetof(ZSTDMT_RustCParamsUpdateProjection, ldmDefaultWindowLog) == (sizeof(void*) == 8 ? 32 : 28) && offsetof(ZSTDMT_RustCParamsUpdateProjection, overrides) == (sizeof(void*) == 8 ? 36 : 32) && offsetof(ZSTDMT_RustCParamsUpdateProjection, useRowMatchFinder) == (sizeof(void*) == 8 ? 64 : 60) && offsetof(ZSTDMT_RustCParamsUpdateProjection, exclusionMask) == (sizeof(void*) == 8 ? 68 : 64) && offsetof(ZSTDMT_RustCParamsUpdateProjection, savedWindowLog) == (sizeof(void*) == 8 ? 72 : 68) && sizeof(ZSTDMT_RustCParamsUpdateProjection) == (sizeof(void*) == 8 ? 80 : 72)) ? 1 : -1]; typedef struct { int compressionLevel; ZSTD_compressionParameters cParams; } ZSTDMT_RustCParamsUpdateResult; typedef char ZSTDMT_rust_cparams_update_result_layout[ (offsetof(ZSTDMT_RustCParamsUpdateResult, compressionLevel) == 0 && offsetof(ZSTDMT_RustCParamsUpdateResult, cParams) == sizeof(int) && sizeof(ZSTDMT_RustCParamsUpdateResult) == sizeof(int) + sizeof(ZSTD_compressionParameters)) ? 1 : -1]; ZSTDMT_RustCParamsUpdateResult ZSTDMT_rust_updateCParamsWhileCompressing( ZSTDMT_RustCParamsUpdateProjection projection); /* Defined in zstd_compress.c so all C translation units use the same * preprocessor-derived block-compressor exclusion mask. */ U32 ZSTD_getCParamsExclusionMask(void); /* Guards code to support resizing the SeqPool. * We will want to resize the SeqPool to save memory in the future. * Until then, comment the code out since it is unused. */ #define ZSTD_RESIZE_SEQPOOL 0 /* ====== Debug ====== */ #if defined(DEBUGLEVEL) && (DEBUGLEVEL>=2) \ && !defined(_MSC_VER) \ && !defined(__MINGW32__) # include # include # include # define DEBUG_PRINTHEX(l,p,n) \ do { \ unsigned debug_u; \ for (debug_u=0; debug_u<(n); debug_u++) \ RAWLOG(l, "%02X ", ((const unsigned char*)(p))[debug_u]); \ RAWLOG(l, " \n"); \ } while (0) static unsigned long long GetCurrentClockTimeMicroseconds(void) { static clock_t _ticksPerSecond = 0; if (_ticksPerSecond <= 0) _ticksPerSecond = sysconf(_SC_CLK_TCK); { struct tms junk; clock_t newTicks = (clock_t) times(&junk); return ((((unsigned long long)newTicks)*(1000000))/_ticksPerSecond); } } #define MUTEX_WAIT_TIME_DLEVEL 6 #define ZSTD_PTHREAD_MUTEX_LOCK(mutex) \ do { \ if (DEBUGLEVEL >= MUTEX_WAIT_TIME_DLEVEL) { \ unsigned long long const beforeTime = GetCurrentClockTimeMicroseconds(); \ ZSTD_pthread_mutex_lock(mutex); \ { unsigned long long const afterTime = GetCurrentClockTimeMicroseconds(); \ unsigned long long const elapsedTime = (afterTime-beforeTime); \ if (elapsedTime > 1000) { \ /* or whatever threshold you like; I'm using 1 millisecond here */ \ DEBUGLOG(MUTEX_WAIT_TIME_DLEVEL, \ "Thread took %llu microseconds to acquire mutex %s \n", \ elapsedTime, #mutex); \ } } \ } else { \ ZSTD_pthread_mutex_lock(mutex); \ } \ } while (0) #else # define ZSTD_PTHREAD_MUTEX_LOCK(m) ZSTD_pthread_mutex_lock(m) # define DEBUG_PRINTHEX(l,p,n) do { } while (0) #endif /* ===== Buffer Pool ===== */ /* a single Buffer Pool can be invoked from multiple threads in parallel */ typedef struct buffer_s { void* start; size_t capacity; } Buffer; static const Buffer g_nullBuffer = { NULL, 0 }; /* The Rust module owns the synchronized pool state. Keep a C wrapper so * custom allocation of the containing ZSTDMT context remains unchanged. */ typedef struct ZSTDMT_RustBufferPool_s ZSTDMT_RustBufferPool; typedef struct { void* start; size_t capacity; } ZSTDMT_RustBuffer; ZSTDMT_RustBufferPool* ZSTDMT_rust_buffer_pool_create(unsigned maxNbBuffers, ZSTD_customMem cMem); void ZSTDMT_rust_buffer_pool_free(ZSTDMT_RustBufferPool* pool); size_t ZSTDMT_rust_buffer_pool_sizeof(const ZSTDMT_RustBufferPool* pool); size_t ZSTDMT_rust_sizeofBufferPool(size_t wrapperSize, size_t rustPoolSize); void ZSTDMT_rust_buffer_pool_set_size(ZSTDMT_RustBufferPool* pool, size_t bSize); ZSTDMT_RustBufferPool* ZSTDMT_rust_buffer_pool_expand(ZSTDMT_RustBufferPool* pool, unsigned maxNbBuffers); ZSTDMT_RustBuffer ZSTDMT_rust_buffer_pool_get(ZSTDMT_RustBufferPool* pool); void ZSTDMT_rust_buffer_pool_release(ZSTDMT_RustBufferPool* pool, ZSTDMT_RustBuffer buffer); ZSTDMT_RustBuffer ZSTDMT_rust_buffer_pool_resize(ZSTDMT_RustBufferPool* pool, ZSTDMT_RustBuffer buffer); typedef struct { size_t error; size_t lastBlockSize; } ZSTDMT_chunkProcessResult; typedef struct { void* dst; size_t dstCapacity; int* stage; int noDictIDFlag; int checksumFlag; int contentSizeFlag; int format; U32 windowLog; U64 pledgedSrcSizePlusOne; U32 dictID; U32* repCodes; } ZSTDMT_RustCompressionJobFrameHeaderState; typedef char ZSTDMT_compression_job_frame_header_state_layout[ (offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, dst) == 0 && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, dstCapacity) == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, stage) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, noDictIDFlag) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, checksumFlag) == 3 * sizeof(void*) + sizeof(int) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, contentSizeFlag) == 3 * sizeof(void*) + 2 * sizeof(int) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, format) == 3 * sizeof(void*) + 3 * sizeof(int) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, windowLog) == 3 * sizeof(void*) + 4 * sizeof(int) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, pledgedSrcSizePlusOne) == (sizeof(void*) == 8 ? 6 * sizeof(void*) : 8 * sizeof(void*)) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, dictID) == (sizeof(void*) == 8 ? 6 * sizeof(void*) : 8 * sizeof(void*)) + sizeof(U64) && offsetof(ZSTDMT_RustCompressionJobFrameHeaderState, repCodes) == (sizeof(void*) == 8 ? 8 * sizeof(void*) : 11 * sizeof(void*)) && sizeof(ZSTDMT_RustCompressionJobFrameHeaderState) == (sizeof(void*) == 8 ? 72 : 48)) ? 1 : -1]; typedef struct ZSTDMT_RustCompressionJobSequenceState_s ZSTDMT_RustCompressionJobSequenceState; typedef struct { unsigned firstJob; unsigned lastJob; const ZSTDMT_RustCompressionJobFrameHeaderState* frameHeaderState; const ZSTDMT_RustCompressionJobSequenceState* sequenceState; } ZSTDMT_RustCompressionJobProjection; typedef char ZSTDMT_compression_job_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobProjection, firstJob) == 0 && offsetof(ZSTDMT_RustCompressionJobProjection, lastJob) == sizeof(unsigned) && offsetof(ZSTDMT_RustCompressionJobProjection, frameHeaderState) == 2 * sizeof(unsigned) && offsetof(ZSTDMT_RustCompressionJobProjection, sequenceState) == 2 * sizeof(unsigned) + sizeof(void*) && sizeof(ZSTDMT_RustCompressionJobProjection) == 2 * sizeof(unsigned) + 2 * sizeof(void*)) ? 1 : -1]; typedef struct { unsigned firstJob; unsigned hasCDict; } ZSTDMT_RustCompressionJobBeginProjection; typedef char ZSTDMT_compression_job_begin_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobBeginProjection, firstJob) == 0 && offsetof(ZSTDMT_RustCompressionJobBeginProjection, hasCDict) == sizeof(unsigned) && sizeof(ZSTDMT_RustCompressionJobBeginProjection) == 2 * sizeof(unsigned)) ? 1 : -1]; typedef struct { int status; size_t toFlush; size_t outputPos; size_t dstFlushed; } ZSTDMT_flushPublicationResult; typedef void (*ZSTDMT_chunkProgressFn)(void* opaque, size_t cSize, size_t consumed); typedef size_t (*ZSTDMT_compressionJobStepFn)(void* opaque); typedef void (*ZSTDMT_compressionJobVoidFn)(void* opaque); typedef ZSTDMT_chunkProcessResult (*ZSTDMT_compressionJobCompressFn)( void* opaque, unsigned lastJob); typedef void (*ZSTDMT_compressionJobSizeFn)(void* opaque, size_t size); typedef size_t (*ZSTDMT_compressionJobSetParameterFn)(void* opaque, int value); typedef void (*ZSTDMT_compressionJobApplySequencesFn)( void* opaque, void* cctx, void* sequences, size_t nbSequences); typedef struct { void* callbackContext; size_t* cSize; size_t* consumed; ZSTDMT_compressionJobVoidFn lock; ZSTDMT_compressionJobVoidFn signal; ZSTDMT_compressionJobVoidFn unlock; ZSTDMT_chunkProgressFn debug; } ZSTDMT_RustCompressionJobProgressProjection; typedef char ZSTDMT_compression_job_progress_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobProgressProjection, callbackContext) == 0 && offsetof(ZSTDMT_RustCompressionJobProgressProjection, cSize) == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobProgressProjection, consumed) == sizeof(void*) + sizeof(size_t) && offsetof(ZSTDMT_RustCompressionJobProgressProjection, lock) == sizeof(void*) + 2 * sizeof(size_t) && offsetof(ZSTDMT_RustCompressionJobProgressProjection, signal) == sizeof(void*) + 2 * sizeof(size_t) + sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobProgressProjection, unlock) == sizeof(void*) + 2 * sizeof(size_t) + 2 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobProgressProjection, debug) == sizeof(void*) + 2 * sizeof(size_t) + 3 * sizeof(void*) && sizeof(ZSTDMT_RustCompressionJobProgressProjection) == sizeof(void*) + 2 * sizeof(size_t) + 4 * sizeof(void*)) ? 1 : -1]; void ZSTDMT_rust_compressionJobProgress( void* opaque, size_t cSize, size_t consumed); typedef struct { void* callbackContext; size_t* cSize; size_t* consumed; size_t srcSize; ZSTDMT_compressionJobVoidFn lock; ZSTDMT_compressionJobVoidFn signal; ZSTDMT_compressionJobVoidFn unlock; ZSTDMT_compressionJobSizeFn validateConsumed; } ZSTDMT_RustCompressionJobFinishPublicationProjection; typedef char ZSTDMT_compression_job_finish_publication_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, callbackContext) == 0 && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, cSize) == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, consumed) == sizeof(void*) + sizeof(size_t) && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, srcSize) == sizeof(void*) + 2 * sizeof(size_t) && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, lock) == sizeof(void*) + 3 * sizeof(size_t) && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, signal) == sizeof(void*) + 3 * sizeof(size_t) + sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, unlock) == sizeof(void*) + 3 * sizeof(size_t) + 2 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishPublicationProjection, validateConsumed) == sizeof(void*) + 3 * sizeof(size_t) + 3 * sizeof(void*) && sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection) == sizeof(void*) + 3 * sizeof(size_t) + 4 * sizeof(void*)) ? 1 : -1]; typedef struct { void* callbackContext; size_t srcSize; ZSTDMT_compressionJobVoidFn ensureFinished; ZSTDMT_RustCompressionJobFinishPublicationProjection publication; ZSTDMT_compressionJobVoidFn releaseSeq; ZSTDMT_compressionJobVoidFn releaseCCtx; } ZSTDMT_RustCompressionJobFinishProjection; typedef char ZSTDMT_compression_job_finish_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobFinishProjection, callbackContext) == 0 && offsetof(ZSTDMT_RustCompressionJobFinishProjection, srcSize) == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishProjection, ensureFinished) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishProjection, publication) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseSeq) == 3 * sizeof(void*) + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection) && offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseCCtx) == 4 * sizeof(void*) + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection) && sizeof(ZSTDMT_RustCompressionJobFinishProjection) == 5 * sizeof(void*) + sizeof(ZSTDMT_RustCompressionJobFinishPublicationProjection)) ? 1 : -1]; typedef struct { void* callbackContext; unsigned firstJob; ZSTDMT_compressionJobVoidFn assertNoExtDict; ZSTDMT_compressionJobVoidFn trace; } ZSTDMT_RustCompressionJobTraceProjection; typedef char ZSTDMT_compression_job_trace_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobTraceProjection, callbackContext) == 0 && offsetof(ZSTDMT_RustCompressionJobTraceProjection, firstJob) == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobTraceProjection, assertNoExtDict) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobTraceProjection, trace) == 3 * sizeof(void*) && sizeof(ZSTDMT_RustCompressionJobTraceProjection) == 4 * sizeof(void*)) ? 1 : -1]; typedef struct { int checksumFlag; int ldmEnable; unsigned nbWorkers; } ZSTDMT_RustCompressionJobParameters; typedef char ZSTDMT_compression_job_parameters_layout[ (offsetof(ZSTDMT_RustCompressionJobParameters, checksumFlag) == 0 && offsetof(ZSTDMT_RustCompressionJobParameters, ldmEnable) == sizeof(int) && offsetof(ZSTDMT_RustCompressionJobParameters, nbWorkers) == 2 * sizeof(int) && sizeof(ZSTDMT_RustCompressionJobParameters) == 2 * sizeof(int) + sizeof(unsigned)) ? 1 : -1]; typedef struct { int ldmEnabled; int hasDestination; } ZSTDMT_RustCompressionJobResourceProjection; typedef char ZSTDMT_compression_job_resource_projection_layout[ (offsetof(ZSTDMT_RustCompressionJobResourceProjection, ldmEnabled) == 0 && offsetof(ZSTDMT_RustCompressionJobResourceProjection, hasDestination) == sizeof(int) && sizeof(ZSTDMT_RustCompressionJobResourceProjection) == 2 * sizeof(int)) ? 1 : -1]; typedef int (*ZSTDMT_compressionJobResourceReadyFn)(void* opaque); ZSTDMT_RustCompressionJobParameters ZSTDMT_rust_prepareCompressionJobParameters( unsigned jobID, int checksumFlag, int ldmEnable, unsigned nbWorkers); size_t ZSTDMT_rust_compressionJobAcquireResources( const ZSTDMT_RustCompressionJobResourceProjection* projection, void* opaque, ZSTDMT_compressionJobVoidFn acquireCCtx, ZSTDMT_compressionJobVoidFn acquireRawSeqStore, ZSTDMT_compressionJobResourceReadyFn hasCCtx, ZSTDMT_compressionJobStepFn acquireDestinationBuffer, ZSTDMT_compressionJobResourceReadyFn hasRawSeqStore, ZSTDMT_compressionJobVoidFn publishDestination); size_t ZSTDMT_rust_compressionJobBegin( const ZSTDMT_RustCompressionJobBeginProjection* projection, void* opaque, ZSTDMT_compressionJobStepFn beginWithCDict, ZSTDMT_compressionJobSetParameterFn setForceMaxWindow, ZSTDMT_compressionJobStepFn setDeterministicRefPrefix, ZSTDMT_compressionJobStepFn beginWithPrefix, ZSTDMT_compressionJobVoidFn publishFrameHeader); void ZSTDMT_rust_compressionJob( const ZSTDMT_RustCompressionJobProjection* projection, const ZSTDMT_RustCompressionJobFinishProjection* finishProjection, void* opaque, ZSTDMT_compressionJobStepFn acquireResources, ZSTDMT_compressionJobVoidFn prepareParameters, ZSTDMT_compressionJobVoidFn generateSequences, ZSTDMT_compressionJobStepFn beginJob, ZSTDMT_compressionJobApplySequencesFn applySequences, ZSTDMT_compressionJobCompressFn compressJob, ZSTDMT_compressionJobVoidFn traceJob); void ZSTDMT_rust_compressionJobTrace( const ZSTDMT_RustCompressionJobTraceProjection* projection); ZSTDMT_chunkProcessResult ZSTDMT_rust_compressJobChunks( ZSTD_CCtx* cctx, const void* src, size_t srcSize, void* dst, size_t dstCapacity, size_t chunkSize, unsigned lastJob, void* progressContext, ZSTDMT_chunkProgressFn progressCallback); ZSTDMT_flushPublicationResult ZSTDMT_rust_publishJobOutput( void* outputDst, size_t outputSize, size_t outputPos, const void* jobDst, size_t jobCapacity, size_t cSize, size_t dstFlushed); typedef struct { size_t consumed; size_t cSize; size_t srcSize; const void* dstStart; size_t dstCapacity; size_t dstFlushed; unsigned frameChecksumNeeded; } ZSTDMT_RustFlushJobProjection; typedef struct { void* dst; size_t dstCapacity; size_t cSize; const XXH64_state_t* checksumState; } ZSTDMT_RustFrameChecksumProjection; typedef char ZSTDMT_frame_checksum_projection_layout[ (offsetof(ZSTDMT_RustFrameChecksumProjection, dst) == 0 && offsetof(ZSTDMT_RustFrameChecksumProjection, dstCapacity) == sizeof(void*) && offsetof(ZSTDMT_RustFrameChecksumProjection, cSize) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustFrameChecksumProjection, checksumState) == 3 * sizeof(void*) && sizeof(ZSTDMT_RustFrameChecksumProjection) == 4 * sizeof(void*)) ? 1 : -1]; typedef struct { unsigned doneJobID; unsigned nextJobID; unsigned jobIDMask; unsigned jobReady; unsigned frameEnded; size_t inBuffFilled; } ZSTDMT_RustFlushContextProjection; typedef struct { size_t result; size_t outputPos; unsigned updateAllJobsCompleted; unsigned allJobsCompleted; } ZSTDMT_RustFlushProducedResult; typedef void (*ZSTDMT_flushProjectJobFn)( void* opaque, unsigned jobID, unsigned blockToFlush, ZSTDMT_RustFlushJobProjection* projection); typedef void (*ZSTDMT_flushChecksumFn)( void* opaque, unsigned jobID, ZSTDMT_RustFlushJobProjection* projection); typedef void (*ZSTDMT_flushUpdateJobFn)( void* opaque, unsigned jobID, size_t dstFlushed); typedef void (*ZSTDMT_flushCompleteJobFn)( void* opaque, unsigned jobID, size_t srcSize, size_t cSize); typedef void (*ZSTDMT_flushWaitForAllJobsFn)(void* opaque); typedef void (*ZSTDMT_flushReleaseAllJobResourcesFn)(void* opaque); ZSTDMT_RustFlushProducedResult ZSTDMT_rust_flushProduced( const ZSTDMT_RustFlushContextProjection* context, void* outputDst, size_t outputSize, size_t outputPos, unsigned blockToFlush, unsigned end, void* opaque, ZSTDMT_flushProjectJobFn projectJob, ZSTDMT_flushChecksumFn addFrameChecksum, ZSTDMT_flushUpdateJobFn updateJob, ZSTDMT_flushCompleteJobFn completeJob, ZSTDMT_flushWaitForAllJobsFn waitForAllJobs, ZSTDMT_flushReleaseAllJobResourcesFn releaseAllJobResources); size_t ZSTDMT_rust_writeFrameChecksum( const ZSTDMT_RustFrameChecksumProjection* projection); /* The Rust outer scheduler sees only this scalar snapshot. The MT context, * reusable input buffer, worker pool, and all synchronization remain private * to this translation unit. */ typedef struct { unsigned frameEnded; unsigned jobReady; void* inBuffStart; size_t inBuffCapacity; size_t inBuffFilled; size_t targetSectionSize; int rsyncable; U64 rsyncPrimePower; U64 rsyncHitMask; } ZSTDMT_RustCompressStreamContextProjection; typedef struct { void* bufferStart; size_t bufferCapacity; size_t bufferFilled; } ZSTDMT_RustStreamInputRangeProjection; typedef struct { const void* src; size_t size; size_t pos; } ZSTDMT_RustStreamInputProjection; typedef struct { void* dst; size_t size; size_t pos; } ZSTDMT_RustStreamOutputProjection; typedef struct { size_t toLoad; int flush; } ZSTDMT_RustSyncPointProjection; typedef struct { size_t result; size_t outputPos; } ZSTDMT_RustStreamFlushResult; typedef struct { size_t result; unsigned inBuffActive; } ZSTDMT_RustStreamCreateJobResult; typedef char ZSTDMT_stream_create_job_result_layout[ (offsetof(ZSTDMT_RustStreamCreateJobResult, result) == 0 && offsetof(ZSTDMT_RustStreamCreateJobResult, inBuffActive) == sizeof(size_t) && sizeof(ZSTDMT_RustStreamCreateJobResult) == 2 * sizeof(size_t)) ? 1 : -1]; typedef struct { size_t result; size_t inputPos; size_t outputPos; size_t inBuffFilled; } ZSTDMT_RustCompressStreamResult; typedef char ZSTDMT_compress_stream_result_layout[ (offsetof(ZSTDMT_RustCompressStreamResult, result) == 0 && offsetof(ZSTDMT_RustCompressStreamResult, inputPos) == sizeof(size_t) && offsetof(ZSTDMT_RustCompressStreamResult, outputPos) == 2 * sizeof(size_t) && offsetof(ZSTDMT_RustCompressStreamResult, inBuffFilled) == 3 * sizeof(size_t) && sizeof(ZSTDMT_RustCompressStreamResult) == 4 * sizeof(size_t)) ? 1 : -1]; typedef int (*ZSTDMT_streamTryGetInputRangeFn)( void* opaque, ZSTDMT_RustStreamInputRangeProjection* projection); typedef ZSTDMT_RustStreamCreateJobResult (*ZSTDMT_streamCreateJobFn)( void* opaque, size_t srcSize, unsigned end); typedef ZSTDMT_RustStreamFlushResult (*ZSTDMT_streamFlushProducedFn)( void* opaque, void* outputDst, size_t outputSize, size_t outputPos, unsigned blockToFlush, unsigned end); ZSTDMT_RustCompressStreamResult ZSTDMT_rust_compressStreamGeneric( const ZSTDMT_RustCompressStreamContextProjection* context, const ZSTDMT_RustStreamInputProjection* input, const ZSTDMT_RustStreamOutputProjection* output, unsigned end, void* opaque, ZSTDMT_streamTryGetInputRangeFn tryGetInputRange, ZSTDMT_streamCreateJobFn createJob, ZSTDMT_streamFlushProducedFn flushProduced); typedef struct { unsigned lastJob; size_t srcSize; unsigned firstJob; void* dstStart; size_t dstCapacity; size_t consumed; } ZSTDMT_RustEmptyBlockJobProjection; typedef struct { ZSTDMT_RustBuffer buffer; size_t cSize; unsigned clearSource; } ZSTDMT_RustEmptyBlockResult; typedef ZSTDMT_RustBuffer (*ZSTDMT_bufferGetFn)(void* opaque); ZSTDMT_RustEmptyBlockResult ZSTDMT_rust_writeLastEmptyBlock( const ZSTDMT_RustEmptyBlockJobProjection* projection, void* opaque, ZSTDMT_bufferGetFn getBuffer); typedef struct { unsigned doneJobID; unsigned nextJobID; unsigned jobIDMask; unsigned jobReady; const void* srcStart; size_t srcSize; size_t inBuffFilled; const void* prefixStart; size_t prefixSize; size_t targetPrefixSize; unsigned endFrame; unsigned checksumFlag; } ZSTDMT_RustCreateJobProjection; typedef struct { const void* srcStart; size_t srcSize; const void* prefixStart; size_t prefixSize; const void* nextPrefixStart; size_t nextPrefixSize; size_t roundBuffPosDelta; unsigned jobNumber; unsigned firstJob; unsigned lastJob; unsigned frameChecksumNeeded; unsigned clearChecksumFlag; } ZSTDMT_RustJobInitialization; typedef struct { size_t returnCode; unsigned action; unsigned jobID; unsigned jobNumber; unsigned nextJobID; unsigned jobReady; } ZSTDMT_RustCreateJobResult; enum { ZSTDMT_CREATE_JOB_TABLE_FULL = 0, ZSTDMT_CREATE_JOB_POST = 1, ZSTDMT_CREATE_JOB_EMPTY = 2 }; typedef void (*ZSTDMT_prepareJobFn)(void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* init); typedef void (*ZSTDMT_prepareFrameStateFn)(void* opaque); typedef void (*ZSTDMT_writeEmptyJobFn)(void* opaque, unsigned jobID); typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID); void ZSTDMT_rust_prepareCompressionJob( void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* initialization, ZSTDMT_prepareJobFn initializeJob, ZSTDMT_prepareJobFn resetInputState, ZSTDMT_prepareFrameStateFn setFrameEnded, ZSTDMT_prepareFrameStateFn clearChecksumFlag); ZSTDMT_RustCreateJobResult ZSTDMT_rust_createCompressionJob( const ZSTDMT_RustCreateJobProjection* projection, void* opaque, ZSTDMT_prepareJobFn prepareJob, ZSTDMT_writeEmptyJobFn writeEmptyJob, ZSTDMT_tryAddJobFn tryAddJob); typedef struct { rawSeq* seq; size_t pos; size_t posInSequence; size_t size; size_t capacity; } ZSTDMT_RustRawSeqStore; typedef char ZSTDMT_rust_raw_seq_layout[ (sizeof(rawSeq) == 3 * sizeof(U32)) ? 1 : -1]; typedef char ZSTDMT_rust_raw_seq_store_layout[ (sizeof(ZSTDMT_RustRawSeqStore) == sizeof(RawSeqStore_t) && offsetof(ZSTDMT_RustRawSeqStore, pos) == offsetof(RawSeqStore_t, pos) && offsetof(ZSTDMT_RustRawSeqStore, capacity) == offsetof(RawSeqStore_t, capacity)) ? 1 : -1]; struct ZSTDMT_RustCompressionJobSequenceState_s { ZSTD_CCtx** cctx; ZSTDMT_RustRawSeqStore* rawSeqStore; int ldmEnabled; }; typedef char ZSTDMT_compression_job_sequence_state_layout[ (offsetof(ZSTDMT_RustCompressionJobSequenceState, cctx) == 0 && offsetof(ZSTDMT_RustCompressionJobSequenceState, rawSeqStore) == sizeof(void*) && offsetof(ZSTDMT_RustCompressionJobSequenceState, ldmEnabled) == 2 * sizeof(void*) && sizeof(ZSTDMT_RustCompressionJobSequenceState) == (sizeof(void*) == 8 ? 24 : 12)) ? 1 : -1]; ZSTDMT_RustRawSeqStore ZSTDMT_rust_bufferToSeq(ZSTDMT_RustBuffer buffer); ZSTDMT_RustBuffer ZSTDMT_rust_seqToBuffer(ZSTDMT_RustRawSeqStore seq); typedef int (*ZSTDMT_serialWaitForTurnFn)(void* opaque, unsigned jobID); typedef void (*ZSTDMT_serialLdmWindowUpdateFn)( void* opaque, ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize); typedef void (*ZSTDMT_serialLdmGenerateSequencesFn)( void* opaque, ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize); typedef void (*ZSTDMT_serialLdmPublishWindowFn)(void* opaque); typedef void (*ZSTDMT_serialAdvanceFn)(void* opaque); typedef void (*ZSTDMT_serialStateLockFn)(void* opaque); typedef void (*ZSTDMT_serialStateWaitFn)(void* opaque); typedef struct { void* callbackContext; unsigned* nextJobID; ZSTDMT_serialStateLockFn lock; ZSTDMT_serialStateWaitFn wait; unsigned jobID; } ZSTDMT_RustSerialWaitForTurnState; typedef char ZSTDMT_rust_serial_wait_for_turn_state_layout[ (offsetof(ZSTDMT_RustSerialWaitForTurnState, callbackContext) == 0 && offsetof(ZSTDMT_RustSerialWaitForTurnState, nextJobID) == sizeof(void*) && offsetof(ZSTDMT_RustSerialWaitForTurnState, lock) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustSerialWaitForTurnState, wait) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustSerialWaitForTurnState, jobID) == 4 * sizeof(void*) && sizeof(ZSTDMT_RustSerialWaitForTurnState) == 5 * sizeof(void*)) ? 1 : -1]; int ZSTDMT_rust_serialStateWaitForTurn( const ZSTDMT_RustSerialWaitForTurnState* state); typedef void (*ZSTDMT_serialStateBroadcastFn)(void* opaque); typedef void (*ZSTDMT_serialStateUnlockFn)(void* opaque); typedef struct { void* callbackContext; unsigned* nextJobID; ZSTDMT_serialStateBroadcastFn broadcast; ZSTDMT_serialStateUnlockFn unlock; } ZSTDMT_RustSerialAdvanceState; typedef char ZSTDMT_rust_serial_advance_state_layout[ (offsetof(ZSTDMT_RustSerialAdvanceState, callbackContext) == 0 && offsetof(ZSTDMT_RustSerialAdvanceState, nextJobID) == sizeof(void*) && offsetof(ZSTDMT_RustSerialAdvanceState, broadcast) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustSerialAdvanceState, unlock) == 3 * sizeof(void*) && sizeof(ZSTDMT_RustSerialAdvanceState) == 4 * sizeof(void*)) ? 1 : -1]; void ZSTDMT_rust_serialStateAdvance(const ZSTDMT_RustSerialAdvanceState* state); void ZSTDMT_rust_serialStateGenSequences( ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize, unsigned jobID, int ldmEnabled, int checksumEnabled, void* opaque, XXH64_state_t* checksumState, ZSTDMT_serialWaitForTurnFn waitForTurn, ZSTDMT_serialLdmWindowUpdateFn updateLdmWindow, ZSTDMT_serialLdmGenerateSequencesFn generateLdmSequences, ZSTDMT_serialLdmPublishWindowFn publishLdmWindow, ZSTDMT_serialAdvanceFn advance); typedef struct { unsigned skip; unsigned nextJobID; } ZSTDMT_RustSerialEnsureFinishedResult; ZSTDMT_RustSerialEnsureFinishedResult ZSTDMT_rust_serialStateEnsureFinished( unsigned nextJobID, unsigned jobID); typedef void (*ZSTDMT_serialStateCallbackFn)(void* opaque); typedef void (*ZSTDMT_serialStateSkipFn)( void* opaque, unsigned jobID, size_t cSize); typedef struct { void* callbackContext; unsigned* nextJobID; ZSTDMT_serialStateCallbackFn lock; ZSTDMT_serialStateCallbackFn broadcast; ZSTDMT_serialStateCallbackFn ldmLock; ZSTDMT_serialStateCallbackFn clearLdmWindow; ZSTDMT_serialStateCallbackFn ldmSignal; ZSTDMT_serialStateCallbackFn ldmUnlock; ZSTDMT_serialStateCallbackFn unlock; ZSTDMT_serialStateSkipFn onSkip; size_t cSize; unsigned jobID; } ZSTDMT_RustSerialEnsureFinishedState; typedef char ZSTDMT_rust_serial_ensure_finished_state_layout[ (offsetof(ZSTDMT_RustSerialEnsureFinishedState, callbackContext) == 0 && offsetof(ZSTDMT_RustSerialEnsureFinishedState, nextJobID) == sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, lock) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, broadcast) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmLock) == 4 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, clearLdmWindow) == 5 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmSignal) == 6 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, ldmUnlock) == 7 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, unlock) == 8 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, onSkip) == 9 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, cSize) == 10 * sizeof(void*) && offsetof(ZSTDMT_RustSerialEnsureFinishedState, jobID) == 10 * sizeof(void*) + sizeof(size_t) && sizeof(ZSTDMT_RustSerialEnsureFinishedState) == 12 * sizeof(void*)) ? 1 : -1]; void ZSTDMT_rust_serialStateEnsureFinishedOrchestrated( const ZSTDMT_RustSerialEnsureFinishedState* state); typedef void (*ZSTDMT_waitForJobCompleteFn)( void* opaque, unsigned jobID, unsigned doneJobID); unsigned ZSTDMT_rust_waitForAllJobsCompleted( unsigned doneJobID, unsigned nextJobID, unsigned jobIDMask, void* opaque, ZSTDMT_waitForJobCompleteFn waitForJob); typedef void (*ZSTDMT_releaseJobResourceFn)(void* opaque, unsigned jobID); void ZSTDMT_rust_releaseAllJobResources( unsigned jobIDMask, void* opaque, ZSTDMT_releaseJobResourceFn releaseJob); typedef void (*ZSTDMT_waitForLdmLockFn)(void* opaque); typedef int (*ZSTDMT_waitForLdmOverlapFn)( void* opaque, void* bufferStart, size_t bufferCapacity); typedef void (*ZSTDMT_waitForLdmWaitFn)(void* opaque); typedef void (*ZSTDMT_waitForLdmUnlockFn)(void* opaque); typedef struct { void* callbackContext; void* bufferStart; size_t bufferCapacity; int ldmEnabled; ZSTDMT_waitForLdmLockFn lock; ZSTDMT_waitForLdmOverlapFn overlaps; ZSTDMT_waitForLdmWaitFn wait; ZSTDMT_waitForLdmUnlockFn unlock; } ZSTDMT_RustWaitForLdmState; typedef char ZSTDMT_rust_wait_for_ldm_state_layout[ (offsetof(ZSTDMT_RustWaitForLdmState, callbackContext) == 0 && offsetof(ZSTDMT_RustWaitForLdmState, bufferStart) == sizeof(void*) && offsetof(ZSTDMT_RustWaitForLdmState, bufferCapacity) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustWaitForLdmState, ldmEnabled) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustWaitForLdmState, lock) == 4 * sizeof(void*) && offsetof(ZSTDMT_RustWaitForLdmState, overlaps) == 5 * sizeof(void*) && offsetof(ZSTDMT_RustWaitForLdmState, wait) == 6 * sizeof(void*) && offsetof(ZSTDMT_RustWaitForLdmState, unlock) == 7 * sizeof(void*) && sizeof(ZSTDMT_RustWaitForLdmState) == 8 * sizeof(void*)) ? 1 : -1]; void ZSTDMT_rust_waitForLdmComplete( const ZSTDMT_RustWaitForLdmState* state); unsigned ZSTDMT_rust_computeTargetJobLog(unsigned windowLog, unsigned chainLog, int strategy, int enableLdm); int ZSTDMT_rust_overlapLog(int overlapLog, int strategy); size_t ZSTDMT_rust_computeOverlapSize(unsigned windowLog, unsigned chainLog, int strategy, int overlapLog, int enableLdm); int ZSTDMT_rust_doesOverlapWindow(const void* bufferStart, size_t bufferCapacity, const void* nextSrc, const void* base, const void* dictBase, U32 dictLimit, U32 lowLimit); void ZSTDMT_rust_findSynchronizationPoint(const void* inputSrc, size_t inputSize, size_t inputPos, size_t targetSectionSize, const void* inBuffStart, size_t inBuffFilled, int rsyncable, U64 primePower, U64 hitMask, size_t* toLoad, int* flush); U64 ZSTDMT_rust_rollingHashPrimePower(U32 length); size_t ZSTDMT_rust_nextInputSizeHint(size_t targetSectionSize, size_t inBuffFilled); typedef struct { unsigned requestedNbWorkers; unsigned currentNbWorkers; size_t jobSize; size_t jobSizeMin; size_t jobSizeMax; int enableLdm; unsigned windowLog; unsigned chainLog; int strategy; int overlapLog; int rsyncable; size_t roundBuffCapacity; unsigned allJobsCompleted; } ZSTDMT_RustInitCStreamProjection; typedef size_t (*ZSTDMT_initResizeFn)(void* opaque, unsigned nbWorkers); typedef void (*ZSTDMT_initWaitForAllJobsFn)(void* opaque); typedef void (*ZSTDMT_initReleaseAllJobResourcesFn)(void* opaque); typedef void (*ZSTDMT_initApplyParametersFn)(void* opaque, size_t jobSize); typedef size_t (*ZSTDMT_initDictionaryFn)(void* opaque); typedef void (*ZSTDMT_initSetSizeFn)(void* opaque, size_t size); typedef void (*ZSTDMT_initSetRsyncFn)(void* opaque, U64 hitMask, U64 primePower); typedef void (*ZSTDMT_initSetBufferSizeFn)(void* opaque, size_t size); typedef size_t (*ZSTDMT_initResizeRoundBufferFn)(void* opaque, size_t capacity); typedef void (*ZSTDMT_initResetStreamFn)(void* opaque); typedef size_t (*ZSTDMT_initSerialResetFn)(void* opaque, size_t targetSectionSize); typedef struct { unsigned hasDictionary; unsigned rawContent; } ZSTDMT_RustDictionaryProjection; typedef char ZSTDMT_rust_dictionary_projection_layout[ (offsetof(ZSTDMT_RustDictionaryProjection, hasDictionary) == 0 && offsetof(ZSTDMT_RustDictionaryProjection, rawContent) == sizeof(unsigned) && sizeof(ZSTDMT_RustDictionaryProjection) == 2 * sizeof(unsigned)) ? 1 : -1]; size_t ZSTDMT_rust_prepareCStreamDictionary( const ZSTDMT_RustDictionaryProjection* projection, void* opaque, ZSTDMT_initResetStreamFn releaseLocal, ZSTDMT_initDictionaryFn createCopied, ZSTDMT_initResetStreamFn attachBorrowed); size_t ZSTDMT_rust_updateCStreamDictionary( const ZSTDMT_RustDictionaryProjection* projection, void* opaque, ZSTDMT_initResetStreamFn clearLocal, ZSTDMT_initResetStreamFn attachBorrowed, ZSTDMT_initResetStreamFn setRawPrefix, ZSTDMT_initDictionaryFn createReferenced); typedef struct { int enableLdm; int checksumEnabled; size_t jobSize; unsigned windowLog; int strategy; unsigned hashLog; unsigned bucketSizeLog; unsigned minMatchLength; unsigned hashRateLog; unsigned previousHashLog; unsigned previousBucketSizeLog; size_t ldmEntrySize; } ZSTDMT_RustSerialResetProjection; typedef char ZSTDMT_rust_serial_reset_projection_layout[ (offsetof(ZSTDMT_RustSerialResetProjection, enableLdm) == 0 && offsetof(ZSTDMT_RustSerialResetProjection, checksumEnabled) == sizeof(int) && offsetof(ZSTDMT_RustSerialResetProjection, jobSize) == 8 && offsetof(ZSTDMT_RustSerialResetProjection, windowLog) == (sizeof(void*) == 8 ? 16 : 12) && offsetof(ZSTDMT_RustSerialResetProjection, strategy) == (sizeof(void*) == 8 ? 20 : 16) && offsetof(ZSTDMT_RustSerialResetProjection, hashLog) == (sizeof(void*) == 8 ? 24 : 20) && offsetof(ZSTDMT_RustSerialResetProjection, bucketSizeLog) == (sizeof(void*) == 8 ? 28 : 24) && offsetof(ZSTDMT_RustSerialResetProjection, minMatchLength) == (sizeof(void*) == 8 ? 32 : 28) && offsetof(ZSTDMT_RustSerialResetProjection, hashRateLog) == (sizeof(void*) == 8 ? 36 : 32) && offsetof(ZSTDMT_RustSerialResetProjection, previousHashLog) == (sizeof(void*) == 8 ? 40 : 36) && offsetof(ZSTDMT_RustSerialResetProjection, previousBucketSizeLog) == (sizeof(void*) == 8 ? 44 : 40) && offsetof(ZSTDMT_RustSerialResetProjection, ldmEntrySize) == (sizeof(void*) == 8 ? 48 : 44) && sizeof(ZSTDMT_RustSerialResetProjection) == (sizeof(void*) == 8 ? 56 : 48)) ? 1 : -1]; typedef void (*ZSTDMT_serialResetSetLdmParamsFn)( void* opaque, int enableLdm, unsigned hashLog, unsigned bucketSizeLog, unsigned minMatchLength, unsigned hashRateLog, unsigned windowLog); typedef void (*ZSTDMT_serialResetVoidFn)(void* opaque); typedef void (*ZSTDMT_serialResetSetNbSeqFn)(void* opaque, size_t nbSeq); typedef int (*ZSTDMT_serialResetResizeFn)( void* opaque, size_t hashSize, size_t numBuckets, unsigned bucketLog, unsigned previousBucketLog); typedef void (*ZSTDMT_serialResetZeroTablesFn)( void* opaque, size_t hashSize, size_t numBuckets); typedef size_t (*ZSTDMT_serialResetLoadRawDictionaryFn)( void* opaque, const void* dict, size_t dictSize); typedef void (*ZSTDMT_serialResetSetLoadedDictEndFn)( void* opaque, U32 loadedDictEnd); typedef void (*ZSTDMT_serialResetPublishFn)(void* opaque, size_t jobSize); typedef struct { const void* dict; size_t dictSize; int dictContentType; int forceWindow; } ZSTDMT_RustSerialDictionaryProjection; typedef char ZSTDMT_rust_serial_dictionary_projection_layout[ (offsetof(ZSTDMT_RustSerialDictionaryProjection, dict) == 0 && offsetof(ZSTDMT_RustSerialDictionaryProjection, dictSize) == sizeof(void*) && offsetof(ZSTDMT_RustSerialDictionaryProjection, dictContentType) == sizeof(void*) + sizeof(size_t) && offsetof(ZSTDMT_RustSerialDictionaryProjection, forceWindow) == sizeof(void*) + sizeof(size_t) + sizeof(int) && sizeof(ZSTDMT_RustSerialDictionaryProjection) == sizeof(void*) + sizeof(size_t) + 2 * sizeof(int)) ? 1 : -1]; void ZSTDMT_rust_serialStateLoadDictionary( const ZSTDMT_RustSerialDictionaryProjection* projection, void* opaque, ZSTDMT_serialResetLoadRawDictionaryFn loadRawDictionary, ZSTDMT_serialResetSetLoadedDictEndFn setLoadedDictEnd); int ZSTDMT_rust_serialStateReset( const ZSTDMT_RustSerialResetProjection* projection, void* opaque, ZSTDMT_serialResetSetLdmParamsFn setLdmParams, ZSTDMT_serialResetVoidFn resetNextJob, ZSTDMT_serialResetVoidFn resetChecksum, ZSTDMT_serialResetSetNbSeqFn setNbSeq, ZSTDMT_serialResetVoidFn resetWindow, ZSTDMT_serialResetResizeFn resizeTables, ZSTDMT_serialResetZeroTablesFn zeroTables, ZSTDMT_serialResetVoidFn loadDictionary, ZSTDMT_serialResetVoidFn copyWindow, ZSTDMT_serialResetPublishFn publishParams); typedef void (*ZSTDMT_serialStateVoidFn)(void* opaque); typedef int (*ZSTDMT_serialStateInitFn)(void* opaque); int ZSTDMT_rust_serialStateInit( void* opaque, ZSTDMT_serialStateVoidFn zeroState, ZSTDMT_serialStateInitFn initMutex, ZSTDMT_serialStateInitFn initCond, ZSTDMT_serialStateInitFn initLdmMutex, ZSTDMT_serialStateInitFn initLdmCond); void ZSTDMT_rust_serialStateFree( void* opaque, ZSTDMT_serialStateVoidFn destroyMutex, ZSTDMT_serialStateVoidFn destroyCond, ZSTDMT_serialStateVoidFn destroyLdmMutex, ZSTDMT_serialStateVoidFn destroyLdmCond, ZSTDMT_serialStateVoidFn freeTables); typedef void (*ZSTDMT_freeCCtxCallbackFn)(void* opaque); typedef struct { void* callbackContext; size_t providedFactory; size_t roundBuffPresent; ZSTDMT_freeCCtxCallbackFn freeFactory; ZSTDMT_freeCCtxCallbackFn releaseAllJobResources; ZSTDMT_freeCCtxCallbackFn freeJobs; ZSTDMT_freeCCtxCallbackFn freeBufferPool; ZSTDMT_freeCCtxCallbackFn freeCCtxPool; ZSTDMT_freeCCtxCallbackFn freeSeqPool; ZSTDMT_freeCCtxCallbackFn freeSerialState; ZSTDMT_freeCCtxCallbackFn freeCDict; ZSTDMT_freeCCtxCallbackFn freeRoundBuffer; ZSTDMT_freeCCtxCallbackFn freeMTctx; } ZSTDMT_RustFreeCCtxState; typedef char ZSTDMT_rust_free_cctx_state_layout[ (sizeof(ZSTDMT_freeCCtxCallbackFn) == sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, callbackContext) == 0 && offsetof(ZSTDMT_RustFreeCCtxState, providedFactory) == sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, roundBuffPresent) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeFactory) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, releaseAllJobResources) == 4 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeJobs) == 5 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeBufferPool) == 6 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeCCtxPool) == 7 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeSeqPool) == 8 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeSerialState) == 9 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeCDict) == 10 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeRoundBuffer) == 11 * sizeof(void*) && offsetof(ZSTDMT_RustFreeCCtxState, freeMTctx) == 12 * sizeof(void*) && sizeof(ZSTDMT_RustFreeCCtxState) == 13 * sizeof(void*)) ? 1 : -1]; size_t ZSTDMT_rust_freeCCtx(const ZSTDMT_RustFreeCCtxState* state); typedef struct { void* callbackContext; unsigned requestedNbWorkers; unsigned maxNbWorkers; size_t contextSize; ZSTD_customMem customMem; } ZSTDMT_RustCreateCCtxProjection; typedef char ZSTDMT_rust_create_cctx_projection_layout[ (offsetof(ZSTDMT_RustCreateCCtxProjection, callbackContext) == 0 && offsetof(ZSTDMT_RustCreateCCtxProjection, requestedNbWorkers) == sizeof(void*) && offsetof(ZSTDMT_RustCreateCCtxProjection, maxNbWorkers) == sizeof(void*) + sizeof(unsigned) && offsetof(ZSTDMT_RustCreateCCtxProjection, contextSize) == (sizeof(void*) == 8 ? 16 : 12) && offsetof(ZSTDMT_RustCreateCCtxProjection, customMem) == (sizeof(void*) == 8 ? 24 : 16) && sizeof(ZSTDMT_RustCreateCCtxProjection) == (sizeof(void*) == 8 ? 48 : 28)) ? 1 : -1]; typedef void* (*ZSTDMT_createCCtxAllocateFn)(void* opaque, size_t size); typedef size_t (*ZSTDMT_createCCtxSetWorkersFn)(void* opaque, unsigned nbWorkers); typedef void (*ZSTDMT_createCCtxSetInitialStateFn)(void* opaque); typedef void* (*ZSTDMT_createCCtxFactoryFn)(void* opaque, unsigned nbWorkers); typedef void* (*ZSTDMT_createCCtxJobsFn)(void* opaque, unsigned* nbJobs); typedef void* (*ZSTDMT_createCCtxResourceFn)(void* opaque, unsigned nbWorkers); typedef int (*ZSTDMT_createCCtxSerialInitFn)(void* opaque); typedef void (*ZSTDMT_createCCtxFreeFn)(void* opaque); void* ZSTDMT_rust_createCCtx( const ZSTDMT_RustCreateCCtxProjection* projection, ZSTDMT_createCCtxAllocateFn allocate, ZSTDMT_createCCtxSetWorkersFn setWorkers, ZSTDMT_createCCtxSetInitialStateFn setInitialState, ZSTDMT_createCCtxFactoryFn createFactory, ZSTDMT_createCCtxJobsFn createJobs, ZSTDMT_createCCtxResourceFn createBufferPool, ZSTDMT_createCCtxResourceFn createCCtxPool, ZSTDMT_createCCtxResourceFn createSeqPool, ZSTDMT_createCCtxSerialInitFn initSerial, ZSTDMT_createCCtxFreeFn freeContext); size_t ZSTDMT_rust_initCStream( const ZSTDMT_RustInitCStreamProjection* projection, void* opaque, ZSTDMT_initResizeFn resize, ZSTDMT_initWaitForAllJobsFn waitForAllJobs, ZSTDMT_initReleaseAllJobResourcesFn releaseAllJobResources, ZSTDMT_initApplyParametersFn applyParameters, ZSTDMT_initDictionaryFn prepareDictionary, ZSTDMT_initSetSizeFn setTargetPrefixSize, ZSTDMT_initSetSizeFn setTargetSectionSize, ZSTDMT_initSetRsyncFn setRsync, ZSTDMT_initSetBufferSizeFn setBufferSize, ZSTDMT_initResizeRoundBufferFn resizeRoundBuffer, ZSTDMT_initResetStreamFn resetStream, ZSTDMT_initDictionaryFn updateDictionary, ZSTDMT_initSerialResetFn serialReset); size_t ZSTDMT_rust_initSerialResetResult(int status); typedef void (*ZSTDMT_resetStreamCallbackFn)(void* opaque); typedef struct { void* callbackContext; ZSTDMT_resetStreamCallbackFn resetRoundBuffer; ZSTDMT_resetStreamCallbackFn resetInputBuffer; ZSTDMT_resetStreamCallbackFn resetJobIDs; ZSTDMT_resetStreamCallbackFn resetFrameFlags; ZSTDMT_resetStreamCallbackFn resetProgress; } ZSTDMT_RustResetStreamState; typedef char ZSTDMT_rust_reset_stream_state_layout[ (sizeof(ZSTDMT_resetStreamCallbackFn) == sizeof(void*) && offsetof(ZSTDMT_RustResetStreamState, callbackContext) == 0 && offsetof(ZSTDMT_RustResetStreamState, resetRoundBuffer) == sizeof(void*) && offsetof(ZSTDMT_RustResetStreamState, resetInputBuffer) == 2 * sizeof(void*) && offsetof(ZSTDMT_RustResetStreamState, resetJobIDs) == 3 * sizeof(void*) && offsetof(ZSTDMT_RustResetStreamState, resetFrameFlags) == 4 * sizeof(void*) && offsetof(ZSTDMT_RustResetStreamState, resetProgress) == 5 * sizeof(void*) && sizeof(ZSTDMT_RustResetStreamState) == 6 * sizeof(void*)) ? 1 : -1]; void ZSTDMT_rust_resetStream(const ZSTDMT_RustResetStreamState* state); typedef struct { void* callbackContext; unsigned nbWorkers; } ZSTDMT_RustResizeProjection; typedef char ZSTDMT_rust_resize_projection_layout[ (offsetof(ZSTDMT_RustResizeProjection, callbackContext) == 0 && offsetof(ZSTDMT_RustResizeProjection, nbWorkers) == sizeof(void*) && sizeof(ZSTDMT_RustResizeProjection) == (sizeof(void*) == 8 ? 16 : 8)) ? 1 : -1]; typedef size_t (*ZSTDMT_resizeStepFn)(void* opaque, unsigned nbWorkers); typedef void* (*ZSTDMT_resizeResourceFn)(void* opaque, unsigned nbWorkers); size_t ZSTDMT_rust_resize( const ZSTDMT_RustResizeProjection* projection, ZSTDMT_resizeStepFn resizeFactory, ZSTDMT_resizeStepFn expandJobs, ZSTDMT_resizeResourceFn expandBufferPool, ZSTDMT_resizeResourceFn expandCCtxPool, ZSTDMT_resizeResourceFn expandSeqPool, ZSTDMT_resizeStepFn setNbWorkers); typedef struct { size_t consumed; size_t cSize; const void* srcStart; size_t srcSize; const void* prefixStart; size_t prefixSize; size_t dstFlushed; } ZSTDMT_RustJobProjection; typedef struct { const void* start; size_t size; } ZSTDMT_RustInputRange; typedef struct { const void* roundBufferStart; size_t roundBufferCapacity; size_t roundBufferPos; const void* prefixStart; size_t prefixSize; size_t targetSectionSize; const void* inUseStart; size_t inUseSize; } ZSTDMT_RustTryGetInputRangeProjection; typedef struct { unsigned ready; unsigned movePrefix; void* bufferStart; size_t bufferCapacity; size_t roundBufferPos; } ZSTDMT_RustTryGetInputRangeResult; typedef char ZSTDMT_rust_try_get_input_range_result_layout[ (offsetof(ZSTDMT_RustTryGetInputRangeResult, ready) == 0 && offsetof(ZSTDMT_RustTryGetInputRangeResult, movePrefix) == sizeof(unsigned) && offsetof(ZSTDMT_RustTryGetInputRangeResult, bufferStart) == 2 * sizeof(unsigned) && offsetof(ZSTDMT_RustTryGetInputRangeResult, bufferCapacity) == 2 * sizeof(unsigned) + sizeof(void*) && offsetof(ZSTDMT_RustTryGetInputRangeResult, roundBufferPos) == 2 * sizeof(unsigned) + sizeof(void*) + sizeof(size_t) && sizeof(ZSTDMT_RustTryGetInputRangeResult) == 2 * sizeof(unsigned) + sizeof(void*) + 2 * sizeof(size_t)) ? 1 : -1]; typedef void (*ZSTDMT_jobProjectionFn)(void* opaque, unsigned jobID, ZSTDMT_RustJobProjection* projection); ZSTDMT_RustInputRange ZSTDMT_rust_getInputDataInUse( unsigned firstJobID, unsigned lastJobID, unsigned jobIDMask, size_t roundBufferCapacity, size_t targetSectionSize, void* opaque, ZSTDMT_jobProjectionFn projectJob); ZSTDMT_RustTryGetInputRangeResult ZSTDMT_rust_tryGetInputRange( const ZSTDMT_RustTryGetInputRangeProjection* projection); typedef void (*ZSTDMT_inputRangeWaitFn)( void* opaque, void* bufferStart, size_t bufferCapacity); typedef void (*ZSTDMT_inputRangeMovePrefixFn)( void* opaque, void* destination, const void* source, size_t size, size_t roundBufferPos); typedef void (*ZSTDMT_inputRangePublishBufferFn)( void* opaque, void* bufferStart, size_t bufferCapacity); int ZSTDMT_rust_commitInputRange( const ZSTDMT_RustTryGetInputRangeResult* result, void* roundBufferStart, const void* prefixStart, size_t prefixSize, void* opaque, ZSTDMT_inputRangeWaitFn waitForLdm, ZSTDMT_inputRangeMovePrefixFn movePrefix, ZSTDMT_inputRangePublishBufferFn publishBuffer); size_t ZSTDMT_rust_toFlushNow(unsigned doneJobID, unsigned nextJobID, unsigned jobIDMask, void* opaque, ZSTDMT_jobProjectionFn projectJob); size_t ZSTDMT_rust_sizeofCCtx(size_t mtctxSize, size_t factorySize, size_t bufferPoolSize, size_t jobsSize, size_t cctxPoolSize, size_t seqPoolSize, size_t cdictSize, size_t roundBuffSize); ZSTD_frameProgression ZSTDMT_rust_frameProgression( unsigned long long consumed, size_t inBuffFilled, unsigned long long produced, unsigned currentJobID); ZSTD_frameProgression ZSTDMT_rust_frameProgressionAddJob( ZSTD_frameProgression progression, size_t srcSize, size_t consumed, size_t produced, size_t flushed); ZSTD_frameProgression ZSTDMT_rust_frameProgressionWithJobs( unsigned long long consumed, size_t inBuffFilled, unsigned long long produced, unsigned currentJobID, unsigned doneJobID, unsigned nextJobID, unsigned jobReady, unsigned jobIDMask, void* opaque, ZSTDMT_jobProjectionFn projectJob); typedef int (*ZSTDMT_jobTableInitFn)(void* jobTable, unsigned nbJobs, size_t jobSize); typedef void (*ZSTDMT_jobTableDestroyFn)(void* jobTable, unsigned nbJobs, size_t jobSize); typedef struct { void* jobTable; unsigned nbJobs; size_t jobSize; ZSTD_customMem cMem; ZSTDMT_jobTableDestroyFn destroySync; } ZSTDMT_RustFreeJobsTableProjection; typedef char ZSTDMT_rust_free_jobs_table_projection_layout[ (sizeof(ZSTDMT_jobTableDestroyFn) == sizeof(void*) && offsetof(ZSTDMT_RustFreeJobsTableProjection, jobTable) == 0 && offsetof(ZSTDMT_RustFreeJobsTableProjection, nbJobs) == sizeof(void*) && offsetof(ZSTDMT_RustFreeJobsTableProjection, jobSize) == (sizeof(void*) == 8 ? 16 : 8) && offsetof(ZSTDMT_RustFreeJobsTableProjection, cMem) == offsetof(ZSTDMT_RustFreeJobsTableProjection, jobSize) + sizeof(size_t) && offsetof(ZSTDMT_RustFreeJobsTableProjection, destroySync) == offsetof(ZSTDMT_RustFreeJobsTableProjection, cMem) + sizeof(ZSTD_customMem) && sizeof(ZSTDMT_RustFreeJobsTableProjection) == offsetof(ZSTDMT_RustFreeJobsTableProjection, destroySync) + sizeof(ZSTDMT_jobTableDestroyFn)) ? 1 : -1]; void* ZSTDMT_rust_createJobsTable( unsigned* nbJobsPtr, size_t jobSize, ZSTD_customMem cMem, ZSTDMT_jobTableInitFn initSync, ZSTDMT_jobTableDestroyFn destroySync); size_t ZSTDMT_rust_expandJobsTable( void** jobTablePtr, unsigned* jobIDMaskPtr, unsigned nbWorkers, size_t jobSize, ZSTD_customMem cMem, ZSTDMT_jobTableInitFn initSync, ZSTDMT_jobTableDestroyFn destroySync); void ZSTDMT_rust_freeJobsTable( const ZSTDMT_RustFreeJobsTableProjection* projection); typedef struct ZSTDMT_bufferPool_s { ZSTDMT_RustBufferPool* rustPool; ZSTD_customMem cMem; } ZSTDMT_bufferPool; static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool) { if (!bufPool) return; /* compatibility with free on NULL */ ZSTDMT_rust_buffer_pool_free(bufPool->rustPool); ZSTD_customFree(bufPool, bufPool->cMem); } static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned maxNbBuffers, ZSTD_customMem cMem) { ZSTDMT_bufferPool* const bufPool = (ZSTDMT_bufferPool*)ZSTD_customCalloc(sizeof(ZSTDMT_bufferPool), cMem); if (bufPool==NULL) return NULL; bufPool->rustPool = ZSTDMT_rust_buffer_pool_create(maxNbBuffers, cMem); if (bufPool->rustPool == NULL) { ZSTD_customFree(bufPool, cMem); return NULL; } bufPool->cMem = cMem; return bufPool; } /* only works at initialization, not during compression */ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool) { if (bufPool == NULL) return 0; return ZSTDMT_rust_sizeofBufferPool( sizeof(*bufPool), ZSTDMT_rust_buffer_pool_sizeof(bufPool->rustPool)); } /* ZSTDMT_setBufferSize() : * all future buffers provided by this buffer pool will have _at least_ this size * note : it's better for all buffers to have same size, * as they become freely interchangeable, reducing malloc/free usages and memory fragmentation */ static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* const bufPool, size_t const bSize) { DEBUGLOG(4, "ZSTDMT_setBufferSize: bSize = %u", (U32)bSize); ZSTDMT_rust_buffer_pool_set_size(bufPool->rustPool, bSize); } static ZSTDMT_bufferPool* ZSTDMT_expandBufferPool(ZSTDMT_bufferPool* srcBufPool, unsigned maxNbBuffers) { if (srcBufPool==NULL) return NULL; srcBufPool->rustPool = ZSTDMT_rust_buffer_pool_expand(srcBufPool->rustPool, maxNbBuffers); if (srcBufPool->rustPool == NULL) { ZSTD_customMem const cMem = srcBufPool->cMem; ZSTD_customFree(srcBufPool, cMem); return NULL; } return srcBufPool; } /** ZSTDMT_getBuffer() : * assumption : bufPool must be valid * @return : a buffer, with start pointer and size * note: allocation may fail, in this case, start==NULL and size==0 */ static Buffer ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool) { ZSTDMT_RustBuffer const rustBuffer = ZSTDMT_rust_buffer_pool_get(bufPool->rustPool); Buffer buffer = { rustBuffer.start, rustBuffer.capacity }; return buffer; } static ZSTDMT_RustBuffer ZSTDMT_getBufferForRust(void* opaque) { Buffer const buffer = ZSTDMT_getBuffer((ZSTDMT_bufferPool*)opaque); ZSTDMT_RustBuffer const result = { buffer.start, buffer.capacity }; return result; } #if ZSTD_RESIZE_SEQPOOL /** ZSTDMT_resizeBuffer() : * assumption : bufPool must be valid * @return : a buffer that is at least the buffer pool buffer size. * If a reallocation happens, the data in the input buffer is copied. */ static Buffer ZSTDMT_resizeBuffer(ZSTDMT_bufferPool* bufPool, Buffer buffer) { ZSTDMT_RustBuffer const rustBuffer = { buffer.start, buffer.capacity }; ZSTDMT_RustBuffer const resized = ZSTDMT_rust_buffer_pool_resize(bufPool->rustPool, rustBuffer); Buffer const result = { resized.start, resized.capacity }; return result; } #endif /* store buffer for later re-use, up to pool capacity */ static void ZSTDMT_releaseBuffer(ZSTDMT_bufferPool* bufPool, Buffer buf) { ZSTDMT_RustBuffer const rustBuffer = { buf.start, buf.capacity }; ZSTDMT_rust_buffer_pool_release(bufPool->rustPool, rustBuffer); } /* We need 2 output buffers per worker since each dstBuff must be flushed after it is released. * The 3 additional buffers are as follows: * 1 buffer for input loading * 1 buffer for "next input" when submitting current one * 1 buffer stuck in queue */ #define BUF_POOL_MAX_NB_BUFFERS(nbWorkers) (2*(nbWorkers) + 3) /* After a worker releases its rawSeqStore, it is immediately ready for reuse. * So we only need one seq buffer per worker. */ #define SEQ_POOL_MAX_NB_BUFFERS(nbWorkers) (nbWorkers) /* ===== Seq Pool Wrapper ====== */ typedef ZSTDMT_bufferPool ZSTDMT_seqPool; static size_t ZSTDMT_sizeof_seqPool(ZSTDMT_seqPool* seqPool) { return ZSTDMT_sizeof_bufferPool(seqPool); } static RawSeqStore_t bufferToSeq(Buffer buffer) { ZSTDMT_RustRawSeqStore const rustSeq = ZSTDMT_rust_bufferToSeq((ZSTDMT_RustBuffer){ buffer.start, buffer.capacity }); RawSeqStore_t seq = { (rawSeq*)rustSeq.seq, rustSeq.pos, rustSeq.posInSequence, rustSeq.size, rustSeq.capacity }; return seq; } static Buffer seqToBuffer(RawSeqStore_t seq) { ZSTDMT_RustRawSeqStore const rustSeq = { seq.seq, seq.pos, seq.posInSequence, seq.size, seq.capacity }; ZSTDMT_RustBuffer const rustBuffer = ZSTDMT_rust_seqToBuffer(rustSeq); return (Buffer){ rustBuffer.start, rustBuffer.capacity }; } static RawSeqStore_t ZSTDMT_getSeq(ZSTDMT_seqPool* seqPool) { return bufferToSeq(ZSTDMT_getBuffer(seqPool)); } #if ZSTD_RESIZE_SEQPOOL static RawSeqStore_t ZSTDMT_resizeSeq(ZSTDMT_seqPool* seqPool, RawSeqStore_t seq) { return bufferToSeq(ZSTDMT_resizeBuffer(seqPool, seqToBuffer(seq))); } #endif static void ZSTDMT_releaseSeq(ZSTDMT_seqPool* seqPool, RawSeqStore_t seq) { ZSTDMT_releaseBuffer(seqPool, seqToBuffer(seq)); } static void ZSTDMT_setNbSeq(ZSTDMT_seqPool* const seqPool, size_t const nbSeq) { ZSTDMT_setBufferSize(seqPool, nbSeq * sizeof(rawSeq)); } static ZSTDMT_seqPool* ZSTDMT_createSeqPool(unsigned nbWorkers, ZSTD_customMem cMem) { ZSTDMT_seqPool* const seqPool = ZSTDMT_createBufferPool(SEQ_POOL_MAX_NB_BUFFERS(nbWorkers), cMem); if (seqPool == NULL) return NULL; ZSTDMT_setNbSeq(seqPool, 0); return seqPool; } static void ZSTDMT_freeSeqPool(ZSTDMT_seqPool* seqPool) { ZSTDMT_freeBufferPool(seqPool); } static ZSTDMT_seqPool* ZSTDMT_expandSeqPool(ZSTDMT_seqPool* pool, U32 nbWorkers) { return ZSTDMT_expandBufferPool(pool, SEQ_POOL_MAX_NB_BUFFERS(nbWorkers)); } /* ===== CCtx Pool ===== */ /* a single CCtx Pool can be invoked from multiple threads in parallel */ typedef struct ZSTDMT_RustCCtxPool_s ZSTDMT_RustCCtxPool; ZSTDMT_RustCCtxPool* ZSTDMT_rust_cctx_pool_create(unsigned nbWorkers, ZSTD_customMem cMem); void ZSTDMT_rust_cctx_pool_free(ZSTDMT_RustCCtxPool* pool); size_t ZSTDMT_rust_cctx_pool_sizeof(const ZSTDMT_RustCCtxPool* pool); size_t ZSTDMT_rust_sizeofCCtxPool(size_t wrapperSize, size_t rustPoolSize); ZSTDMT_RustCCtxPool* ZSTDMT_rust_cctx_pool_expand(ZSTDMT_RustCCtxPool* pool, unsigned nbWorkers); ZSTD_CCtx* ZSTDMT_rust_cctx_pool_get(ZSTDMT_RustCCtxPool* pool); void ZSTDMT_rust_cctx_pool_release(ZSTDMT_RustCCtxPool* pool, ZSTD_CCtx* cctx); typedef struct { ZSTDMT_RustCCtxPool* rustPool; int totalCCtx; /* kept for the existing MT parameter diagnostics */ ZSTD_customMem cMem; } ZSTDMT_CCtxPool; /* note : all CCtx borrowed from the pool must be reverted back to the pool _before_ freeing the pool */ static void ZSTDMT_freeCCtxPool(ZSTDMT_CCtxPool* pool) { if (!pool) return; ZSTDMT_rust_cctx_pool_free(pool->rustPool); ZSTD_customFree(pool, pool->cMem); } /* ZSTDMT_createCCtxPool() : * implies nbWorkers >= 1 , checked by caller ZSTDMT_createCCtx() */ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(int nbWorkers, ZSTD_customMem cMem) { ZSTDMT_CCtxPool* const cctxPool = (ZSTDMT_CCtxPool*) ZSTD_customCalloc(sizeof(ZSTDMT_CCtxPool), cMem); assert(nbWorkers > 0); if (!cctxPool) return NULL; cctxPool->rustPool = ZSTDMT_rust_cctx_pool_create((unsigned)nbWorkers, cMem); if (cctxPool->rustPool == NULL) { ZSTD_customFree(cctxPool, cMem); return NULL; } cctxPool->totalCCtx = nbWorkers; cctxPool->cMem = cMem; DEBUGLOG(3, "cctxPool created, with %u workers", nbWorkers); return cctxPool; } static ZSTDMT_CCtxPool* ZSTDMT_expandCCtxPool(ZSTDMT_CCtxPool* srcPool, int nbWorkers) { if (srcPool==NULL) return NULL; srcPool->rustPool = ZSTDMT_rust_cctx_pool_expand(srcPool->rustPool, (unsigned)nbWorkers); if (srcPool->rustPool == NULL) { ZSTD_customMem const cMem = srcPool->cMem; ZSTD_customFree(srcPool, cMem); return NULL; } srcPool->totalCCtx = nbWorkers; return srcPool; } /* only works during initialization phase, not during compression */ static size_t ZSTDMT_sizeof_CCtxPool(ZSTDMT_CCtxPool* cctxPool) { size_t rustPoolSize; if (cctxPool == NULL) return 0; rustPoolSize = ZSTDMT_rust_cctx_pool_sizeof(cctxPool->rustPool); return ZSTDMT_rust_sizeofCCtxPool(sizeof(*cctxPool), rustPoolSize); } static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool) { DEBUGLOG(5, "ZSTDMT_getCCtx"); return ZSTDMT_rust_cctx_pool_get(cctxPool->rustPool); } static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx) { ZSTDMT_rust_cctx_pool_release(pool->rustPool, cctx); } /* ==== Serial State ==== */ typedef struct { void const* start; size_t size; } Range; typedef struct { /* All variables in the struct are protected by mutex. */ ZSTD_pthread_mutex_t mutex; ZSTD_pthread_cond_t cond; ZSTD_CCtx_params params; ldmState_t ldmState; XXH64_state_t xxhState; unsigned nextJobID; /* Protects ldmWindow. * Must be acquired after the main mutex when acquiring both. */ ZSTD_pthread_mutex_t ldmWindowMutex; ZSTD_pthread_cond_t ldmWindowCond; /* Signaled when ldmWindow is updated */ ZSTD_window_t ldmWindow; /* A thread-safe copy of ldmState.window */ } SerialState; typedef struct { SerialState* serialState; ZSTDMT_seqPool* seqPool; ZSTD_CCtx_params* params; const void* dict; size_t dictSize; ZSTD_dictContentType_e dictContentType; ZSTD_customMem cMem; } ZSTDMT_serialResetContext; static void ZSTDMT_serialResetSetLdmParams( void* opaque, int enableLdm, unsigned hashLog, unsigned bucketSizeLog, unsigned minMatchLength, unsigned hashRateLog, unsigned windowLog) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; ldmParams_t* const ldmParams = &context->params->ldmParams; ldmParams->enableLdm = (ZSTD_ParamSwitch_e)enableLdm; ldmParams->hashLog = hashLog; ldmParams->bucketSizeLog = bucketSizeLog; ldmParams->minMatchLength = minMatchLength; ldmParams->hashRateLog = hashRateLog; ldmParams->windowLog = windowLog; } static void ZSTDMT_serialResetNextJob(void* opaque) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; context->serialState->nextJobID = 0; } static void ZSTDMT_serialResetChecksum(void* opaque) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; XXH64_reset(&context->serialState->xxhState, 0); } static void ZSTDMT_serialResetSetNbSeq(void* opaque, size_t nbSeq) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; ZSTDMT_setNbSeq(context->seqPool, nbSeq); } static void ZSTDMT_serialResetWindow(void* opaque) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; ZSTD_window_init(&context->serialState->ldmState.window); } static int ZSTDMT_serialResetResizeTables( void* opaque, size_t hashSize, size_t numBuckets, unsigned bucketLog, unsigned previousBucketLog) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; SerialState* const serialState = context->serialState; ldmParams_t const* const ldmParams = &context->params->ldmParams; if (serialState->ldmState.hashTable == NULL || serialState->params.ldmParams.hashLog < ldmParams->hashLog) { ZSTD_customFree(serialState->ldmState.hashTable, context->cMem); serialState->ldmState.hashTable = (ldmEntry_t*)ZSTD_customMalloc( hashSize, context->cMem); } if (serialState->ldmState.bucketOffsets == NULL || previousBucketLog < bucketLog) { ZSTD_customFree(serialState->ldmState.bucketOffsets, context->cMem); serialState->ldmState.bucketOffsets = (BYTE*)ZSTD_customMalloc( numBuckets, context->cMem); } return !serialState->ldmState.hashTable || !serialState->ldmState.bucketOffsets; } static void ZSTDMT_serialResetZeroTables( void* opaque, size_t hashSize, size_t numBuckets) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; SerialState* const serialState = context->serialState; ZSTD_memset(serialState->ldmState.hashTable, 0, hashSize); ZSTD_memset(serialState->ldmState.bucketOffsets, 0, numBuckets); } static size_t ZSTDMT_serialResetLoadRawDictionary( void* opaque, const void* dict, size_t dictSize) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; SerialState* const serialState = context->serialState; BYTE const* const dictEnd = (const BYTE*)dict + dictSize; ZSTD_window_update(&serialState->ldmState.window, dict, dictSize, /* forceNonContiguous */ 0); ZSTD_ldm_fillHashTable(&serialState->ldmState, (const BYTE*)dict, dictEnd, &context->params->ldmParams); return (size_t)(dictEnd - serialState->ldmState.window.base); } static void ZSTDMT_serialResetSetLoadedDictEnd( void* opaque, U32 loadedDictEnd) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; context->serialState->ldmState.loadedDictEnd = loadedDictEnd; } static void ZSTDMT_serialResetLoadDictionary(void* opaque) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; ZSTDMT_RustSerialDictionaryProjection const projection = { context->dict, context->dictSize, (int)context->dictContentType, context->params->forceWindow }; ZSTDMT_rust_serialStateLoadDictionary( &projection, context, ZSTDMT_serialResetLoadRawDictionary, ZSTDMT_serialResetSetLoadedDictEnd); } static void ZSTDMT_serialResetCopyWindow(void* opaque) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; context->serialState->ldmWindow = context->serialState->ldmState.window; } static void ZSTDMT_serialResetPublishParams(void* opaque, size_t jobSize) { ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque; context->serialState->params = *context->params; context->serialState->params.jobSize = (U32)jobSize; } static int ZSTDMT_serialState_reset(SerialState* serialState, ZSTDMT_seqPool* seqPool, ZSTD_CCtx_params params, size_t jobSize, const void* dict, size_t const dictSize, ZSTD_dictContentType_e dictContentType) { ZSTDMT_serialResetContext context; ZSTDMT_RustSerialResetProjection const projection = { params.ldmParams.enableLdm, params.fParams.checksumFlag, jobSize, params.cParams.windowLog, params.cParams.strategy, params.ldmParams.hashLog, params.ldmParams.bucketSizeLog, params.ldmParams.minMatchLength, params.ldmParams.hashRateLog, serialState->params.ldmParams.hashLog, serialState->params.ldmParams.bucketSizeLog, sizeof(ldmEntry_t) }; /* Rust owns LDM normalization; retain the original diagnostic only. */ if (params.ldmParams.enableLdm == ZSTD_ps_enable) { DEBUGLOG(4, "LDM window size = %u KB", (1U << params.cParams.windowLog) >> 10); } context.serialState = serialState; context.seqPool = seqPool; context.params = ¶ms; context.dict = dict; context.dictSize = dictSize; context.dictContentType = dictContentType; context.cMem = params.customMem; if (ZSTDMT_rust_serialStateReset( &projection, &context, ZSTDMT_serialResetSetLdmParams, ZSTDMT_serialResetNextJob, ZSTDMT_serialResetChecksum, ZSTDMT_serialResetSetNbSeq, ZSTDMT_serialResetWindow, ZSTDMT_serialResetResizeTables, ZSTDMT_serialResetZeroTables, ZSTDMT_serialResetLoadDictionary, ZSTDMT_serialResetCopyWindow, ZSTDMT_serialResetPublishParams)) { return 1; } return 0; } static void ZSTDMT_serialState_zero(void* opaque) { ZSTD_memset(opaque, 0, sizeof(SerialState)); } static int ZSTDMT_serialState_initMutex(void* opaque) { SerialState* const serialState = (SerialState*)opaque; return ZSTD_pthread_mutex_init(&serialState->mutex, NULL); } static int ZSTDMT_serialState_initCond(void* opaque) { SerialState* const serialState = (SerialState*)opaque; return ZSTD_pthread_cond_init(&serialState->cond, NULL); } static int ZSTDMT_serialState_initLdmMutex(void* opaque) { SerialState* const serialState = (SerialState*)opaque; return ZSTD_pthread_mutex_init(&serialState->ldmWindowMutex, NULL); } static int ZSTDMT_serialState_initLdmCond(void* opaque) { SerialState* const serialState = (SerialState*)opaque; return ZSTD_pthread_cond_init(&serialState->ldmWindowCond, NULL); } static void ZSTDMT_serialState_destroyMutex(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_mutex_destroy(&serialState->mutex); } static void ZSTDMT_serialState_destroyCond(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_cond_destroy(&serialState->cond); } static void ZSTDMT_serialState_destroyLdmMutex(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_mutex_destroy(&serialState->ldmWindowMutex); } static void ZSTDMT_serialState_destroyLdmCond(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_cond_destroy(&serialState->ldmWindowCond); } static void ZSTDMT_serialState_freeTables(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_customMem const cMem = serialState->params.customMem; ZSTD_customFree(serialState->ldmState.hashTable, cMem); ZSTD_customFree(serialState->ldmState.bucketOffsets, cMem); } static int ZSTDMT_serialState_init(SerialState* serialState) { return ZSTDMT_rust_serialStateInit( serialState, ZSTDMT_serialState_zero, ZSTDMT_serialState_initMutex, ZSTDMT_serialState_initCond, ZSTDMT_serialState_initLdmMutex, ZSTDMT_serialState_initLdmCond); } static void ZSTDMT_serialState_free(SerialState* serialState) { ZSTDMT_rust_serialStateFree( serialState, ZSTDMT_serialState_destroyMutex, ZSTDMT_serialState_destroyCond, ZSTDMT_serialState_destroyLdmMutex, ZSTDMT_serialState_destroyLdmCond, ZSTDMT_serialState_freeTables); } static void ZSTDMT_serialState_lock(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_PTHREAD_MUTEX_LOCK(&serialState->mutex); } static void ZSTDMT_serialState_wait(void* opaque) { SerialState* const serialState = (SerialState*)opaque; DEBUGLOG(5, "wait for serialState->cond"); ZSTD_pthread_cond_wait(&serialState->cond, &serialState->mutex); } static void ZSTDMT_serialState_broadcast(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_cond_broadcast(&serialState->cond); } static void ZSTDMT_serialState_unlock(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_mutex_unlock(&serialState->mutex); } /* Rust owns the serial turn/skip decision and operation ordering. The wait * callback intentionally leaves the main serial mutex locked; the advance * callback releases it after Rust has performed the current turn's work. */ static int ZSTDMT_serialState_waitForTurn(void* opaque, unsigned jobID) { SerialState* const serialState = (SerialState*)opaque; ZSTDMT_RustSerialWaitForTurnState state; state.callbackContext = serialState; state.nextJobID = &serialState->nextJobID; state.lock = ZSTDMT_serialState_lock; state.wait = ZSTDMT_serialState_wait; state.jobID = jobID; return ZSTDMT_rust_serialStateWaitForTurn(&state); } static void ZSTDMT_serialState_updateLdmWindow( void* opaque, ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize) { SerialState* const serialState = (SerialState*)opaque; RawSeqStore_t* const cSeqStore = (RawSeqStore_t*)seqStore; DEBUGLOG(6, "ZSTDMT_serialState_genSequences: LDM update"); assert(cSeqStore->seq != NULL && cSeqStore->pos == 0 && cSeqStore->size == 0 && cSeqStore->capacity > 0); assert(srcSize <= serialState->params.jobSize); ZSTD_window_update(&serialState->ldmState.window, src, srcSize, /* forceNonContiguous */ 0); } static void ZSTDMT_serialState_generateLdmSequences( void* opaque, ZSTDMT_RustRawSeqStore* seqStore, const void* src, size_t srcSize) { SerialState* const serialState = (SerialState*)opaque; RawSeqStore_t* const cSeqStore = (RawSeqStore_t*)seqStore; size_t error; error = ZSTD_ldm_generateSequences( &serialState->ldmState, cSeqStore, &serialState->params.ldmParams, src, srcSize); /* We provide a large enough buffer to never fail. */ assert(!ZSTD_isError(error)); (void)error; } static void ZSTDMT_serialState_publishLdmWindow(void* opaque) { SerialState* const serialState = (SerialState*)opaque; /* Update ldmWindow to match the ldmState.window and signal the main * thread if it is waiting for a buffer. */ ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex); serialState->ldmWindow = serialState->ldmState.window; ZSTD_pthread_cond_signal(&serialState->ldmWindowCond); ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex); } static void ZSTDMT_serialState_advance(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTDMT_RustSerialAdvanceState state; state.callbackContext = serialState; state.nextJobID = &serialState->nextJobID; state.broadcast = ZSTDMT_serialState_broadcast; state.unlock = ZSTDMT_serialState_unlock; ZSTDMT_rust_serialStateAdvance(&state); } static void ZSTDMT_serialState_ldmLock(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_PTHREAD_MUTEX_LOCK(&serialState->ldmWindowMutex); } static void ZSTDMT_serialState_clearLdmWindow(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_rust_windowClear((size_t)(serialState->ldmWindow.nextSrc - serialState->ldmWindow.base), &serialState->ldmWindow.lowLimit, &serialState->ldmWindow.dictLimit); } static void ZSTDMT_serialState_ldmSignal(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_cond_signal(&serialState->ldmWindowCond); } static void ZSTDMT_serialState_ldmUnlock(void* opaque) { SerialState* const serialState = (SerialState*)opaque; ZSTD_pthread_mutex_unlock(&serialState->ldmWindowMutex); } static void ZSTDMT_serialState_onSkip(void* opaque, unsigned jobID, size_t cSize) { (void)opaque; assert(ZSTD_isError(cSize)); (void)cSize; DEBUGLOG(5, "Skipping past job %u because of error", jobID); } /* Rust owns the non-empty/LDM gate and the point at which this callback is * invoked. Keep only the private codec operation on the C side. */ static void ZSTDMT_compressionJobReferenceSequences( void* opaque, void* cctx, void* sequences, size_t nbSequences) { (void)opaque; assert(cctx != NULL); DEBUGLOG(5, "ZSTDMT_compressionJob: uploading %u external sequences", (unsigned)nbSequences); ZSTD_referenceExternalSequences((ZSTD_CCtx*)cctx, (rawSeq*)sequences, nbSequences); } static void ZSTDMT_serialState_ensureFinished(SerialState* serialState, unsigned jobID, size_t cSize) { ZSTDMT_RustSerialEnsureFinishedState state; state.callbackContext = serialState; state.nextJobID = &serialState->nextJobID; state.lock = ZSTDMT_serialState_lock; state.broadcast = ZSTDMT_serialState_broadcast; state.ldmLock = ZSTDMT_serialState_ldmLock; state.clearLdmWindow = ZSTDMT_serialState_clearLdmWindow; state.ldmSignal = ZSTDMT_serialState_ldmSignal; state.ldmUnlock = ZSTDMT_serialState_ldmUnlock; state.unlock = ZSTDMT_serialState_unlock; state.onSkip = ZSTDMT_serialState_onSkip; state.cSize = cSize; state.jobID = jobID; ZSTDMT_rust_serialStateEnsureFinishedOrchestrated(&state); } /* ------------------------------------------ */ /* ===== Worker thread ===== */ /* ------------------------------------------ */ static const Range kNullRange = { NULL, 0 }; typedef struct { size_t consumed; /* SHARED - set0 by mtctx, then modified by worker AND read by mtctx */ size_t cSize; /* SHARED - set0 by mtctx, then modified by worker AND read by mtctx, then set0 by mtctx */ ZSTD_pthread_mutex_t job_mutex; /* Thread-safe - used by mtctx and worker */ ZSTD_pthread_cond_t job_cond; /* Thread-safe - used by mtctx and worker */ ZSTDMT_CCtxPool* cctxPool; /* Thread-safe - used by mtctx and (all) workers */ ZSTDMT_bufferPool* bufPool; /* Thread-safe - used by mtctx and (all) workers */ ZSTDMT_seqPool* seqPool; /* Thread-safe - used by mtctx and (all) workers */ SerialState* serial; /* Thread-safe - used by mtctx and (all) workers */ Buffer dstBuff; /* set by worker (or mtctx), then read by worker & mtctx, then modified by mtctx => no barrier */ Range prefix; /* set by mtctx, then read by worker & mtctx => no barrier */ Range src; /* set by mtctx, then read by worker & mtctx => no barrier */ unsigned jobID; /* set by mtctx, then read by worker => no barrier */ unsigned firstJob; /* set by mtctx, then read by worker => no barrier */ unsigned lastJob; /* set by mtctx, then read by worker => no barrier */ ZSTD_CCtx_params params; /* set by mtctx, then read by worker => no barrier */ const ZSTD_CDict* cdict; /* set by mtctx, then read by worker => no barrier */ unsigned long long fullFrameSize; /* set by mtctx, then read by worker => no barrier */ size_t dstFlushed; /* used only by mtctx */ unsigned frameChecksumNeeded; /* used only by mtctx */ } ZSTDMT_jobDescription; static void ZSTDMT_compressionJobProgressLock(void* opaque) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque; ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); } static void ZSTDMT_compressionJobProgressSignal(void* opaque) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque; ZSTD_pthread_cond_signal(&job->job_cond); /* warns some more data is ready to be flushed */ } static void ZSTDMT_compressionJobProgressUnlock(void* opaque) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque; ZSTD_pthread_mutex_unlock(&job->job_mutex); } static void ZSTDMT_compressionJobProgressDebug( void* opaque, size_t cSize, size_t totalCSize) { (void)opaque; DEBUGLOG(5, "ZSTDMT_compressionJob: compress new block : cSize==%u bytes (total: %u)", (U32)cSize, (U32)totalCSize); } typedef struct { ZSTDMT_jobDescription* job; ZSTD_CCtx_params jobParams; ZSTD_CCtx* cctx; RawSeqStore_t rawSeqStore; Buffer dstBuff; ZSTDMT_RustCompressionJobFrameHeaderState frameHeaderState; } ZSTDMT_compressionJobState; static void ZSTDMT_compressionJobAcquireCCtx(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; state->cctx = ZSTDMT_getCCtx(state->job->cctxPool); } static void ZSTDMT_compressionJobAcquireRawSeqStore(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; state->rawSeqStore = ZSTDMT_getSeq(state->job->seqPool); } static int ZSTDMT_compressionJobHasCCtx(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; return state->cctx != NULL; } static size_t ZSTDMT_compressionJobAcquireDestinationBuffer(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; state->dstBuff = ZSTDMT_getBuffer(job->bufPool); if (state->dstBuff.start == NULL) return ERROR(memory_allocation); job->dstBuff = state->dstBuff; return 0; } static int ZSTDMT_compressionJobHasRawSeqStore(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; return state->rawSeqStore.seq != NULL; } static void ZSTDMT_compressionJobPublishDestination(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; state->frameHeaderState.dst = state->dstBuff.start; state->frameHeaderState.dstCapacity = state->dstBuff.capacity; } static size_t ZSTDMT_compressionJobAcquireResources(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_RustCompressionJobResourceProjection const projection = { state->jobParams.ldmParams.enableLdm == ZSTD_ps_enable, job->dstBuff.start != NULL }; state->dstBuff = job->dstBuff; DEBUGLOG(5, "ZSTDMT_compressionJob: job %u", job->jobID); return ZSTDMT_rust_compressionJobAcquireResources( &projection, state, ZSTDMT_compressionJobAcquireCCtx, ZSTDMT_compressionJobAcquireRawSeqStore, ZSTDMT_compressionJobHasCCtx, ZSTDMT_compressionJobAcquireDestinationBuffer, ZSTDMT_compressionJobHasRawSeqStore, ZSTDMT_compressionJobPublishDestination); } static void ZSTDMT_compressionJobPrepareParameters(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_RustCompressionJobParameters const parameters = ZSTDMT_rust_prepareCompressionJobParameters( job->jobID, state->jobParams.fParams.checksumFlag, state->jobParams.ldmParams.enableLdm, (unsigned)state->jobParams.nbWorkers); state->jobParams.fParams.checksumFlag = parameters.checksumFlag; state->jobParams.ldmParams.enableLdm = parameters.ldmEnable; state->jobParams.nbWorkers = (int)parameters.nbWorkers; } static void ZSTDMT_compressionJobGenerateSequences(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; /* Perform serial step as early as possible. */ ZSTDMT_rust_serialStateGenSequences( (ZSTDMT_RustRawSeqStore*)&state->rawSeqStore, job->src.start, job->src.size, job->jobID, job->serial->params.ldmParams.enableLdm == ZSTD_ps_enable, job->serial->params.fParams.checksumFlag, job->serial, &job->serial->xxhState, ZSTDMT_serialState_waitForTurn, ZSTDMT_serialState_updateLdmWindow, ZSTDMT_serialState_generateLdmSequences, ZSTDMT_serialState_publishLdmWindow, ZSTDMT_serialState_advance); } static size_t ZSTDMT_compressionJobBeginWithCDict(void* opaque); static size_t ZSTDMT_compressionJobSetForceMaxWindow(void* opaque, int value); static size_t ZSTDMT_compressionJobSetDeterministicRefPrefix(void* opaque); static size_t ZSTDMT_compressionJobBeginWithPrefix(void* opaque); static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque); static size_t ZSTDMT_compressionJobBegin(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_RustCompressionJobBeginProjection const projection = { job->firstJob, (unsigned)(job->cdict != NULL) }; return ZSTDMT_rust_compressionJobBegin( &projection, state, ZSTDMT_compressionJobBeginWithCDict, ZSTDMT_compressionJobSetForceMaxWindow, ZSTDMT_compressionJobSetDeterministicRefPrefix, ZSTDMT_compressionJobBeginWithPrefix, ZSTDMT_compressionJobPublishFrameHeader); } static size_t ZSTDMT_compressionJobBeginWithCDict(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; return ZSTD_compressBegin_advanced_internal( state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict, &state->jobParams, job->fullFrameSize); } static size_t ZSTDMT_compressionJobSetForceMaxWindow(void* opaque, int value) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; return ZSTD_CCtxParams_setParameter( &state->jobParams, ZSTD_c_forceMaxWindow, value); } static size_t ZSTDMT_compressionJobSetDeterministicRefPrefix(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; return ZSTD_CCtxParams_setParameter( &state->jobParams, ZSTD_c_deterministicRefPrefix, 0); } static size_t ZSTDMT_compressionJobBeginWithPrefix(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size; DEBUGLOG(6, "ZSTDMT_compressionJob: job %u: loading prefix of size %zu", job->jobID, job->prefix.size); return ZSTD_compressBegin_advanced_internal( state->cctx, job->prefix.start, job->prefix.size, ZSTD_dct_rawContent, ZSTD_dtlm_fast, NULL, /*cdict*/ &state->jobParams, pledgedSrcSize); } static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; state->frameHeaderState.stage = (int*)&state->cctx->stage; state->frameHeaderState.noDictIDFlag = state->cctx->appliedParams.fParams.noDictIDFlag; state->frameHeaderState.checksumFlag = state->cctx->appliedParams.fParams.checksumFlag; state->frameHeaderState.contentSizeFlag = state->cctx->appliedParams.fParams.contentSizeFlag; state->frameHeaderState.format = (int)state->cctx->appliedParams.format; state->frameHeaderState.windowLog = state->cctx->appliedParams.cParams.windowLog; state->frameHeaderState.pledgedSrcSizePlusOne = state->cctx->pledgedSrcSizePlusOne; state->frameHeaderState.dictID = state->cctx->dictID; state->frameHeaderState.repCodes = state->cctx->blockState.prevCBlock == NULL ? NULL : state->cctx->blockState.prevCBlock->rep; } static ZSTDMT_chunkProcessResult ZSTDMT_compressionJobCompress( void* opaque, unsigned lastJob) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_RustCompressionJobProgressProjection progress; size_t const chunkSize = 4*ZSTD_BLOCKSIZE_MAX; assert(lastJob == job->lastJob); if (sizeof(size_t) > sizeof(int)) assert(job->src.size < ((size_t)INT_MAX) * chunkSize); /* check overflow */ DEBUGLOG(5, "ZSTDMT_compressionJob: compress %u bytes in %zu blocks", (U32)job->src.size, (job->src.size + (chunkSize-1)) / chunkSize); assert(job->cSize == 0); assert(chunkSize > 0); assert((chunkSize & (chunkSize - 1)) == 0); /* chunkSize must be power of 2 for mask==(chunkSize-1) to work */ progress.callbackContext = job; progress.cSize = &job->cSize; progress.consumed = &job->consumed; progress.lock = ZSTDMT_compressionJobProgressLock; progress.signal = ZSTDMT_compressionJobProgressSignal; progress.unlock = ZSTDMT_compressionJobProgressUnlock; progress.debug = ZSTDMT_compressionJobProgressDebug; return ZSTDMT_rust_compressJobChunks( state->cctx, job->src.start, job->src.size, state->dstBuff.start, state->dstBuff.capacity, chunkSize, lastJob, &progress, ZSTDMT_rust_compressionJobProgress); } static void ZSTDMT_compressionJobTraceAssertNoExtDict(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; /* Double check that we don't have an ext-dict, because then our repcode * invalidation doesn't work. */ assert(!ZSTD_window_hasExtDict(state->cctx->blockState.matchState.window)); } static void ZSTDMT_compressionJobTraceRecord(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTD_CCtx_trace(state->cctx, 0); } static void ZSTDMT_compressionJobTrace(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_RustCompressionJobTraceProjection const projection = { state, state->job->firstJob, ZSTDMT_compressionJobTraceAssertNoExtDict, ZSTDMT_compressionJobTraceRecord }; ZSTDMT_rust_compressionJobTrace(&projection); } static void ZSTDMT_compressionJobEnsureFinished(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_serialState_ensureFinished(job->serial, job->jobID, job->cSize); if (job->prefix.size > 0) DEBUGLOG(5, "Finished with prefix: %zx", (size_t)job->prefix.start); DEBUGLOG(5, "Finished with source: %zx", (size_t)job->src.start); } static void ZSTDMT_compressionJobReleaseSeq(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_releaseSeq(job->seqPool, state->rawSeqStore); } static void ZSTDMT_compressionJobReleaseCCtx(void* opaque) { ZSTDMT_compressionJobState* const state = (ZSTDMT_compressionJobState*)opaque; ZSTDMT_jobDescription* const job = state->job; ZSTDMT_releaseCCtx(job->cctxPool, state->cctx); } static void ZSTDMT_compressionJobValidateConsumed(void* opaque, size_t srcSize) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque; assert(srcSize == job->src.size); } /* ZSTDMT_compressionJob() is a POOL_function type. Rust owns the stage * ordering; these callbacks retain the private job and codec operations. */ static void ZSTDMT_compressionJob(void* jobDescription) { ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription; ZSTDMT_compressionJobState state; ZSTDMT_RustCompressionJobSequenceState sequenceState; ZSTDMT_RustCompressionJobFinishPublicationProjection publication; ZSTDMT_RustCompressionJobFinishProjection finishProjection; ZSTDMT_RustCompressionJobProjection projection; ZSTD_memset(&state, 0, sizeof(state)); state.job = job; state.jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */ state.dstBuff = job->dstBuff; sequenceState = (ZSTDMT_RustCompressionJobSequenceState){ &state.cctx, (ZSTDMT_RustRawSeqStore*)&state.rawSeqStore, job->serial->params.ldmParams.enableLdm }; projection = (ZSTDMT_RustCompressionJobProjection){ job->firstJob, job->lastJob, &state.frameHeaderState, &sequenceState }; publication = (ZSTDMT_RustCompressionJobFinishPublicationProjection){ job, &job->cSize, &job->consumed, job->src.size, ZSTDMT_compressionJobProgressLock, ZSTDMT_compressionJobProgressSignal, ZSTDMT_compressionJobProgressUnlock, ZSTDMT_compressionJobValidateConsumed }; finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){ &state, job->src.size, ZSTDMT_compressionJobEnsureFinished, publication, ZSTDMT_compressionJobReleaseSeq, ZSTDMT_compressionJobReleaseCCtx }; ZSTDMT_rust_compressionJob( &projection, &finishProjection, &state, ZSTDMT_compressionJobAcquireResources, ZSTDMT_compressionJobPrepareParameters, ZSTDMT_compressionJobGenerateSequences, ZSTDMT_compressionJobBegin, ZSTDMT_compressionJobReferenceSequences, ZSTDMT_compressionJobCompress, ZSTDMT_compressionJobTrace); } /* ------------------------------------------ */ /* ===== Multi-threaded compression ===== */ /* ------------------------------------------ */ typedef struct { Range prefix; /* read-only non-owned prefix buffer */ Buffer buffer; size_t filled; } InBuff_t; typedef struct { BYTE* buffer; /* The round input buffer. All jobs get references * to pieces of the buffer. ZSTDMT_tryGetInputRange() * handles handing out job input buffers, and makes * sure it doesn't overlap with any pieces still in use. */ size_t capacity; /* The capacity of buffer. */ size_t pos; /* The position of the current inBuff in the round * buffer. Updated past the end if the inBuff once * the inBuff is sent to the worker thread. * pos <= capacity. */ } RoundBuff_t; #define RSYNC_LENGTH 32 /* Don't create chunks smaller than the zstd block size. * This stops us from regressing compression ratio too much, * and ensures our output fits in ZSTD_compressBound(). * * If this is shrunk < ZSTD_BLOCKSIZELOG_MIN then * ZSTD_COMPRESSBOUND() will need to be updated. */ #define RSYNC_MIN_BLOCK_LOG ZSTD_BLOCKSIZELOG_MAX #define RSYNC_MIN_BLOCK_SIZE (1< one job is already prepared, but pool has shortage of workers. Don't create a new job. */ InBuff_t inBuff; RoundBuff_t roundBuff; SerialState serial; RSyncState_t rsync; unsigned jobIDMask; unsigned doneJobID; unsigned nextJobID; unsigned frameEnded; unsigned allJobsCompleted; unsigned long long frameContentSize; unsigned long long consumed; unsigned long long produced; ZSTD_customMem cMem; ZSTD_CDict* cdictLocal; const ZSTD_CDict* cdict; unsigned providedFactory: 1; }; /* Project only the scalar job state needed by Rust's read-only MT * orchestration. The descriptor layout and its mutex remain private here. */ static void ZSTDMT_projectJob(void* opaque, unsigned jobID, ZSTDMT_RustJobProjection* projection) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; ZSTD_pthread_mutex_lock(&job->job_mutex); projection->consumed = job->consumed; projection->cSize = job->cSize; projection->srcStart = job->src.start; projection->srcSize = job->src.size; projection->prefixStart = job->prefix.start; projection->prefixSize = job->prefix.size; projection->dstFlushed = job->dstFlushed; ZSTD_pthread_mutex_unlock(&job->job_mutex); } static void ZSTDMT_freeJobsTable(ZSTDMT_jobDescription* jobTable, U32 nbJobs, ZSTD_customMem cMem) { ZSTDMT_RustFreeJobsTableProjection const projection = { jobTable, nbJobs, sizeof(*jobTable), cMem, ZSTDMT_job_table_destroy_sync }; if (jobTable == NULL) return; ZSTDMT_rust_freeJobsTable(&projection); } int ZSTDMT_job_table_init_sync(void* jobTable, unsigned nbJobs, size_t jobSize) { U32 jobNb; int initError = 0; BYTE* const table = (BYTE*)jobTable; assert(jobSize == sizeof(ZSTDMT_jobDescription)); if (jobTable == NULL) return 1; for (jobNb=0; jobNbjob_mutex, NULL); initError |= ZSTD_pthread_cond_init(&job->job_cond, NULL); } return initError; } void ZSTDMT_job_table_destroy_sync(void* jobTable, unsigned nbJobs, size_t jobSize) { U32 jobNb; BYTE* const table = (BYTE*)jobTable; if (jobTable == NULL) return; assert(jobSize == sizeof(ZSTDMT_jobDescription)); for (jobNb=0; jobNbjob_mutex); ZSTD_pthread_cond_destroy(&job->job_cond); } } /* ZSTDMT_allocJobsTable() * allocate and init a job table. * update *nbJobsPtr to next power of 2 value, as size of table */ static ZSTDMT_jobDescription* ZSTDMT_createJobsTable(U32* nbJobsPtr, ZSTD_customMem cMem) { return (ZSTDMT_jobDescription*) ZSTDMT_rust_createJobsTable( nbJobsPtr, sizeof(ZSTDMT_jobDescription), cMem, ZSTDMT_job_table_init_sync, ZSTDMT_job_table_destroy_sync); } static size_t ZSTDMT_expandJobsTable (ZSTDMT_CCtx* mtctx, U32 nbWorkers) { void* jobs = mtctx->jobs; U32 jobIDMask = mtctx->jobIDMask; size_t const error = ZSTDMT_rust_expandJobsTable( &jobs, &jobIDMask, nbWorkers, sizeof(ZSTDMT_jobDescription), mtctx->cMem, ZSTDMT_job_table_init_sync, ZSTDMT_job_table_destroy_sync); mtctx->jobs = (ZSTDMT_jobDescription*)jobs; mtctx->jobIDMask = jobIDMask; return error; } typedef struct { ZSTDMT_CCtx* mtctx; ZSTD_customMem cMem; ZSTD_threadPool* pool; } ZSTDMT_rust_createCCtx_context; static void* ZSTDMT_rust_createCCtx_allocate(void* opaque, size_t size) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; context->mtctx = (ZSTDMT_CCtx*)ZSTD_customCalloc(size, context->cMem); return context->mtctx; } static size_t ZSTDMT_rust_createCCtx_setWorkers(void* opaque, unsigned nbWorkers) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; return ZSTD_CCtxParams_setParameter( &context->mtctx->params, ZSTD_c_nbWorkers, (int)nbWorkers); } static void ZSTDMT_rust_createCCtx_setInitialState(void* opaque) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; context->mtctx->cMem = context->cMem; context->mtctx->allJobsCompleted = 1; } static void* ZSTDMT_rust_createCCtx_factory(void* opaque, unsigned nbWorkers) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; if (context->pool != NULL) { context->mtctx->factory = context->pool; context->mtctx->providedFactory = 1; } else { context->mtctx->factory = POOL_create_advanced(nbWorkers, 0, context->cMem); context->mtctx->providedFactory = 0; } return context->mtctx->factory; } static void* ZSTDMT_rust_createCCtx_jobs(void* opaque, unsigned* nbJobs) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; context->mtctx->jobs = ZSTDMT_createJobsTable((U32*)nbJobs, context->cMem); if (context->mtctx->jobs != NULL) context->mtctx->jobIDMask = *nbJobs - 1; return context->mtctx->jobs; } static void* ZSTDMT_rust_createCCtx_bufferPool(void* opaque, unsigned nbWorkers) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; context->mtctx->bufPool = ZSTDMT_createBufferPool( BUF_POOL_MAX_NB_BUFFERS(nbWorkers), context->cMem); return context->mtctx->bufPool; } static void* ZSTDMT_rust_createCCtx_cctxPool(void* opaque, unsigned nbWorkers) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; context->mtctx->cctxPool = ZSTDMT_createCCtxPool(nbWorkers, context->cMem); return context->mtctx->cctxPool; } static void* ZSTDMT_rust_createCCtx_seqPool(void* opaque, unsigned nbWorkers) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; context->mtctx->seqPool = ZSTDMT_createSeqPool(nbWorkers, context->cMem); return context->mtctx->seqPool; } static int ZSTDMT_rust_createCCtx_initSerial(void* opaque) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; return ZSTDMT_serialState_init(&context->mtctx->serial); } static void ZSTDMT_rust_createCCtx_free(void* opaque) { ZSTDMT_rust_createCCtx_context* const context = (ZSTDMT_rust_createCCtx_context*)opaque; if (context->mtctx != NULL) { ZSTDMT_freeCCtx(context->mtctx); context->mtctx = NULL; } } MEM_STATIC ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced_internal(unsigned nbWorkers, ZSTD_customMem cMem, ZSTD_threadPool* pool) { ZSTDMT_rust_createCCtx_context context = { NULL, cMem, pool }; ZSTDMT_RustCreateCCtxProjection const projection = { &context, nbWorkers, ZSTDMT_NBWORKERS_MAX, sizeof(ZSTDMT_CCtx), cMem }; DEBUGLOG(3, "ZSTDMT_createCCtx_advanced (nbWorkers = %u)", nbWorkers); return (ZSTDMT_CCtx*)ZSTDMT_rust_createCCtx( &projection, ZSTDMT_rust_createCCtx_allocate, ZSTDMT_rust_createCCtx_setWorkers, ZSTDMT_rust_createCCtx_setInitialState, ZSTDMT_rust_createCCtx_factory, ZSTDMT_rust_createCCtx_jobs, ZSTDMT_rust_createCCtx_bufferPool, ZSTDMT_rust_createCCtx_cctxPool, ZSTDMT_rust_createCCtx_seqPool, ZSTDMT_rust_createCCtx_initSerial, ZSTDMT_rust_createCCtx_free); } ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbWorkers, ZSTD_customMem cMem, ZSTD_threadPool* pool) { #ifdef ZSTD_MULTITHREAD return ZSTDMT_createCCtx_advanced_internal(nbWorkers, cMem, pool); #else (void)nbWorkers; (void)cMem; (void)pool; return NULL; #endif } static void ZSTDMT_releaseJobResource(void* opaque, unsigned jobID) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; /* Copy the mutex/cond out */ ZSTD_pthread_mutex_t const mutex = mtctx->jobs[jobID].job_mutex; ZSTD_pthread_cond_t const cond = mtctx->jobs[jobID].job_cond; DEBUGLOG(4, "job%02u: release dst address %08X", jobID, (U32)(size_t)mtctx->jobs[jobID].dstBuff.start); ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[jobID].dstBuff); /* Clear the job description, but keep the mutex/cond */ ZSTD_memset(&mtctx->jobs[jobID], 0, sizeof(mtctx->jobs[jobID])); mtctx->jobs[jobID].job_mutex = mutex; mtctx->jobs[jobID].job_cond = cond; } /* ZSTDMT_releaseAllJobResources() : * note : ensure all workers are killed first ! */ static void ZSTDMT_releaseAllJobResources(ZSTDMT_CCtx* mtctx) { DEBUGLOG(3, "ZSTDMT_releaseAllJobResources"); ZSTDMT_rust_releaseAllJobResources( mtctx->jobIDMask, mtctx, ZSTDMT_releaseJobResource); mtctx->inBuff.buffer = g_nullBuffer; mtctx->inBuff.filled = 0; mtctx->allJobsCompleted = 1; } static void ZSTDMT_waitForJobComplete( void* opaque, unsigned jobID, unsigned doneJobID) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; (void)doneJobID; ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); while (job->consumed < job->src.size) { DEBUGLOG(4, "waiting for jobCompleted signal from job %u", doneJobID); ZSTD_pthread_cond_wait(&job->job_cond, &job->job_mutex); } ZSTD_pthread_mutex_unlock(&job->job_mutex); } static void ZSTDMT_waitForAllJobsCompleted(ZSTDMT_CCtx* mtctx) { DEBUGLOG(4, "ZSTDMT_waitForAllJobsCompleted"); mtctx->doneJobID = ZSTDMT_rust_waitForAllJobsCompleted( mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask, mtctx, ZSTDMT_waitForJobComplete); } static void ZSTDMT_rust_freeCCtxFactory(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; POOL_free(mtctx->factory); /* stop and free worker threads */ } static void ZSTDMT_rust_freeCCtxReleaseJobResources(void* opaque) { ZSTDMT_releaseAllJobResources((ZSTDMT_CCtx*)opaque); } static void ZSTDMT_rust_freeCCtxJobs(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_freeJobsTable(mtctx->jobs, mtctx->jobIDMask+1, mtctx->cMem); } static void ZSTDMT_rust_freeCCtxBufferPool(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_freeBufferPool(mtctx->bufPool); } static void ZSTDMT_rust_freeCCtxPool(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_freeCCtxPool(mtctx->cctxPool); } static void ZSTDMT_rust_freeCCtxSeqPool(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_freeSeqPool(mtctx->seqPool); } static void ZSTDMT_rust_freeCCtxSerialState(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_serialState_free(&mtctx->serial); } static void ZSTDMT_rust_freeCCtxCDict(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTD_freeCDict(mtctx->cdictLocal); } static void ZSTDMT_rust_freeCCtxRoundBuffer(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTD_customFree(mtctx->roundBuff.buffer, mtctx->cMem); } static void ZSTDMT_rust_freeCCtxContext(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTD_customFree(mtctx, mtctx->cMem); } size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx) { ZSTDMT_RustFreeCCtxState const state = { mtctx, mtctx != NULL && mtctx->providedFactory, mtctx != NULL && mtctx->roundBuff.buffer != NULL, ZSTDMT_rust_freeCCtxFactory, ZSTDMT_rust_freeCCtxReleaseJobResources, ZSTDMT_rust_freeCCtxJobs, ZSTDMT_rust_freeCCtxBufferPool, ZSTDMT_rust_freeCCtxPool, ZSTDMT_rust_freeCCtxSeqPool, ZSTDMT_rust_freeCCtxSerialState, ZSTDMT_rust_freeCCtxCDict, ZSTDMT_rust_freeCCtxRoundBuffer, ZSTDMT_rust_freeCCtxContext }; return ZSTDMT_rust_freeCCtx(mtctx == NULL ? NULL : &state); } size_t ZSTDMT_sizeof_CCtx(ZSTDMT_CCtx* mtctx) { size_t mtctxSize; size_t factorySize; size_t bufferPoolSize; size_t jobsSize; size_t cctxPoolSize; size_t seqPoolSize; size_t cdictSize; size_t roundBuffSize; if (mtctx == NULL) return 0; /* supports sizeof NULL */ mtctxSize = sizeof(*mtctx); factorySize = POOL_sizeof(mtctx->factory); bufferPoolSize = ZSTDMT_sizeof_bufferPool(mtctx->bufPool); jobsSize = (mtctx->jobIDMask+1) * sizeof(ZSTDMT_jobDescription); cctxPoolSize = ZSTDMT_sizeof_CCtxPool(mtctx->cctxPool); seqPoolSize = ZSTDMT_sizeof_seqPool(mtctx->seqPool); cdictSize = ZSTD_sizeof_CDict(mtctx->cdictLocal); roundBuffSize = mtctx->roundBuff.capacity; return ZSTDMT_rust_sizeofCCtx(mtctxSize, factorySize, bufferPoolSize, jobsSize, cctxPoolSize, seqPoolSize, cdictSize, roundBuffSize); } static size_t ZSTDMT_rust_resizeFactory(void* opaque, unsigned nbWorkers) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; return POOL_resize(mtctx->factory, nbWorkers) ? ERROR(memory_allocation) : 0; } static size_t ZSTDMT_rust_resizeJobs(void* opaque, unsigned nbWorkers) { return ZSTDMT_expandJobsTable((ZSTDMT_CCtx*)opaque, nbWorkers); } static void* ZSTDMT_rust_resizeBufferPool(void* opaque, unsigned nbWorkers) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; mtctx->bufPool = ZSTDMT_expandBufferPool( mtctx->bufPool, BUF_POOL_MAX_NB_BUFFERS(nbWorkers)); return mtctx->bufPool; } static void* ZSTDMT_rust_resizeCCtxPool(void* opaque, unsigned nbWorkers) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; mtctx->cctxPool = ZSTDMT_expandCCtxPool(mtctx->cctxPool, nbWorkers); return mtctx->cctxPool; } static void* ZSTDMT_rust_resizeSeqPool(void* opaque, unsigned nbWorkers) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; mtctx->seqPool = ZSTDMT_expandSeqPool(mtctx->seqPool, nbWorkers); return mtctx->seqPool; } static size_t ZSTDMT_rust_resizeSetNbWorkers(void* opaque, unsigned nbWorkers) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; (void)ZSTD_CCtxParams_setParameter(&mtctx->params, ZSTD_c_nbWorkers, (int)nbWorkers); return 0; } /* ZSTDMT_resize() : * @return : error code if fails, 0 on success */ static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers) { ZSTDMT_RustResizeProjection const projection = { mtctx, nbWorkers }; return ZSTDMT_rust_resize( &projection, ZSTDMT_rust_resizeFactory, ZSTDMT_rust_resizeJobs, ZSTDMT_rust_resizeBufferPool, ZSTDMT_rust_resizeCCtxPool, ZSTDMT_rust_resizeSeqPool, ZSTDMT_rust_resizeSetNbWorkers); } /*! ZSTDMT_updateCParams_whileCompressing() : * Updates a selected set of compression parameters, remaining compatible with currently active frame. * New parameters will be applied to next compression job. */ void ZSTDMT_updateCParams_whileCompressing(ZSTDMT_CCtx* mtctx, const ZSTD_CCtx_params* cctxParams) { U32 const saved_wlog = mtctx->params.cParams.windowLog; /* Do not modify windowLog while compressing */ DEBUGLOG(5, "ZSTDMT_updateCParams_whileCompressing (level:%i)", cctxParams->compressionLevel); { ZSTDMT_RustCParamsUpdateProjection const projection = { cctxParams->compressionLevel, cctxParams->srcSizeHint, ZSTD_CONTENTSIZE_UNKNOWN, 0, ZSTD_cpm_noAttachDict, cctxParams->ldmParams.enableLdm, ZSTD_LDM_DEFAULT_WINDOW_LOG, cctxParams->cParams, cctxParams->useRowMatchFinder, ZSTD_getCParamsExclusionMask(), saved_wlog }; ZSTDMT_RustCParamsUpdateResult const result = ZSTDMT_rust_updateCParamsWhileCompressing(projection); mtctx->params.compressionLevel = result.compressionLevel; mtctx->params.cParams = result.cParams; } } /* ZSTDMT_getFrameProgression(): * tells how much data has been consumed (input) and produced (output) for current frame. * able to count progression inside worker threads. * Note : mutex will be acquired during statistics collection inside workers. */ ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx) { ZSTD_frameProgression const fps = ZSTDMT_rust_frameProgressionWithJobs( mtctx->consumed, mtctx->inBuff.filled, mtctx->produced, mtctx->nextJobID, mtctx->doneJobID, mtctx->nextJobID, mtctx->jobReady, mtctx->jobIDMask, mtctx, ZSTDMT_projectJob); DEBUGLOG(5, "ZSTDMT_getFrameProgression"); return fps; } size_t ZSTDMT_toFlushNow(ZSTDMT_CCtx* mtctx) { assert(mtctx->doneJobID <= mtctx->nextJobID); return ZSTDMT_rust_toFlushNow(mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask, mtctx, ZSTDMT_projectJob); } /* ------------------------------------------ */ /* ===== Multi-threaded compression ===== */ /* ------------------------------------------ */ typedef struct { ZSTDMT_CCtx* mtctx; ZSTD_CCtx_params params; const void* dict; size_t dictSize; ZSTD_dictContentType_e dictContentType; const ZSTD_CDict* cdict; unsigned long long pledgedSrcSize; } ZSTDMT_initCStreamState; static size_t ZSTDMT_initCStreamResize(void* opaque, unsigned nbWorkers) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; return ZSTDMT_resize(state->mtctx, nbWorkers); } static void ZSTDMT_initCStreamWaitForAllJobs(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_waitForAllJobsCompleted(state->mtctx); } static void ZSTDMT_initCStreamReleaseAllJobResources(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_releaseAllJobResources(state->mtctx); } static void ZSTDMT_initCStreamApplyParameters(void* opaque, size_t jobSize) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->params.jobSize = jobSize; state->mtctx->params = state->params; state->mtctx->frameContentSize = state->pledgedSrcSize; } static void ZSTDMT_initCStreamReleaseLocalCDict(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTD_freeCDict(state->mtctx->cdictLocal); } static size_t ZSTDMT_initCStreamCreateCopiedCDict(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_CCtx* const mtctx = state->mtctx; mtctx->cdictLocal = ZSTD_createCDict_advanced( state->dict, state->dictSize, ZSTD_dlm_byCopy, state->dictContentType, state->params.cParams, mtctx->cMem); mtctx->cdict = mtctx->cdictLocal; return mtctx->cdictLocal == NULL ? ERROR(memory_allocation) : 0; } static void ZSTDMT_initCStreamAttachBorrowedCDict(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->cdictLocal = NULL; state->mtctx->cdict = state->cdict; } static size_t ZSTDMT_initCStreamPrepareDictionary(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_RustDictionaryProjection const projection = { state->dict != NULL, state->dictContentType == ZSTD_dct_rawContent }; return ZSTDMT_rust_prepareCStreamDictionary( &projection, state, ZSTDMT_initCStreamReleaseLocalCDict, ZSTDMT_initCStreamCreateCopiedCDict, ZSTDMT_initCStreamAttachBorrowedCDict); } static void ZSTDMT_initCStreamSetTargetPrefixSize(void* opaque, size_t size) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->targetPrefixSize = size; DEBUGLOG(4, "overlapLog=%i => %u KB", state->params.overlapLog, (U32)(size >> 10)); } static void ZSTDMT_initCStreamSetTargetSectionSize(void* opaque, size_t size) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->targetSectionSize = size; } static void ZSTDMT_initCStreamSetRsync(void* opaque, U64 hitMask, U64 primePower) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->rsync.hash = 0; state->mtctx->rsync.hitMask = hitMask; state->mtctx->rsync.primePower = primePower; DEBUGLOG(4, "rsyncLog = %u", ZSTD_highbit32((U32)(hitMask + 1))); } static void ZSTDMT_initCStreamSetBufferSize(void* opaque, size_t size) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; DEBUGLOG(4, "Job Size : %u KB (note : set to %u)", (U32)(state->mtctx->targetSectionSize >> 10), (U32)state->params.jobSize); DEBUGLOG(4, "inBuff Size : %u KB", (U32)(state->mtctx->targetSectionSize >> 10)); ZSTDMT_setBufferSize(state->mtctx->bufPool, size); } static size_t ZSTDMT_initCStreamResizeRoundBuffer(void* opaque, size_t capacity) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_CCtx* const mtctx = state->mtctx; if (mtctx->roundBuff.capacity < capacity) { if (mtctx->roundBuff.buffer) ZSTD_customFree(mtctx->roundBuff.buffer, mtctx->cMem); mtctx->roundBuff.buffer = (BYTE*)ZSTD_customMalloc(capacity, mtctx->cMem); if (mtctx->roundBuff.buffer == NULL) { mtctx->roundBuff.capacity = 0; return ERROR(memory_allocation); } mtctx->roundBuff.capacity = capacity; } return 0; } static void ZSTDMT_initCStreamResetRoundBuffer(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->roundBuff.pos = 0; } static void ZSTDMT_initCStreamResetInputBuffer(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->inBuff.buffer = g_nullBuffer; state->mtctx->inBuff.filled = 0; state->mtctx->inBuff.prefix = kNullRange; } static void ZSTDMT_initCStreamResetJobIDs(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->doneJobID = 0; state->mtctx->nextJobID = 0; } static void ZSTDMT_initCStreamResetFrameFlags(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->frameEnded = 0; state->mtctx->allJobsCompleted = 0; } static void ZSTDMT_initCStreamResetProgress(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->consumed = 0; state->mtctx->produced = 0; } static void ZSTDMT_initCStreamResetStream(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_CCtx* const mtctx = state->mtctx; ZSTDMT_RustResetStreamState const resetState = { state, ZSTDMT_initCStreamResetRoundBuffer, ZSTDMT_initCStreamResetInputBuffer, ZSTDMT_initCStreamResetJobIDs, ZSTDMT_initCStreamResetFrameFlags, ZSTDMT_initCStreamResetProgress }; DEBUGLOG(4, "roundBuff capacity : %u KB", (U32)(mtctx->roundBuff.capacity >> 10)); ZSTDMT_rust_resetStream(&resetState); } static void ZSTDMT_initCStreamClearDictionary(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_CCtx* const mtctx = state->mtctx; ZSTD_freeCDict(mtctx->cdictLocal); mtctx->cdictLocal = NULL; mtctx->cdict = NULL; } static void ZSTDMT_initCStreamSetRawPrefix(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; state->mtctx->inBuff.prefix.start = (const BYTE*)state->dict; state->mtctx->inBuff.prefix.size = state->dictSize; } static size_t ZSTDMT_initCStreamCreateReferencedCDict(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_CCtx* const mtctx = state->mtctx; /* note : a loadPrefix becomes an internal CDict */ mtctx->cdictLocal = ZSTD_createCDict_advanced( state->dict, state->dictSize, ZSTD_dlm_byRef, state->dictContentType, state->params.cParams, mtctx->cMem); mtctx->cdict = mtctx->cdictLocal; return mtctx->cdictLocal == NULL ? ERROR(memory_allocation) : 0; } static size_t ZSTDMT_initCStreamUpdateDictionary(void* opaque) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; ZSTDMT_RustDictionaryProjection const projection = { state->dict != NULL, state->dictContentType == ZSTD_dct_rawContent }; return ZSTDMT_rust_updateCStreamDictionary( &projection, state, ZSTDMT_initCStreamClearDictionary, ZSTDMT_initCStreamAttachBorrowedCDict, ZSTDMT_initCStreamSetRawPrefix, ZSTDMT_initCStreamCreateReferencedCDict); } static size_t ZSTDMT_initCStreamSerialReset(void* opaque, size_t targetSectionSize) { ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque; return ZSTDMT_rust_initSerialResetResult(ZSTDMT_serialState_reset( &state->mtctx->serial, state->mtctx->seqPool, state->params, targetSectionSize, state->dict, state->dictSize, state->dictContentType)); } /* ====================================== */ /* ======= Streaming API ======= */ /* ====================================== */ size_t ZSTDMT_initCStream_internal( ZSTDMT_CCtx* mtctx, const void* dict, size_t dictSize, ZSTD_dictContentType_e dictContentType, const ZSTD_CDict* cdict, ZSTD_CCtx_params params, unsigned long long pledgedSrcSize) { ZSTDMT_initCStreamState state; ZSTDMT_RustInitCStreamProjection projection; DEBUGLOG(4, "ZSTDMT_initCStream_internal (pledgedSrcSize=%u, nbWorkers=%u, cctxPool=%u)", (U32)pledgedSrcSize, params.nbWorkers, mtctx->cctxPool->totalCCtx); /* params supposed partially fully validated at this point */ assert(!ZSTD_isError(ZSTD_checkCParams(params.cParams))); assert(!((dict) && (cdict))); /* either dict or cdict, not both */ state = (ZSTDMT_initCStreamState){ mtctx, params, dict, dictSize, dictContentType, cdict, pledgedSrcSize }; projection = (ZSTDMT_RustInitCStreamProjection){ (unsigned)params.nbWorkers, (unsigned)mtctx->params.nbWorkers, params.jobSize, ZSTDMT_JOBSIZE_MIN, (size_t)ZSTDMT_JOBSIZE_MAX, params.ldmParams.enableLdm, params.cParams.windowLog, params.cParams.chainLog, params.cParams.strategy, params.overlapLog, params.rsyncable, mtctx->roundBuff.capacity, mtctx->allJobsCompleted }; return ZSTDMT_rust_initCStream( &projection, &state, ZSTDMT_initCStreamResize, ZSTDMT_initCStreamWaitForAllJobs, ZSTDMT_initCStreamReleaseAllJobResources, ZSTDMT_initCStreamApplyParameters, ZSTDMT_initCStreamPrepareDictionary, ZSTDMT_initCStreamSetTargetPrefixSize, ZSTDMT_initCStreamSetTargetSectionSize, ZSTDMT_initCStreamSetRsync, ZSTDMT_initCStreamSetBufferSize, ZSTDMT_initCStreamResizeRoundBuffer, ZSTDMT_initCStreamResetStream, ZSTDMT_initCStreamUpdateDictionary, ZSTDMT_initCStreamSerialReset); } /* ZSTDMT_writeLastEmptyBlock() * Write a single empty block with an end-of-frame to finish a frame. * Job must be created from streaming variant. * This function is always successful if expected conditions are fulfilled. */ static void ZSTDMT_writeLastEmptyBlock(ZSTDMT_jobDescription* job) { ZSTDMT_RustEmptyBlockJobProjection projection; ZSTDMT_RustEmptyBlockResult result; assert(job->lastJob == 1); assert(job->src.size == 0); /* last job is empty -> will be simplified into a last empty block */ assert(job->firstJob == 0); /* cannot be first job, as it also needs to create frame header */ assert(job->dstBuff.start == NULL); /* invoked from streaming variant only (otherwise, dstBuff might be user's output) */ assert(job->consumed == 0); projection = (ZSTDMT_RustEmptyBlockJobProjection){ job->lastJob, job->src.size, job->firstJob, job->dstBuff.start, job->dstBuff.capacity, job->consumed }; result = ZSTDMT_rust_writeLastEmptyBlock( &projection, job->bufPool, ZSTDMT_getBufferForRust); job->dstBuff = (Buffer){ result.buffer.start, result.buffer.capacity }; if (job->dstBuff.start == NULL) { assert(!result.clearSource); job->cSize = result.cSize; return; } assert(result.clearSource); assert(job->dstBuff.capacity >= ZSTD_blockHeaderSize); /* no buffer should ever be that small */ if (result.clearSource) job->src = kNullRange; job->cSize = result.cSize; assert(!ZSTD_isError(job->cSize)); } static void ZSTDMT_prepareCompressionJobDescriptor( void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* initialization) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; BYTE const* const src = (const BYTE*)initialization->srcStart; DEBUGLOG(5, "ZSTDMT_createCompressionJob: preparing job %u to compress %u bytes with %u preload ", initialization->jobNumber, (U32)initialization->srcSize, (U32)initialization->prefixSize); assert(mtctx->inBuff.filled >= initialization->srcSize); job->src = (Range){ src, initialization->srcSize }; job->prefix = (Range){ (const BYTE*)initialization->prefixStart, initialization->prefixSize }; job->consumed = 0; job->cSize = 0; job->params = mtctx->params; job->cdict = initialization->firstJob ? mtctx->cdict : NULL; job->fullFrameSize = mtctx->frameContentSize; job->dstBuff = g_nullBuffer; job->cctxPool = mtctx->cctxPool; job->bufPool = mtctx->bufPool; job->seqPool = mtctx->seqPool; job->serial = &mtctx->serial; job->jobID = initialization->jobNumber; job->firstJob = initialization->firstJob; job->lastJob = initialization->lastJob; job->frameChecksumNeeded = initialization->frameChecksumNeeded; job->dstFlushed = 0; } static void ZSTDMT_prepareCompressionJobInputState( void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* initialization) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; (void)jobID; /* Update the round buffer position and clear the input buffer to be reset. */ mtctx->roundBuff.pos += initialization->roundBuffPosDelta; mtctx->inBuff.buffer = g_nullBuffer; mtctx->inBuff.filled = 0; mtctx->inBuff.prefix = (Range){ (const BYTE*)initialization->nextPrefixStart, initialization->nextPrefixSize }; } static void ZSTDMT_prepareCompressionJobSetFrameEnded(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; mtctx->frameEnded = 1; } static void ZSTDMT_prepareCompressionJobClearChecksumFlag(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; mtctx->params.fParams.checksumFlag = 0; } static void ZSTDMT_prepareCompressionJob( void* opaque, unsigned jobID, const ZSTDMT_RustJobInitialization* initialization) { ZSTDMT_rust_prepareCompressionJob( opaque, jobID, initialization, ZSTDMT_prepareCompressionJobDescriptor, ZSTDMT_prepareCompressionJobInputState, ZSTDMT_prepareCompressionJobSetFrameEnded, ZSTDMT_prepareCompressionJobClearChecksumFlag); } static void ZSTDMT_writeEmptyCompressionJob(void* opaque, unsigned jobID) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_writeLastEmptyBlock(&mtctx->jobs[jobID]); } static int ZSTDMT_tryAddCompressionJob(void* opaque, unsigned jobID) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; return POOL_tryAdd(mtctx->factory, ZSTDMT_compressionJob, &mtctx->jobs[jobID]); } static ZSTDMT_RustStreamCreateJobResult ZSTDMT_createCompressionJob( void* opaque, size_t srcSize, unsigned end) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTD_EndDirective const endOp = (ZSTD_EndDirective)end; ZSTDMT_RustCreateJobProjection projection; ZSTDMT_RustCreateJobResult result; ZSTDMT_RustStreamCreateJobResult streamResult; /* Rust fills the reusable input range in its local projection before * calling this C-owned job-construction callback. Publish that count * before the callback prepares a job and checks the C context. */ assert(srcSize <= mtctx->inBuff.buffer.capacity); mtctx->inBuff.filled = srcSize; projection = (ZSTDMT_RustCreateJobProjection){ mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask, mtctx->jobReady, mtctx->inBuff.buffer.start, srcSize, mtctx->inBuff.filled, mtctx->inBuff.prefix.start, mtctx->inBuff.prefix.size, mtctx->targetPrefixSize, (unsigned)(endOp == ZSTD_e_end), (unsigned)mtctx->params.fParams.checksumFlag }; result = ZSTDMT_rust_createCompressionJob( &projection, mtctx, ZSTDMT_prepareCompressionJob, ZSTDMT_writeEmptyCompressionJob, ZSTDMT_tryAddCompressionJob); /* Keep the private C mutation in this callback, but let Rust decide * whether its local inBuffFilled value may be published afterwards. */ streamResult = (ZSTDMT_RustStreamCreateJobResult){ result.returnCode, (unsigned)(mtctx->inBuff.buffer.start != NULL) }; if (result.action == ZSTDMT_CREATE_JOB_TABLE_FULL) { DEBUGLOG(5, "ZSTDMT_createCompressionJob: will not create new job : table is full"); assert((mtctx->nextJobID & mtctx->jobIDMask) == (mtctx->doneJobID & mtctx->jobIDMask)); return streamResult; } mtctx->nextJobID = result.nextJobID; mtctx->jobReady = result.jobReady; if (result.action == ZSTDMT_CREATE_JOB_EMPTY) { DEBUGLOG(5, "ZSTDMT_createCompressionJob: creating a last empty block to end frame"); assert(endOp == ZSTD_e_end); /* only possible case : need to end the frame with an empty last block */ return streamResult; } DEBUGLOG(5, "ZSTDMT_createCompressionJob: posting job %u : %u bytes (end:%u, jobNb == %u (mod:%u))", result.jobNumber, (U32)mtctx->jobs[result.jobID].src.size, mtctx->jobs[result.jobID].lastJob, result.jobNumber, result.jobID); if (result.jobReady) DEBUGLOG(5, "ZSTDMT_createCompressionJob: no worker available for job %u", result.jobNumber); return streamResult; } /* The Rust flush state machine receives only this synchronized scalar view. * The descriptor, condition variable, serial checksum state, and buffer pool * remain private to C. */ static void ZSTDMT_projectFlushJob(void* opaque, unsigned jobID, unsigned blockToFlush, ZSTDMT_RustFlushJobProjection* projection) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex); if (blockToFlush && (mtctx->doneJobID < mtctx->nextJobID)) { assert(job->dstFlushed <= job->cSize); while (job->dstFlushed == job->cSize) { /* nothing to flush */ if (job->consumed == job->src.size) break; ZSTD_pthread_cond_wait(&job->job_cond, &job->job_mutex); } } *projection = (ZSTDMT_RustFlushJobProjection){ job->consumed, job->cSize, job->src.size, job->dstBuff.start, job->dstBuff.capacity, job->dstFlushed, job->frameChecksumNeeded }; ZSTD_pthread_mutex_unlock(&job->job_mutex); } static void ZSTDMT_addFrameChecksum(void* opaque, unsigned jobID, ZSTDMT_RustFlushJobProjection* projection) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; ZSTDMT_RustFrameChecksumProjection const checksumProjection = { job->dstBuff.start, job->dstBuff.capacity, job->cSize, &mtctx->serial.xxhState }; size_t const cSize = ZSTDMT_rust_writeFrameChecksum(&checksumProjection); assert(projection->frameChecksumNeeded); assert(projection->consumed == projection->srcSize); assert(projection->cSize == job->cSize); job->cSize = cSize; projection->cSize = cSize; if (!ZSTD_isError(cSize)) { job->frameChecksumNeeded = 0; projection->frameChecksumNeeded = 0; } } static void ZSTDMT_updateFlushJob(void* opaque, unsigned jobID, size_t dstFlushed) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; assert(dstFlushed <= job->cSize); job->dstFlushed = dstFlushed; } static void ZSTDMT_completeFlushJob(void* opaque, unsigned jobID, size_t srcSize, size_t cSize) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_jobDescription* const job = &mtctx->jobs[jobID]; assert(jobID == (mtctx->doneJobID & mtctx->jobIDMask)); assert(job->src.size == srcSize); assert(job->cSize == cSize); assert(job->dstFlushed == cSize); ZSTDMT_releaseBuffer(mtctx->bufPool, job->dstBuff); job->dstBuff = g_nullBuffer; job->cSize = 0; /* ensure this job slot is considered "not started" in future check */ mtctx->consumed += srcSize; mtctx->produced += cSize; mtctx->doneJobID++; } static void ZSTDMT_flushWaitForAllJobs(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_waitForAllJobsCompleted(mtctx); } static void ZSTDMT_flushReleaseAllJobResources(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTDMT_releaseAllJobResources(mtctx); } /*! ZSTDMT_flushProduced() : * flush whatever data has been produced but not yet flushed in current job. * move to next job if current one is fully flushed. * `output` : `pos` will be updated with amount of data flushed . * `blockToFlush` : if >0, the function will block and wait if there is no data available to flush . * @return : amount of data remaining within internal buffer, 0 if no more, 1 if unknown but > 0, or an error code */ static size_t ZSTDMT_flushProduced(ZSTDMT_CCtx* mtctx, ZSTD_outBuffer* output, unsigned blockToFlush, ZSTD_EndDirective end) { ZSTDMT_RustFlushContextProjection const context = { mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask, mtctx->jobReady, mtctx->frameEnded, mtctx->inBuff.filled }; ZSTDMT_RustFlushProducedResult const result = ZSTDMT_rust_flushProduced( &context, output->dst, output->size, output->pos, blockToFlush, (unsigned)end, mtctx, ZSTDMT_projectFlushJob, ZSTDMT_addFrameChecksum, ZSTDMT_updateFlushJob, ZSTDMT_completeFlushJob, ZSTDMT_flushWaitForAllJobs, ZSTDMT_flushReleaseAllJobResources); assert(output->size >= output->pos); output->pos = result.outputPos; if (result.updateAllJobsCompleted) mtctx->allJobsCompleted = result.allJobsCompleted; return result.result; } static ZSTDMT_RustStreamFlushResult ZSTDMT_streamFlushProduced( void* opaque, void* outputDst, size_t outputSize, size_t outputPos, unsigned blockToFlush, unsigned end) { ZSTD_outBuffer output = { outputDst, outputSize, outputPos }; size_t const result = ZSTDMT_flushProduced( (ZSTDMT_CCtx*)opaque, &output, blockToFlush, (ZSTD_EndDirective)end); return (ZSTDMT_RustStreamFlushResult){ result, output.pos }; } /** * Returns the range of data used by the earliest job that is not yet complete. * If the data of the first job is broken up into two segments, we cover both * sections. */ static Range ZSTDMT_getInputDataInUse(ZSTDMT_CCtx* mtctx) { ZSTDMT_RustInputRange const range = ZSTDMT_rust_getInputDataInUse( mtctx->doneJobID, mtctx->nextJobID, mtctx->jobIDMask, mtctx->roundBuff.capacity, mtctx->targetSectionSize, mtctx, ZSTDMT_projectJob); return (Range){ range.start, range.size }; } int ZSTDMT_doesOverlapWindow(Buffer buffer, ZSTD_window_t window); static void ZSTDMT_waitForLdmLock(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->serial.ldmWindowMutex); } static int ZSTDMT_waitForLdmOverlap( void* opaque, void* bufferStart, size_t bufferCapacity) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; Buffer const buffer = { bufferStart, bufferCapacity }; return ZSTDMT_doesOverlapWindow(buffer, mtctx->serial.ldmWindow); } static void ZSTDMT_waitForLdmWait(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; DEBUGLOG(5, "Waiting for LDM to finish..."); ZSTD_pthread_cond_wait(&mtctx->serial.ldmWindowCond, &mtctx->serial.ldmWindowMutex); } static void ZSTDMT_waitForLdmUnlock(void* opaque) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; DEBUGLOG(6, "Done waiting for LDM to finish"); ZSTD_pthread_mutex_unlock(&mtctx->serial.ldmWindowMutex); } static void ZSTDMT_waitForLdmComplete(ZSTDMT_CCtx* mtctx, Buffer buffer) { int const ldmEnabled = mtctx->params.ldmParams.enableLdm == ZSTD_ps_enable; ZSTDMT_RustWaitForLdmState state; if (ldmEnabled) { DEBUGLOG(5, "ZSTDMT_waitForLdmComplete"); DEBUGLOG(5, "source [0x%zx, 0x%zx)", (size_t)buffer.start, (size_t)buffer.start + buffer.capacity); } state.callbackContext = mtctx; state.bufferStart = buffer.start; state.bufferCapacity = buffer.capacity; state.ldmEnabled = ldmEnabled; state.lock = ZSTDMT_waitForLdmLock; state.overlaps = ZSTDMT_waitForLdmOverlap; state.wait = ZSTDMT_waitForLdmWait; state.unlock = ZSTDMT_waitForLdmUnlock; ZSTDMT_rust_waitForLdmComplete(&state); } static void ZSTDMT_inputRangeWaitForLdm( void* opaque, void* bufferStart, size_t bufferCapacity) { ZSTDMT_waitForLdmComplete((ZSTDMT_CCtx*)opaque, (Buffer){ bufferStart, bufferCapacity }); } static void ZSTDMT_inputRangeMovePrefix( void* opaque, void* destination, const void* source, size_t size, size_t roundBufferPos) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; /* ZSTD_invalidateRepCodes() doesn't work for extDict variants. * Simply copy the prefix to the beginning in that case. */ ZSTD_memmove(destination, source, size); mtctx->inBuff.prefix.start = (const BYTE*)destination; mtctx->roundBuff.pos = roundBufferPos; } static void ZSTDMT_inputRangePublishBuffer( void* opaque, void* bufferStart, size_t bufferCapacity) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; Buffer const buffer = { bufferStart, bufferCapacity }; DEBUGLOG(5, "Using prefix range [%zx, %zx)", (size_t)mtctx->inBuff.prefix.start, (size_t)mtctx->inBuff.prefix.start + mtctx->inBuff.prefix.size); DEBUGLOG(5, "Using source range [%zx, %zx)", (size_t)buffer.start, (size_t)buffer.start + buffer.capacity); mtctx->inBuff.buffer = buffer; mtctx->inBuff.filled = 0; assert(mtctx->roundBuff.pos + buffer.capacity <= mtctx->roundBuff.capacity); } /** * Attempts to set the inBuff to the next section to fill. * If any part of the new section is still in use we give up. * Returns non-zero if the buffer is filled. */ static int ZSTDMT_tryGetInputRange(ZSTDMT_CCtx* mtctx) { Range const inUse = ZSTDMT_getInputDataInUse(mtctx); ZSTDMT_RustTryGetInputRangeProjection const projection = { mtctx->roundBuff.buffer, mtctx->roundBuff.capacity, mtctx->roundBuff.pos, mtctx->inBuff.prefix.start, mtctx->inBuff.prefix.size, mtctx->targetSectionSize, inUse.start, inUse.size }; ZSTDMT_RustTryGetInputRangeResult const result = ZSTDMT_rust_tryGetInputRange(&projection); DEBUGLOG(5, "ZSTDMT_tryGetInputRange"); assert(mtctx->inBuff.buffer.start == NULL); if (!ZSTDMT_rust_commitInputRange( &result, mtctx->roundBuff.buffer, mtctx->inBuff.prefix.start, mtctx->inBuff.prefix.size, mtctx, ZSTDMT_inputRangeWaitForLdm, ZSTDMT_inputRangeMovePrefix, ZSTDMT_inputRangePublishBuffer)) { DEBUGLOG(5, "Waiting for buffer..."); return 0; } return 1; } /** * Searches through the input for a synchronization point. If one is found, we * will instruct the caller to flush, and return the number of bytes to load. * Otherwise, we will load as many bytes as possible and instruct the caller * to continue as normal. */ /* Adapter callback for the C-owned reusable input range. */ static int ZSTDMT_streamTryGetInputRange( void* opaque, ZSTDMT_RustStreamInputRangeProjection* projection) { ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque; int const ready = ZSTDMT_tryGetInputRange(mtctx); *projection = (ZSTDMT_RustStreamInputRangeProjection){ mtctx->inBuff.buffer.start, mtctx->inBuff.buffer.capacity, mtctx->inBuff.filled }; return ready; } size_t ZSTDMT_nextInputSizeHint(const ZSTDMT_CCtx* mtctx) { return ZSTDMT_rust_nextInputSizeHint(mtctx->targetSectionSize, mtctx->inBuff.filled); } /** ZSTDMT_compressStream_generic() : * internal use only - exposed to be invoked from zstd_compress.c * assumption : output and input are valid (pos <= size) * @return : minimum amount of data remaining to flush, 0 if none */ size_t ZSTDMT_compressStream_generic(ZSTDMT_CCtx* mtctx, ZSTD_outBuffer* output, ZSTD_inBuffer* input, ZSTD_EndDirective endOp) { ZSTDMT_RustCompressStreamContextProjection const context = { mtctx->frameEnded, (unsigned)mtctx->jobReady, mtctx->inBuff.buffer.start, mtctx->inBuff.buffer.capacity, mtctx->inBuff.filled, mtctx->targetSectionSize, mtctx->params.rsyncable, mtctx->rsync.primePower, mtctx->rsync.hitMask }; ZSTDMT_RustStreamInputProjection const inputProjection = { input->src, input->size, input->pos }; ZSTDMT_RustStreamOutputProjection const outputProjection = { output->dst, output->size, output->pos }; ZSTDMT_RustCompressStreamResult const result = ZSTDMT_rust_compressStreamGeneric( &context, &inputProjection, &outputProjection, (unsigned)endOp, mtctx, ZSTDMT_streamTryGetInputRange, ZSTDMT_createCompressionJob, ZSTDMT_streamFlushProduced); DEBUGLOG(5, "ZSTDMT_compressStream_generic (endOp=%u, srcSize=%u)", (U32)endOp, (U32)(input->size - input->pos)); assert(output->pos <= output->size); assert(input->pos <= input->size); /* Rust owns the decision to publish its local count. C retains the * private field mutation and receives either the count or zero when the * job callback cleared the range. */ mtctx->inBuff.filled = result.inBuffFilled; input->pos = result.inputPos; output->pos = result.outputPos; DEBUGLOG(5, "end of ZSTDMT_compressStream_generic: remainingToFlush = %u", (U32)result.result); return result.result; }