feat(mt): move produced-output flush into Rust
Move the multithreaded produced-output state machine into Rust while keeping C-owned job descriptors, synchronization, checksum state, and buffer-pool cleanup behind projection callbacks. Rust now decides publication progress, job completion, pending-work results, and frame-completion reporting without crossing the private ZSTDMT context layout. Add focused tests for partial output, worker errors, checksum projection, and frame-end state. Test Plan: - cargo test --manifest-path rust/Cargo.toml --all-targets -- --test-threads=1 - cargo test --manifest-path rust/Cargo.toml --no-default-features --features compression,decompression,dict-builder,legacy-v01,legacy-v02,legacy-v03,legacy-v04,legacy-v05,legacy-v06,legacy-v07 --all-targets -- --test-threads=1 - make -B -C lib -j2 lib - make -B -C tests -j2 test-zstd test-rust-lib-smoke - cargo clippy --manifest-path rust/Cargo.toml --tests -- -D warnings
This commit is contained in:
+147
-84
@@ -140,6 +140,50 @@ ZSTDMT_flushPublicationResult ZSTDMT_rust_publishJobOutput(
|
||||
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);
|
||||
|
||||
typedef struct {
|
||||
unsigned lastJob;
|
||||
size_t srcSize;
|
||||
@@ -1512,6 +1556,87 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* mtctx, size_t srcSize, ZS
|
||||
}
|
||||
|
||||
|
||||
/* 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.
|
||||
@@ -1520,91 +1645,29 @@ static size_t ZSTDMT_createCompressionJob(ZSTDMT_CCtx* mtctx, size_t srcSize, ZS
|
||||
* @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)
|
||||
{
|
||||
unsigned const wJobID = mtctx->doneJobID & mtctx->jobIDMask;
|
||||
DEBUGLOG(5, "ZSTDMT_flushProduced (blocking:%u , job %u <= %u)",
|
||||
blockToFlush, mtctx->doneJobID, mtctx->nextJobID);
|
||||
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);
|
||||
|
||||
ZSTD_PTHREAD_MUTEX_LOCK(&mtctx->jobs[wJobID].job_mutex);
|
||||
if ( blockToFlush
|
||||
&& (mtctx->doneJobID < mtctx->nextJobID) ) {
|
||||
assert(mtctx->jobs[wJobID].dstFlushed <= mtctx->jobs[wJobID].cSize);
|
||||
while (mtctx->jobs[wJobID].dstFlushed == mtctx->jobs[wJobID].cSize) { /* nothing to flush */
|
||||
if (mtctx->jobs[wJobID].consumed == mtctx->jobs[wJobID].src.size) {
|
||||
DEBUGLOG(5, "job %u is completely consumed (%u == %u) => don't wait for cond, there will be none",
|
||||
mtctx->doneJobID, (U32)mtctx->jobs[wJobID].consumed, (U32)mtctx->jobs[wJobID].src.size);
|
||||
break;
|
||||
}
|
||||
DEBUGLOG(5, "waiting for something to flush from job %u (currently flushed: %u bytes)",
|
||||
mtctx->doneJobID, (U32)mtctx->jobs[wJobID].dstFlushed);
|
||||
ZSTD_pthread_cond_wait(&mtctx->jobs[wJobID].job_cond, &mtctx->jobs[wJobID].job_mutex); /* block when nothing to flush but some to come */
|
||||
} }
|
||||
|
||||
/* try to flush something */
|
||||
{ size_t cSize = mtctx->jobs[wJobID].cSize; /* shared */
|
||||
size_t const srcConsumed = mtctx->jobs[wJobID].consumed; /* shared */
|
||||
size_t const srcSize = mtctx->jobs[wJobID].src.size; /* read-only, could be done after mutex lock, but no-declaration-after-statement */
|
||||
ZSTD_pthread_mutex_unlock(&mtctx->jobs[wJobID].job_mutex);
|
||||
if (ZSTD_isError(cSize)) {
|
||||
DEBUGLOG(5, "ZSTDMT_flushProduced: job %u : compression error detected : %s",
|
||||
mtctx->doneJobID, ZSTD_getErrorName(cSize));
|
||||
ZSTDMT_waitForAllJobsCompleted(mtctx);
|
||||
ZSTDMT_releaseAllJobResources(mtctx);
|
||||
return cSize;
|
||||
}
|
||||
/* add frame checksum if necessary (can only happen once) */
|
||||
assert(srcConsumed <= srcSize);
|
||||
if ( (srcConsumed == srcSize) /* job completed -> worker no longer active */
|
||||
&& mtctx->jobs[wJobID].frameChecksumNeeded ) {
|
||||
U32 const checksum = (U32)XXH64_digest(&mtctx->serial.xxhState);
|
||||
DEBUGLOG(4, "ZSTDMT_flushProduced: writing checksum : %08X \n", checksum);
|
||||
MEM_writeLE32((char*)mtctx->jobs[wJobID].dstBuff.start + mtctx->jobs[wJobID].cSize, checksum);
|
||||
cSize += 4;
|
||||
mtctx->jobs[wJobID].cSize += 4; /* can write this shared value, as worker is no longer active */
|
||||
mtctx->jobs[wJobID].frameChecksumNeeded = 0;
|
||||
}
|
||||
|
||||
if (cSize > 0) { /* compression is ongoing or completed */
|
||||
ZSTDMT_flushPublicationResult const publication = ZSTDMT_rust_publishJobOutput(
|
||||
output->dst, output->size, output->pos,
|
||||
mtctx->jobs[wJobID].dstBuff.start, mtctx->jobs[wJobID].dstBuff.capacity,
|
||||
cSize, mtctx->jobs[wJobID].dstFlushed);
|
||||
if (publication.status != 0) {
|
||||
assert(publication.status == 0);
|
||||
return ERROR(GENERIC);
|
||||
}
|
||||
DEBUGLOG(5, "ZSTDMT_flushProduced: Flushing %u bytes from job %u (completion:%u/%u, generated:%u)",
|
||||
(U32)publication.toFlush, mtctx->doneJobID,
|
||||
(U32)srcConsumed, (U32)srcSize, (U32)cSize);
|
||||
assert(mtctx->doneJobID < mtctx->nextJobID);
|
||||
assert(cSize >= mtctx->jobs[wJobID].dstFlushed);
|
||||
assert(mtctx->jobs[wJobID].dstBuff.start != NULL);
|
||||
output->pos = publication.outputPos;
|
||||
mtctx->jobs[wJobID].dstFlushed = publication.dstFlushed; /* can write : this value is only used by mtctx */
|
||||
|
||||
if ( (srcConsumed == srcSize) /* job is completed */
|
||||
&& (mtctx->jobs[wJobID].dstFlushed == cSize) ) { /* output buffer fully flushed => free this job position */
|
||||
DEBUGLOG(5, "Job %u completed (%u bytes), moving to next one",
|
||||
mtctx->doneJobID, (U32)mtctx->jobs[wJobID].dstFlushed);
|
||||
ZSTDMT_releaseBuffer(mtctx->bufPool, mtctx->jobs[wJobID].dstBuff);
|
||||
DEBUGLOG(5, "dstBuffer released");
|
||||
mtctx->jobs[wJobID].dstBuff = g_nullBuffer;
|
||||
mtctx->jobs[wJobID].cSize = 0; /* ensure this job slot is considered "not started" in future check */
|
||||
mtctx->consumed += srcSize;
|
||||
mtctx->produced += cSize;
|
||||
mtctx->doneJobID++;
|
||||
} }
|
||||
|
||||
/* return value : how many bytes left in buffer ; fake it to 1 when unknown but >0 */
|
||||
if (cSize > mtctx->jobs[wJobID].dstFlushed) return (cSize - mtctx->jobs[wJobID].dstFlushed);
|
||||
if (srcSize > srcConsumed) return 1; /* current job not completely compressed */
|
||||
}
|
||||
if (mtctx->doneJobID < mtctx->nextJobID) return 1; /* some more jobs ongoing */
|
||||
if (mtctx->jobReady) return 1; /* one job is ready to push, just not yet in the list */
|
||||
if (mtctx->inBuff.filled > 0) return 1; /* input is not empty, and still needs to be converted into a job */
|
||||
mtctx->allJobsCompleted = mtctx->frameEnded; /* all jobs are entirely flushed => if this one is last one, frame is completed */
|
||||
if (end == ZSTD_e_end) return !mtctx->frameEnded; /* for ZSTD_e_end, question becomes : is frame completed ? instead of : are internal buffers fully flushed ? */
|
||||
return 0; /* internal buffers fully flushed */
|
||||
output->pos = result.outputPos;
|
||||
if (result.updateAllJobsCompleted)
|
||||
mtctx->allJobsCompleted = result.allJobsCompleted;
|
||||
return result.result;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user