ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
a0aa1cd070ad04541d46139c5f4d75001eadf680
zstd-rs/lib/compress/zstdmt_compress.c
T
ddidderr a0aa1cd070 fix(mt): pass boolean LDM resource flag 
The resource-acquisition projection calls its field `ldmEnabled`, but the
first implementation passed the three-state `ZSTD_ParamSwitch_e` value
unchanged. `ZSTD_ps_disable` is 2, so Rust interpreted disabled LDM as enabled
and rejected ordinary jobs whose raw sequence store was intentionally absent.
That surfaced as an allocation error on the small stdin compression path.

Convert the C enum to the same boolean semantic used by the Rust policy before
crossing the ABI boundary. The projection declaration also now precedes the
callback setup statements, keeping the native C90 warning-free style.

Test Plan:
- `ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings` -- passed
- `ulimit -v 41943040; cargo +nightly fmt --manifest-path rust/Cargo.toml --all -- --check` -- passed
- `git diff --cached --check` -- passed
- The first capped full upstream run failed at `test-zstd` on the small stdin path with the enum/boolean regression; the suite will be rerun after this fix.
2026-07-20 17:23:01 +02:00

3337 lines
128 KiB
C
Raw Blame History

/*
* 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 parameter leaf 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];
ZSTD_compressionParameters ZSTD_rust_params_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 <stdio.h>
# include <unistd.h>
# include <sys/times.h>
# 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 {
unsigned firstJob;
unsigned lastJob;
const ZSTDMT_RustCompressionJobFrameHeaderState* frameHeaderState;
} 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)
&& sizeof(ZSTDMT_RustCompressionJobProjection)
== 2 * sizeof(unsigned) + 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 struct {
void* callbackContext;
size_t srcSize;
ZSTDMT_compressionJobVoidFn ensureFinished;
ZSTDMT_compressionJobSizeFn publishError;
ZSTDMT_compressionJobSizeFn publishLastBlockSize;
ZSTDMT_compressionJobVoidFn releaseSeq;
ZSTDMT_compressionJobVoidFn releaseCCtx;
ZSTDMT_compressionJobSizeFn publishConsumed;
ZSTDMT_compressionJobVoidFn signal;
} 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, publishError)
== 3 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishLastBlockSize)
== 4 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseSeq)
== 5 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, releaseCCtx)
== 6 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, publishConsumed)
== 7 * sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobFinishProjection, signal)
== 8 * sizeof(void*)
&& sizeof(ZSTDMT_RustCompressionJobFinishProjection) == 9 * 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_compressionJobVoidFn applySequences,
ZSTDMT_compressionJobCompressFn compressJob,
ZSTDMT_compressionJobVoidFn traceJob);
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;
U32 checksum;
} 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, checksum)
== 3 * sizeof(void*)
&& sizeof(U32) == 4
&& 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;
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 size_t (*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_writeEmptyJobFn)(void* opaque, unsigned jobID);
typedef int (*ZSTDMT_tryAddJobFn)(void* opaque, unsigned jobID);
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];
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_serialGenerateLdmFn)(
void* opaque, ZSTDMT_RustRawSeqStore* seqStore,
const void* src, size_t srcSize);
typedef void (*ZSTDMT_serialUpdateChecksumFn)(
void* opaque, const void* src, size_t srcSize);
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,
ZSTDMT_serialWaitForTurnFn waitForTurn,
ZSTDMT_serialGenerateLdmFn generateLdm,
ZSTDMT_serialUpdateChecksumFn updateChecksum,
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_initDrainFn)(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 void (*ZSTDMT_serialResetVoidFn)(void* opaque);
typedef void (*ZSTDMT_serialResetSetNbSeqFn)(void* opaque, size_t nbSeq);
typedef int (*ZSTDMT_serialResetResizeFn)(void* opaque);
int ZSTDMT_rust_serialStateReset(
int enableLdm, int checksumEnabled, size_t maxNbSeq, void* opaque,
ZSTDMT_serialResetVoidFn resetNextJob,
ZSTDMT_serialResetVoidFn resetChecksum,
ZSTDMT_serialResetSetNbSeqFn setNbSeq,
ZSTDMT_serialResetVoidFn resetWindow,
ZSTDMT_serialResetResizeFn resizeTables,
ZSTDMT_serialResetVoidFn zeroTables,
ZSTDMT_serialResetVoidFn loadDictionary,
ZSTDMT_serialResetVoidFn copyWindow);
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_initDrainFn drain,
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);
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 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);
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);
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);
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;
size_t hashSize;
size_t numBuckets;
unsigned bucketLog;
unsigned prevBucketLog;
ZSTD_customMem cMem;
} ZSTDMT_serialResetContext;
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)
{
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(
context->hashSize, context->cMem);
}
if (serialState->ldmState.bucketOffsets == NULL ||
context->prevBucketLog < context->bucketLog) {
ZSTD_customFree(serialState->ldmState.bucketOffsets, context->cMem);
serialState->ldmState.bucketOffsets = (BYTE*)ZSTD_customMalloc(
context->numBuckets, context->cMem);
}
return !serialState->ldmState.hashTable || !serialState->ldmState.bucketOffsets;
}
static void ZSTDMT_serialResetZeroTables(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
SerialState* const serialState = context->serialState;
ZSTD_memset(serialState->ldmState.hashTable, 0, context->hashSize);
ZSTD_memset(serialState->ldmState.bucketOffsets, 0, context->numBuckets);
}
static void ZSTDMT_serialResetLoadDictionary(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
SerialState* const serialState = context->serialState;
serialState->ldmState.loadedDictEnd = 0;
if (context->dictSize > 0 &&
context->dictContentType == ZSTD_dct_rawContent) {
BYTE const* const dictEnd = (const BYTE*)context->dict + context->dictSize;
ZSTD_window_update(&serialState->ldmState.window,
context->dict, context->dictSize,
/* forceNonContiguous */ 0);
ZSTD_ldm_fillHashTable(&serialState->ldmState,
(const BYTE*)context->dict, dictEnd,
&context->params->ldmParams);
serialState->ldmState.loadedDictEnd = context->params->forceWindow
? 0 : (U32)(dictEnd - serialState->ldmState.window.base);
}
}
static void ZSTDMT_serialResetCopyWindow(void* opaque)
{
ZSTDMT_serialResetContext* const context = (ZSTDMT_serialResetContext*)opaque;
context->serialState->ldmWindow = context->serialState->ldmState.window;
}
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;
/* Adjust parameters */
if (params.ldmParams.enableLdm == ZSTD_ps_enable) {
DEBUGLOG(4, "LDM window size = %u KB", (1U << params.cParams.windowLog) >> 10);
ZSTD_ldm_adjustParameters(&params.ldmParams, &params.cParams);
assert(params.ldmParams.hashLog >= params.ldmParams.bucketSizeLog);
assert(params.ldmParams.hashRateLog < 32);
} else {
ZSTD_memset(&params.ldmParams, 0, sizeof(params.ldmParams));
}
context.serialState = serialState;
context.seqPool = seqPool;
context.params = &params;
context.dict = dict;
context.dictSize = dictSize;
context.dictContentType = dictContentType;
context.cMem = params.customMem;
context.hashSize = 0;
context.numBuckets = 0;
context.bucketLog = 0;
context.prevBucketLog = 0;
if (params.ldmParams.enableLdm == ZSTD_ps_enable) {
unsigned const hashLog = params.ldmParams.hashLog;
context.hashSize = ((size_t)1 << hashLog) * sizeof(ldmEntry_t);
context.bucketLog = params.ldmParams.hashLog - params.ldmParams.bucketSizeLog;
context.prevBucketLog =
serialState->params.ldmParams.hashLog -
serialState->params.ldmParams.bucketSizeLog;
context.numBuckets = (size_t)1 << context.bucketLog;
}
if (ZSTDMT_rust_serialStateReset(
params.ldmParams.enableLdm == ZSTD_ps_enable,
params.fParams.checksumFlag,
ZSTD_ldm_getMaxNbSeq(params.ldmParams, jobSize),
&context,
ZSTDMT_serialResetNextJob,
ZSTDMT_serialResetChecksum,
ZSTDMT_serialResetSetNbSeq,
ZSTDMT_serialResetWindow,
ZSTDMT_serialResetResizeTables,
ZSTDMT_serialResetZeroTables,
ZSTDMT_serialResetLoadDictionary,
ZSTDMT_serialResetCopyWindow)) {
return 1;
}
serialState->params = params;
serialState->params.jobSize = (U32)jobSize;
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_generateLdm(
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;
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);
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;
/* 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_updateChecksum(
void* opaque, const void* src, size_t srcSize)
{
SerialState* const serialState = (SerialState*)opaque;
XXH64_update(&serialState->xxhState, src, srcSize);
}
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);
}
static void
ZSTDMT_serialState_applySequences(const SerialState* serialState, /* just for an assert() check */
ZSTD_CCtx* jobCCtx,
const RawSeqStore_t* seqStore)
{
if (seqStore->size > 0) {
DEBUGLOG(5, "ZSTDMT_serialState_applySequences: uploading %u external sequences", (unsigned)seqStore->size);
assert(serialState->params.ldmParams.enableLdm == ZSTD_ps_enable); (void)serialState;
assert(jobCCtx);
ZSTD_referenceExternalSequences(jobCCtx, seqStore->seq, seqStore->size);
}
}
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_compressionJobProgress(void* opaque, size_t cSize, size_t consumed)
{
ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)opaque;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->cSize += cSize;
job->consumed = consumed;
DEBUGLOG(5, "ZSTDMT_compressionJob: compress new block : cSize==%u bytes (total: %u)",
(U32)cSize, (U32)job->cSize);
ZSTD_pthread_cond_signal(&job->job_cond); /* warns some more data is ready to be flushed */
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
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,
ZSTDMT_serialState_waitForTurn,
ZSTDMT_serialState_generateLdm,
ZSTDMT_serialState_updateChecksum,
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 void ZSTDMT_compressionJobApplySequences(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
/* External Sequences can only be applied after CCtx initialization. */
ZSTDMT_serialState_applySequences(job->serial, state->cctx,
&state->rawSeqStore);
}
static ZSTDMT_chunkProcessResult ZSTDMT_compressionJobCompress(
void* opaque, unsigned lastJob)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
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 */
return ZSTDMT_rust_compressJobChunks(
state->cctx, job->src.start, job->src.size,
state->dstBuff.start, state->dstBuff.capacity, chunkSize, lastJob,
job, ZSTDMT_compressionJobProgress);
}
static void ZSTDMT_compressionJobTrace(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
if (!job->firstJob) {
/* 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));
}
ZSTD_CCtx_trace(state->cctx, 0);
}
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_compressionJobPublishError(void* opaque, size_t error)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->cSize = error;
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
static void ZSTDMT_compressionJobPublishLastBlockSize(void* opaque, size_t lastCBlockSize)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->cSize += lastCBlockSize;
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
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_compressionJobPublishConsumed(void* opaque, size_t srcSize)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
assert(srcSize == job->src.size);
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
job->consumed = srcSize; /* when job->consumed == job->src.size , compression job is presumed completed */
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
static void ZSTDMT_compressionJobSignal(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
ZSTD_pthread_cond_signal(&job->job_cond);
ZSTD_pthread_mutex_unlock(&job->job_mutex);
}
/* 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_RustCompressionJobFinishProjection finishProjection;
ZSTDMT_RustCompressionJobProjection const projection = {
job->firstJob,
job->lastJob,
&state.frameHeaderState
};
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;
finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){
&state,
job->src.size,
ZSTDMT_compressionJobEnsureFinished,
ZSTDMT_compressionJobPublishError,
ZSTDMT_compressionJobPublishLastBlockSize,
ZSTDMT_compressionJobReleaseSeq,
ZSTDMT_compressionJobReleaseCCtx,
ZSTDMT_compressionJobPublishConsumed,
ZSTDMT_compressionJobSignal
};
ZSTDMT_rust_compressionJob(
&projection, &finishProjection, &state,
ZSTDMT_compressionJobAcquireResources,
ZSTDMT_compressionJobPrepareParameters,
ZSTDMT_compressionJobGenerateSequences,
ZSTDMT_compressionJobBegin,
ZSTDMT_compressionJobApplySequences,
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<<RSYNC_MIN_BLOCK_LOG)
typedef struct {
U64 hash;
U64 hitMask;
U64 primePower;
} RSyncState_t;
struct ZSTDMT_CCtx_s {
POOL_ctx* factory;
ZSTDMT_jobDescription* jobs;
ZSTDMT_bufferPool* bufPool;
ZSTDMT_CCtxPool* cctxPool;
ZSTDMT_seqPool* seqPool;
ZSTD_CCtx_params params;
size_t targetSectionSize;
size_t targetPrefixSize;
int jobReady; /* 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)
{
if (jobTable == NULL) return;
ZSTDMT_job_table_destroy_sync(jobTable, nbJobs, sizeof(*jobTable));
ZSTDMT_rust_job_table_free(jobTable, nbJobs, sizeof(*jobTable), cMem);
}
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; jobNb<nbJobs; jobNb++) {
ZSTDMT_jobDescription* const job =
(ZSTDMT_jobDescription*)(table + jobNb * jobSize);
initError |= ZSTD_pthread_mutex_init(&job->job_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; jobNb<nbJobs; jobNb++) {
ZSTDMT_jobDescription* const job =
(ZSTDMT_jobDescription*)(table + jobNb * jobSize);
ZSTD_pthread_mutex_destroy(&job->job_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 */
int const compressionLevel = cctxParams->compressionLevel;
DEBUGLOG(5, "ZSTDMT_updateCParams_whileCompressing (level:%i)",
compressionLevel);
mtctx->params.compressionLevel = 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
};
mtctx->params.cParams =
ZSTD_rust_params_updateCParamsWhileCompressing(projection);
}
}
/* 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_initCStreamDrain(void* opaque)
{
ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque;
ZSTDMT_CCtx* const mtctx = state->mtctx;
ZSTDMT_waitForAllJobsCompleted(mtctx);
ZSTDMT_releaseAllJobResources(mtctx);
mtctx->allJobsCompleted = 1;
}
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 size_t ZSTDMT_initCStreamPrepareDictionary(void* opaque)
{
ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque;
ZSTDMT_CCtx* const mtctx = state->mtctx;
ZSTD_freeCDict(mtctx->cdictLocal);
if (state->dict) {
mtctx->cdictLocal = ZSTD_createCDict_advanced(
state->dict, state->dictSize, ZSTD_dlm_byCopy,
state->dictContentType, state->params.cParams, mtctx->cMem);
mtctx->cdict = mtctx->cdictLocal;
if (mtctx->cdictLocal == NULL) return ERROR(memory_allocation);
} else {
mtctx->cdictLocal = NULL;
mtctx->cdict = state->cdict;
}
return 0;
}
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 size_t ZSTDMT_initCStreamUpdateDictionary(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;
if (state->dict) {
if (state->dictContentType == ZSTD_dct_rawContent) {
mtctx->inBuff.prefix.start = (const BYTE*)state->dict;
mtctx->inBuff.prefix.size = state->dictSize;
} else {
/* 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;
if (mtctx->cdictLocal == NULL) return ERROR(memory_allocation);
}
} else {
mtctx->cdict = state->cdict;
}
return 0;
}
static size_t ZSTDMT_initCStreamSerialReset(void* opaque, size_t targetSectionSize)
{
ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque;
if (ZSTDMT_serialState_reset(
&state->mtctx->serial, state->mtctx->seqPool, state->params,
targetSectionSize, state->dict, state->dictSize,
state->dictContentType))
return ERROR(memory_allocation);
return 0;
}
/* ====================================== */
/* ======= 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_initCStreamDrain,
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_prepareCompressionJob(
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;
/* 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
};
if (initialization->lastJob) {
mtctx->frameEnded = 1;
if (initialization->clearChecksumFlag)
mtctx->params.fParams.checksumFlag = 0;
}
}
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 size_t 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;
/* 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);
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 result.returnCode;
}
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 result.returnCode;
}
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 result.returnCode;
}
/* 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];
U32 const checksum = (U32)XXH64_digest(&mtctx->serial.xxhState);
ZSTDMT_RustFrameChecksumProjection const checksumProjection = {
job->dstBuff.start,
job->dstBuff.capacity,
job->cSize,
checksum
};
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);
}
/**
* 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);
Buffer buffer;
DEBUGLOG(5, "ZSTDMT_tryGetInputRange");
assert(mtctx->inBuff.buffer.start == NULL);
if (!result.ready) {
DEBUGLOG(5, "Waiting for buffer...");
return 0;
}
if (result.movePrefix) {
/* ZSTD_invalidateRepCodes() doesn't work for extDict variants.
* Simply copy the prefix to the beginning in that case.
*/
BYTE* const start = (BYTE*)mtctx->roundBuff.buffer;
size_t const prefixSize = mtctx->inBuff.prefix.size;
buffer.start = start;
buffer.capacity = prefixSize;
ZSTDMT_waitForLdmComplete(mtctx, buffer);
ZSTD_memmove(start, mtctx->inBuff.prefix.start, prefixSize);
mtctx->inBuff.prefix.start = start;
mtctx->roundBuff.pos = result.roundBufferPos;
}
buffer.start = (BYTE*)result.bufferStart;
buffer.capacity = result.bufferCapacity;
ZSTDMT_waitForLdmComplete(mtctx, buffer);
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);
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);
/* A prepared job clears the C range while Rust still has the pre-job
* count in its local projection. Only publish the result when the range
* remains active, which is the no-job-created path. */
if (mtctx->inBuff.buffer.start != NULL)
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;
}
Reference in New Issue View Git Blame Copy Permalink