ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
7b94de78ff2c3634cd7ef32f0e16f4722c03578f
zstd-rs/lib/compress/zstdmt_compress.c
T
ddidderr 591794e15a feat(mt): move failed-job serial cleanup into Rust 
Move the failed-job serial-state orchestration out of zstdmt_compress.c.
Rust now owns the lock, skip decision, serial-counter publication,
broadcast, LDM cleanup ordering, and final unlock. C retains the pthread
objects, private LDM window, and error/debug leaves behind callbacks. Add
ABI layout checks and focused tests for both skipped predecessors and later
jobs that only need the lock/unlock pair.

Test Plan:
- cargo fmt --manifest-path rust/Cargo.toml -- --check
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040 && CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/Cargo.toml zstdmt_compress::tests::serial_ensure_finished -- --nocapture
- ulimit -v 41943040 && make -j1
- ulimit -v 41943040 && make -j1 -C tests test-fuzzer FUZZERTEST=-T3s FUZZER_FLAGS=--no-big-tests
2026-07-19 20:58:18 +02:00

2653 lines
102 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_memcpy, 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"
/* 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 {
unsigned firstJob;
unsigned lastJob;
} ZSTDMT_RustCompressionJobProjection;
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_compressionJobErrorFn)(void* opaque, size_t error);
typedef void (*ZSTDMT_compressionJobFinishFn)(void* opaque, size_t lastBlockSize);
void ZSTDMT_rust_compressionJob(
const ZSTDMT_RustCompressionJobProjection* projection,
void* opaque,
ZSTDMT_compressionJobStepFn acquireResources,
ZSTDMT_compressionJobVoidFn prepareParameters,
ZSTDMT_compressionJobVoidFn generateSequences,
ZSTDMT_compressionJobStepFn beginJob,
ZSTDMT_compressionJobVoidFn applySequences,
ZSTDMT_compressionJobStepFn writeFrameHeader,
ZSTDMT_compressionJobCompressFn compressJob,
ZSTDMT_compressionJobVoidFn traceJob,
ZSTDMT_compressionJobErrorFn setError,
ZSTDMT_compressionJobFinishFn finishJob);
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 {
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_flushErrorFn)(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_flushErrorFn onError);
/* 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;
} ZSTDMT_RustCompressStreamResult;
typedef int (*ZSTDMT_streamTryGetInputRangeFn)(
void* opaque, ZSTDMT_RustStreamInputRangeProjection* projection);
typedef int (*ZSTDMT_streamLoadInputFn)(void* opaque, const void* src, size_t size);
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_streamLoadInputFn loadInput,
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);
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 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);
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_compressionJobState;
static size_t ZSTDMT_compressionJobAcquireResources(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
state->cctx = ZSTDMT_getCCtx(job->cctxPool);
state->rawSeqStore = ZSTDMT_getSeq(job->seqPool);
state->dstBuff = job->dstBuff;
DEBUGLOG(5, "ZSTDMT_compressionJob: job %u", job->jobID);
if (state->cctx == NULL) return ERROR(memory_allocation);
if (state->dstBuff.start == NULL) {
state->dstBuff = ZSTDMT_getBuffer(job->bufPool);
if (state->dstBuff.start == NULL) return ERROR(memory_allocation);
job->dstBuff = state->dstBuff;
}
if (state->jobParams.ldmParams.enableLdm == ZSTD_ps_enable &&
state->rawSeqStore.seq == NULL)
return ERROR(memory_allocation);
return 0;
}
static void ZSTDMT_compressionJobPrepareParameters(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
/* Don't compute the checksum for chunks, since we compute it externally,
* but write it in the header. */
if (job->jobID != 0) state->jobParams.fParams.checksumFlag = 0;
/* Don't run LDM for the chunks, since we handle it externally. */
state->jobParams.ldmParams.enableLdm = ZSTD_ps_disable;
/* Correct nbWorkers to 0. */
state->jobParams.nbWorkers = 0;
}
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_compressionJobBegin(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
if (job->cdict) {
size_t const initError = ZSTD_compressBegin_advanced_internal(
state->cctx, NULL, 0, ZSTD_dct_auto, ZSTD_dtlm_fast, job->cdict,
&state->jobParams, job->fullFrameSize);
assert(job->firstJob); /* only allowed for first job */
return initError;
}
{ U64 const pledgedSrcSize = job->firstJob ? job->fullFrameSize : job->src.size;
size_t const forceWindowError = ZSTD_CCtxParams_setParameter(
&state->jobParams, ZSTD_c_forceMaxWindow, !job->firstJob);
if (ZSTD_isError(forceWindowError)) return forceWindowError;
if (!job->firstJob) {
size_t const err = ZSTD_CCtxParams_setParameter(
&state->jobParams, ZSTD_c_deterministicRefPrefix, 0);
if (ZSTD_isError(err)) return err;
}
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_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 size_t ZSTDMT_compressionJobWriteFrameHeader(void* opaque)
{
ZSTDMT_compressionJobState* const state =
(ZSTDMT_compressionJobState*)opaque;
ZSTDMT_jobDescription* const job = state->job;
size_t const hSize = ZSTD_compressContinue_public(
state->cctx, state->dstBuff.start, state->dstBuff.capacity,
job->src.start, 0);
if (ZSTD_isError(hSize)) return hSize;
DEBUGLOG(5, "ZSTDMT_compressionJob: flush and overwrite %u bytes of frame header (not first job)", (U32)hSize);
ZSTD_invalidateRepCodes(state->cctx);
return hSize;
}
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_compressionJobSetError(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_compressionJobFinish(void* opaque, size_t lastCBlockSize)
{
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);
/* release resources */
ZSTDMT_releaseSeq(job->seqPool, state->rawSeqStore);
ZSTDMT_releaseCCtx(job->cctxPool, state->cctx);
/* report */
ZSTD_PTHREAD_MUTEX_LOCK(&job->job_mutex);
if (ZSTD_isError(job->cSize)) assert(lastCBlockSize == 0);
job->cSize += lastCBlockSize;
job->consumed = job->src.size; /* when job->consumed == job->src.size , compression job is presumed completed */
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_RustCompressionJobProjection const projection = {
job->firstJob,
job->lastJob
};
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;
ZSTDMT_rust_compressionJob(
&projection, &state,
ZSTDMT_compressionJobAcquireResources,
ZSTDMT_compressionJobPrepareParameters,
ZSTDMT_compressionJobGenerateSequences,
ZSTDMT_compressionJobBegin,
ZSTDMT_compressionJobApplySequences,
ZSTDMT_compressionJobWriteFrameHeader,
ZSTDMT_compressionJobCompress,
ZSTDMT_compressionJobTrace,
ZSTDMT_compressionJobSetError,
ZSTDMT_compressionJobFinish);
}
/* ------------------------------------------ */
/* ===== 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;
static const RoundBuff_t kNullRoundBuff = {NULL, 0, 0};
#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->dstFlushed = job->dstFlushed;
ZSTD_pthread_mutex_unlock(&job->job_mutex);
projection->srcStart = job->src.start;
projection->srcSize = job->src.size;
projection->prefixStart = job->prefix.start;
projection->prefixSize = job->prefix.size;
}
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)
{
ZSTDMT_jobDescription* const jobTable = (ZSTDMT_jobDescription*)
ZSTDMT_rust_job_table_create(nbJobsPtr, sizeof(ZSTDMT_jobDescription), cMem);
if (jobTable == NULL) return NULL;
if (ZSTDMT_job_table_init_sync(jobTable, *nbJobsPtr, sizeof(*jobTable)) != 0) {
ZSTDMT_freeJobsTable(jobTable, *nbJobsPtr, cMem);
return NULL;
}
return jobTable;
}
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;
}
/* ZSTDMT_CCtxParam_setNbWorkers():
* Internal use only */
static size_t ZSTDMT_CCtxParam_setNbWorkers(ZSTD_CCtx_params* params, unsigned nbWorkers)
{
return ZSTD_CCtxParams_setParameter(params, ZSTD_c_nbWorkers, (int)nbWorkers);
}
MEM_STATIC ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced_internal(unsigned nbWorkers, ZSTD_customMem cMem, ZSTD_threadPool* pool)
{
ZSTDMT_CCtx* mtctx;
U32 nbJobs = nbWorkers + 2;
int initError;
DEBUGLOG(3, "ZSTDMT_createCCtx_advanced (nbWorkers = %u)", nbWorkers);
if (nbWorkers < 1) return NULL;
nbWorkers = MIN(nbWorkers , ZSTDMT_NBWORKERS_MAX);
if ((cMem.customAlloc!=NULL) ^ (cMem.customFree!=NULL))
/* invalid custom allocator */
return NULL;
mtctx = (ZSTDMT_CCtx*) ZSTD_customCalloc(sizeof(ZSTDMT_CCtx), cMem);
if (!mtctx) return NULL;
ZSTDMT_CCtxParam_setNbWorkers(&mtctx->params, nbWorkers);
mtctx->cMem = cMem;
mtctx->allJobsCompleted = 1;
if (pool != NULL) {
mtctx->factory = pool;
mtctx->providedFactory = 1;
}
else {
mtctx->factory = POOL_create_advanced(nbWorkers, 0, cMem);
mtctx->providedFactory = 0;
}
mtctx->jobs = ZSTDMT_createJobsTable(&nbJobs, cMem);
assert(nbJobs > 0); assert((nbJobs & (nbJobs - 1)) == 0); /* ensure nbJobs is a power of 2 */
mtctx->jobIDMask = nbJobs - 1;
mtctx->bufPool = ZSTDMT_createBufferPool(BUF_POOL_MAX_NB_BUFFERS(nbWorkers), cMem);
mtctx->cctxPool = ZSTDMT_createCCtxPool(nbWorkers, cMem);
mtctx->seqPool = ZSTDMT_createSeqPool(nbWorkers, cMem);
initError = ZSTDMT_serialState_init(&mtctx->serial);
mtctx->roundBuff = kNullRoundBuff;
if (!mtctx->factory | !mtctx->jobs | !mtctx->bufPool | !mtctx->cctxPool | !mtctx->seqPool | initError) {
ZSTDMT_freeCCtx(mtctx);
return NULL;
}
DEBUGLOG(3, "mt_cctx created, for %u threads", nbWorkers);
return mtctx;
}
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);
}
size_t ZSTDMT_freeCCtx(ZSTDMT_CCtx* mtctx)
{
if (mtctx==NULL) return 0; /* compatible with free on NULL */
if (!mtctx->providedFactory)
POOL_free(mtctx->factory); /* stop and free worker threads */
ZSTDMT_releaseAllJobResources(mtctx); /* release job resources into pools first */
ZSTDMT_freeJobsTable(mtctx->jobs, mtctx->jobIDMask+1, mtctx->cMem);
ZSTDMT_freeBufferPool(mtctx->bufPool);
ZSTDMT_freeCCtxPool(mtctx->cctxPool);
ZSTDMT_freeSeqPool(mtctx->seqPool);
ZSTDMT_serialState_free(&mtctx->serial);
ZSTD_freeCDict(mtctx->cdictLocal);
if (mtctx->roundBuff.buffer)
ZSTD_customFree(mtctx->roundBuff.buffer, mtctx->cMem);
ZSTD_customFree(mtctx, mtctx->cMem);
return 0;
}
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);
}
/* ZSTDMT_resize() :
* @return : error code if fails, 0 on success */
static size_t ZSTDMT_resize(ZSTDMT_CCtx* mtctx, unsigned nbWorkers)
{
if (POOL_resize(mtctx->factory, nbWorkers)) return ERROR(memory_allocation);
FORWARD_IF_ERROR( ZSTDMT_expandJobsTable(mtctx, nbWorkers) , "");
mtctx->bufPool = ZSTDMT_expandBufferPool(mtctx->bufPool, BUF_POOL_MAX_NB_BUFFERS(nbWorkers));
if (mtctx->bufPool == NULL) return ERROR(memory_allocation);
mtctx->cctxPool = ZSTDMT_expandCCtxPool(mtctx->cctxPool, nbWorkers);
if (mtctx->cctxPool == NULL) return ERROR(memory_allocation);
mtctx->seqPool = ZSTDMT_expandSeqPool(mtctx->seqPool, nbWorkers);
if (mtctx->seqPool == NULL) return ERROR(memory_allocation);
ZSTDMT_CCtxParam_setNbWorkers(&mtctx->params, nbWorkers);
return 0;
}
/*! 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;
{ ZSTD_compressionParameters cParams = ZSTD_getCParamsFromCCtxParams(cctxParams, ZSTD_CONTENTSIZE_UNKNOWN, 0, ZSTD_cpm_noAttachDict);
cParams.windowLog = saved_wlog;
mtctx->params.cParams = cParams;
}
}
/* ZSTDMT_getFrameProgression():
* tells how much data has been consumed (input) and produced (output) for current frame.
* able to count progression inside worker threads.
* Note : mutex will be acquired during statistics collection inside workers. */
ZSTD_frameProgression ZSTDMT_getFrameProgression(ZSTDMT_CCtx* mtctx)
{
ZSTD_frameProgression const fps = ZSTDMT_rust_frameProgressionWithJobs(
mtctx->consumed, mtctx->inBuff.filled, mtctx->produced,
mtctx->nextJobID, mtctx->doneJobID, mtctx->nextJobID,
mtctx->jobReady, mtctx->jobIDMask, mtctx,
ZSTDMT_projectJob);
DEBUGLOG(5, "ZSTDMT_getFrameProgression");
return fps;
}
size_t ZSTDMT_toFlushNow(ZSTDMT_CCtx* mtctx)
{
assert(mtctx->doneJobID <= mtctx->nextJobID);
return ZSTDMT_rust_toFlushNow(mtctx->doneJobID, mtctx->nextJobID,
mtctx->jobIDMask, mtctx,
ZSTDMT_projectJob);
}
/* ------------------------------------------ */
/* ===== Multi-threaded compression ===== */
/* ------------------------------------------ */
typedef struct {
ZSTDMT_CCtx* mtctx;
ZSTD_CCtx_params params;
const void* dict;
size_t dictSize;
ZSTD_dictContentType_e dictContentType;
const ZSTD_CDict* cdict;
unsigned long long pledgedSrcSize;
} ZSTDMT_initCStreamState;
static size_t ZSTDMT_initCStreamResize(void* opaque, unsigned nbWorkers)
{
ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque;
return ZSTDMT_resize(state->mtctx, nbWorkers);
}
static void ZSTDMT_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_initCStreamResetStream(void* opaque)
{
ZSTDMT_initCStreamState* const state = (ZSTDMT_initCStreamState*)opaque;
ZSTDMT_CCtx* const mtctx = state->mtctx;
DEBUGLOG(4, "roundBuff capacity : %u KB", (U32)(mtctx->roundBuff.capacity >> 10));
mtctx->roundBuff.pos = 0;
mtctx->inBuff.buffer = g_nullBuffer;
mtctx->inBuff.filled = 0;
mtctx->inBuff.prefix = kNullRange;
mtctx->doneJobID = 0;
mtctx->nextJobID = 0;
mtctx->frameEnded = 0;
mtctx->allJobsCompleted = 0;
mtctx->consumed = 0;
mtctx->produced = 0;
}
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(ZSTDMT_CCtx* mtctx, size_t srcSize, ZSTD_EndDirective endOp)
{
ZSTDMT_RustCreateJobProjection const projection = {
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
};
ZSTDMT_RustCreateJobResult const 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;
}
static size_t ZSTDMT_streamCreateJob(void* opaque, size_t srcSize, unsigned end)
{
return ZSTDMT_createCompressionJob((ZSTDMT_CCtx*)opaque, srcSize,
(ZSTD_EndDirective)end);
}
/* 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);
assert(projection->frameChecksumNeeded);
assert(projection->consumed == projection->srcSize);
assert(projection->cSize == job->cSize);
MEM_writeLE32((char*)job->dstBuff.start + job->cSize, checksum);
job->cSize += 4;
job->frameChecksumNeeded = 0;
projection->cSize = job->cSize;
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_flushError(void* opaque)
{
ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
ZSTDMT_waitForAllJobsCompleted(mtctx);
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_flushError);
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 };
}
static int ZSTDMT_doesOverlapWindow(Buffer buffer, ZSTD_window_t window)
{
DEBUGLOG(5, "ZSTDMT_doesOverlapWindow");
DEBUGLOG(5, "extDict [0x%zx, 0x%zx)",
(size_t)window.dictBase + window.lowLimit,
(size_t)window.dictBase + window.dictLimit);
DEBUGLOG(5, "prefix [0x%zx, 0x%zx)",
(size_t)window.base + window.dictLimit,
(size_t)window.nextSrc);
return ZSTDMT_rust_doesOverlapWindow(buffer.start, buffer.capacity,
window.nextSrc, window.base,
window.dictBase, window.dictLimit,
window.lowLimit);
}
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;
}
static int ZSTDMT_streamLoadInput(void* opaque, const void* src, size_t size)
{
ZSTDMT_CCtx* const mtctx = (ZSTDMT_CCtx*)opaque;
assert(mtctx->inBuff.buffer.start != NULL);
assert(mtctx->inBuff.filled <= mtctx->inBuff.buffer.capacity);
assert(size <= mtctx->inBuff.buffer.capacity - mtctx->inBuff.filled);
ZSTD_memcpy((char*)mtctx->inBuff.buffer.start + mtctx->inBuff.filled, src, size);
mtctx->inBuff.filled += size;
return 1;
}
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_streamLoadInput,
ZSTDMT_streamCreateJob, 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);
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