feat(mt): move external sequence gate to Rust

The MT worker previously kept the external-sequence non-empty gate and LDM
invariant assertion in C, bundled with the private CCtx operation. Project the
CCtx pointer and raw-sequence store through a layout-checked seam so Rust owns
the gate and the apply point while C retains only the opaque
ZSTD_referenceExternalSequences() leaf. Preserve the callback ordering after
worker CCtx initialization and keep empty stores as a no-op.

Test Plan:
- git diff --cached --check -- passed before commit.
- Static rg inspection confirmed all projection initializers include the new
  sequence-state pointer and no old serialState_applySequences symbol remains.
- Cargo, make, compiler checks, and full tests were not run per worker scope.
This commit is contained in:
2026-07-21 20:42:30 +02:00
parent a3ffd36a63
commit 484ffa4d93
2 changed files with 182 additions and 32 deletions
+49 -29
View File
@@ -220,18 +220,24 @@ typedef char ZSTDMT_compression_job_frame_header_state_layout[
== (sizeof(void*) == 8 ? 72 : 48))
? 1 : -1];
typedef struct ZSTDMT_RustCompressionJobSequenceState_s
ZSTDMT_RustCompressionJobSequenceState;
typedef struct {
unsigned firstJob;
unsigned lastJob;
const ZSTDMT_RustCompressionJobFrameHeaderState* frameHeaderState;
const ZSTDMT_RustCompressionJobSequenceState* sequenceState;
} ZSTDMT_RustCompressionJobProjection;
typedef char ZSTDMT_compression_job_projection_layout[
(offsetof(ZSTDMT_RustCompressionJobProjection, firstJob) == 0
&& offsetof(ZSTDMT_RustCompressionJobProjection, lastJob) == sizeof(unsigned)
&& offsetof(ZSTDMT_RustCompressionJobProjection, frameHeaderState)
== 2 * sizeof(unsigned)
&& offsetof(ZSTDMT_RustCompressionJobProjection, sequenceState)
== 2 * sizeof(unsigned) + sizeof(void*)
&& sizeof(ZSTDMT_RustCompressionJobProjection)
== 2 * sizeof(unsigned) + sizeof(void*))
== 2 * sizeof(unsigned) + 2 * sizeof(void*))
? 1 : -1];
typedef struct {
@@ -260,6 +266,8 @@ typedef ZSTDMT_chunkProcessResult (*ZSTDMT_compressionJobCompressFn)(
void* opaque, unsigned lastJob);
typedef void (*ZSTDMT_compressionJobSizeFn)(void* opaque, size_t size);
typedef size_t (*ZSTDMT_compressionJobSetParameterFn)(void* opaque, int value);
typedef void (*ZSTDMT_compressionJobApplySequencesFn)(
void* opaque, void* cctx, void* sequences, size_t nbSequences);
typedef struct {
void* callbackContext;
@@ -350,7 +358,7 @@ void ZSTDMT_rust_compressionJob(
ZSTDMT_compressionJobVoidFn prepareParameters,
ZSTDMT_compressionJobVoidFn generateSequences,
ZSTDMT_compressionJobStepFn beginJob,
ZSTDMT_compressionJobVoidFn applySequences,
ZSTDMT_compressionJobApplySequencesFn applySequences,
ZSTDMT_compressionJobCompressFn compressJob,
ZSTDMT_compressionJobVoidFn traceJob);
@@ -599,6 +607,21 @@ typedef char ZSTDMT_rust_raw_seq_store_layout[
&& offsetof(ZSTDMT_RustRawSeqStore, capacity) == offsetof(RawSeqStore_t, capacity))
? 1 : -1];
struct ZSTDMT_RustCompressionJobSequenceState_s {
ZSTD_CCtx** cctx;
ZSTDMT_RustRawSeqStore* rawSeqStore;
int ldmEnabled;
};
typedef char ZSTDMT_compression_job_sequence_state_layout[
(offsetof(ZSTDMT_RustCompressionJobSequenceState, cctx) == 0
&& offsetof(ZSTDMT_RustCompressionJobSequenceState, rawSeqStore)
== sizeof(void*)
&& offsetof(ZSTDMT_RustCompressionJobSequenceState, ldmEnabled)
== 2 * sizeof(void*)
&& sizeof(ZSTDMT_RustCompressionJobSequenceState)
== (sizeof(void*) == 8 ? 24 : 12))
? 1 : -1];
ZSTDMT_RustRawSeqStore ZSTDMT_rust_bufferToSeq(ZSTDMT_RustBuffer buffer);
ZSTDMT_RustBuffer ZSTDMT_rust_seqToBuffer(ZSTDMT_RustRawSeqStore seq);
typedef int (*ZSTDMT_serialWaitForTurnFn)(void* opaque, unsigned jobID);
@@ -1840,17 +1863,17 @@ static void ZSTDMT_serialState_onSkip(void* opaque, unsigned jobID, size_t 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)
/* Rust owns the non-empty/LDM gate and the point at which this callback is
* invoked. Keep only the private codec operation on the C side. */
static void ZSTDMT_compressionJobReferenceSequences(
void* opaque, void* cctx, void* sequences, size_t nbSequences)
{
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);
}
(void)opaque;
assert(cctx != NULL);
DEBUGLOG(5, "ZSTDMT_compressionJob: uploading %u external sequences",
(unsigned)nbSequences);
ZSTD_referenceExternalSequences((ZSTD_CCtx*)cctx,
(rawSeq*)sequences, nbSequences);
}
static void ZSTDMT_serialState_ensureFinished(SerialState* serialState,
@@ -2120,17 +2143,6 @@ static void ZSTDMT_compressionJobPublishFrameHeader(void* opaque)
? 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)
{
@@ -2247,17 +2259,25 @@ static void ZSTDMT_compressionJob(void* jobDescription)
{
ZSTDMT_jobDescription* const job = (ZSTDMT_jobDescription*)jobDescription;
ZSTDMT_compressionJobState state;
ZSTDMT_RustCompressionJobSequenceState sequenceState;
ZSTDMT_RustCompressionJobFinishProjection finishProjection;
ZSTDMT_RustCompressionJobProjection const projection = {
job->firstJob,
job->lastJob,
&state.frameHeaderState
};
ZSTDMT_RustCompressionJobProjection projection;
ZSTD_memset(&state, 0, sizeof(state));
state.job = job;
state.jobParams = job->params; /* do not modify job->params ! copy it, modify the copy */
state.dstBuff = job->dstBuff;
sequenceState = (ZSTDMT_RustCompressionJobSequenceState){
&state.cctx,
(ZSTDMT_RustRawSeqStore*)&state.rawSeqStore,
job->serial->params.ldmParams.enableLdm
};
projection = (ZSTDMT_RustCompressionJobProjection){
job->firstJob,
job->lastJob,
&state.frameHeaderState,
&sequenceState
};
finishProjection = (ZSTDMT_RustCompressionJobFinishProjection){
&state,
job->src.size,
@@ -2276,7 +2296,7 @@ static void ZSTDMT_compressionJob(void* jobDescription)
ZSTDMT_compressionJobPrepareParameters,
ZSTDMT_compressionJobGenerateSequences,
ZSTDMT_compressionJobBegin,
ZSTDMT_compressionJobApplySequences,
ZSTDMT_compressionJobReferenceSequences,
ZSTDMT_compressionJobCompress,
ZSTDMT_compressionJobTrace);
}