ddidderr/zstd-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(mt): move context construction policy to Rust

Browse Source 
Move MT context validation, worker clamping, resource-construction order, and
partial-failure cleanup policy into Rust. Keep the private context layout,
allocator-owned resources, pools, jobs table, and synchronization primitives in
C behind callbacks.

Test Plan:
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo +nightly 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 --all-targets (782 passed)
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo clippy --manifest-path rust/cli/Cargo.toml --all-targets -- -D warnings
- ulimit -v 41943040; CARGO_BUILD_JOBS=1 cargo test --manifest-path rust/cli/Cargo.toml --all-targets (179 passed)
- ulimit -v 41943040; make -j1
- ulimit -v 41943040; make -j1 -C tests test (all tests completed successfully)
This commit is contained in:
ddidderr
2026-07-20 06:44:03 +02:00
parent a00ba4c2b7
commit 2b2f9df5de
2 changed files with 520 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+155 -39
View File
@@ -642,6 +642,46 @@ typedef char ZSTDMT_rust_free_cctx_state_layout[
&& offsetof(ZSTDMT_RustFreeCCtxState, freeMTctx) == 12 * sizeof(void*)
&& sizeof(ZSTDMT_RustFreeCCtxState) == 13 * sizeof(void*)) ? 1 : -1];
size_t ZSTDMT_rust_freeCCtx(const ZSTDMT_RustFreeCCtxState* state);
typedef struct {
void* callbackContext;
unsigned requestedNbWorkers;
unsigned maxNbWorkers;
size_t contextSize;
ZSTD_customMem customMem;
} ZSTDMT_RustCreateCCtxProjection;
typedef char ZSTDMT_rust_create_cctx_projection_layout[
(offsetof(ZSTDMT_RustCreateCCtxProjection, callbackContext) == 0
&& offsetof(ZSTDMT_RustCreateCCtxProjection, requestedNbWorkers)
== sizeof(void*)
&& offsetof(ZSTDMT_RustCreateCCtxProjection, maxNbWorkers)
== sizeof(void*) + sizeof(unsigned)
&& offsetof(ZSTDMT_RustCreateCCtxProjection, contextSize)
== (sizeof(void*) == 8 ? 16 : 12)
&& offsetof(ZSTDMT_RustCreateCCtxProjection, customMem)
== (sizeof(void*) == 8 ? 24 : 16)
&& sizeof(ZSTDMT_RustCreateCCtxProjection)
== (sizeof(void*) == 8 ? 48 : 28)) ? 1 : -1];
typedef void* (*ZSTDMT_createCCtxAllocateFn)(void* opaque, size_t size);
typedef size_t (*ZSTDMT_createCCtxSetWorkersFn)(void* opaque, unsigned nbWorkers);
typedef void (*ZSTDMT_createCCtxSetInitialStateFn)(void* opaque);
typedef void* (*ZSTDMT_createCCtxFactoryFn)(void* opaque, unsigned nbWorkers);
typedef void* (*ZSTDMT_createCCtxJobsFn)(void* opaque, unsigned* nbJobs);
typedef void* (*ZSTDMT_createCCtxResourceFn)(void* opaque, unsigned nbWorkers);
typedef int (*ZSTDMT_createCCtxSerialInitFn)(void* opaque);
typedef void (*ZSTDMT_createCCtxFreeFn)(void* opaque);
void* ZSTDMT_rust_createCCtx(
const ZSTDMT_RustCreateCCtxProjection* projection,
ZSTDMT_createCCtxAllocateFn allocate,
ZSTDMT_createCCtxSetWorkersFn setWorkers,
ZSTDMT_createCCtxSetInitialStateFn setInitialState,
ZSTDMT_createCCtxFactoryFn createFactory,
ZSTDMT_createCCtxJobsFn createJobs,
ZSTDMT_createCCtxResourceFn createBufferPool,
ZSTDMT_createCCtxResourceFn createCCtxPool,
ZSTDMT_createCCtxResourceFn createSeqPool,
ZSTDMT_createCCtxSerialInitFn initSerial,
ZSTDMT_createCCtxFreeFn freeContext);
size_t ZSTDMT_rust_initCStream(
const ZSTDMT_RustInitCStreamProjection* projection, void* opaque,
ZSTDMT_initResizeFn resize, ZSTDMT_initDrainFn drain,
@@ -1700,8 +1740,6 @@ typedef struct {
*/
} 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,
@@ -1831,46 +1869,124 @@ static size_t ZSTDMT_CCtxParam_setNbWorkers(ZSTD_CCtx_params* params, unsigned n
return ZSTD_CCtxParams_setParameter(params, ZSTD_c_nbWorkers, (int)nbWorkers);
}
typedef struct {
ZSTDMT_CCtx* mtctx;
ZSTD_customMem cMem;
ZSTD_threadPool* pool;
} ZSTDMT_rust_createCCtx_context;
static void* ZSTDMT_rust_createCCtx_allocate(void* opaque, size_t size)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
context->mtctx = (ZSTDMT_CCtx*)ZSTD_customCalloc(size, context->cMem);
return context->mtctx;
}
static size_t ZSTDMT_rust_createCCtx_setWorkers(void* opaque, unsigned nbWorkers)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
return ZSTDMT_CCtxParam_setNbWorkers(&context->mtctx->params, nbWorkers);
}
static void ZSTDMT_rust_createCCtx_setInitialState(void* opaque)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
context->mtctx->cMem = context->cMem;
context->mtctx->allJobsCompleted = 1;
}
static void* ZSTDMT_rust_createCCtx_factory(void* opaque, unsigned nbWorkers)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
if (context->pool != NULL) {
context->mtctx->factory = context->pool;
context->mtctx->providedFactory = 1;
} else {
context->mtctx->factory = POOL_create_advanced(nbWorkers, 0, context->cMem);
context->mtctx->providedFactory = 0;
}
return context->mtctx->factory;
}
static void* ZSTDMT_rust_createCCtx_jobs(void* opaque, unsigned* nbJobs)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
context->mtctx->jobs = ZSTDMT_createJobsTable((U32*)nbJobs, context->cMem);
if (context->mtctx->jobs != NULL)
context->mtctx->jobIDMask = *nbJobs - 1;
return context->mtctx->jobs;
}
static void* ZSTDMT_rust_createCCtx_bufferPool(void* opaque, unsigned nbWorkers)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
context->mtctx->bufPool = ZSTDMT_createBufferPool(
BUF_POOL_MAX_NB_BUFFERS(nbWorkers), context->cMem);
return context->mtctx->bufPool;
}
static void* ZSTDMT_rust_createCCtx_cctxPool(void* opaque, unsigned nbWorkers)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
context->mtctx->cctxPool = ZSTDMT_createCCtxPool(nbWorkers, context->cMem);
return context->mtctx->cctxPool;
}
static void* ZSTDMT_rust_createCCtx_seqPool(void* opaque, unsigned nbWorkers)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
context->mtctx->seqPool = ZSTDMT_createSeqPool(nbWorkers, context->cMem);
return context->mtctx->seqPool;
}
static int ZSTDMT_rust_createCCtx_initSerial(void* opaque)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
return ZSTDMT_serialState_init(&context->mtctx->serial);
}
static void ZSTDMT_rust_createCCtx_free(void* opaque)
{
ZSTDMT_rust_createCCtx_context* const context =
(ZSTDMT_rust_createCCtx_context*)opaque;
if (context->mtctx != NULL) {
ZSTDMT_freeCCtx(context->mtctx);
context->mtctx = NULL;
}
}
MEM_STATIC ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced_internal(unsigned nbWorkers, ZSTD_customMem cMem, ZSTD_threadPool* pool)
{
ZSTDMT_CCtx* mtctx;
U32 nbJobs = nbWorkers + 2;
int initError;
ZSTDMT_rust_createCCtx_context context = { NULL, cMem, pool };
ZSTDMT_RustCreateCCtxProjection const projection = {
&context,
nbWorkers,
ZSTDMT_NBWORKERS_MAX,
sizeof(ZSTDMT_CCtx),
cMem
};
DEBUGLOG(3, "ZSTDMT_createCCtx_advanced (nbWorkers = %u)", nbWorkers);
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;
return (ZSTDMT_CCtx*)ZSTDMT_rust_createCCtx(
&projection,
ZSTDMT_rust_createCCtx_allocate,
ZSTDMT_rust_createCCtx_setWorkers,
ZSTDMT_rust_createCCtx_setInitialState,
ZSTDMT_rust_createCCtx_factory,
ZSTDMT_rust_createCCtx_jobs,
ZSTDMT_rust_createCCtx_bufferPool,
ZSTDMT_rust_createCCtx_cctxPool,
ZSTDMT_rust_createCCtx_seqPool,
ZSTDMT_rust_createCCtx_initSerial,
ZSTDMT_rust_createCCtx_free);
}
ZSTDMT_CCtx* ZSTDMT_createCCtx_advanced(unsigned nbWorkers, ZSTD_customMem cMem, ZSTD_threadPool* pool)