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

feat(compress): port multithreaded resource pools

Browse Source 
Move the reusable buffer and compression-context pools behind a Rust
implementation with a narrow C ABI adapter. The scheduler, job table, serial
LDM state, and stream orchestration remain in C until their private layouts
are ported.

Keep context types opaque across Rust modules so the pool bridge does not
depend on private C layout declarations. This also keeps the existing custom
allocator and pool replacement contracts intact.

Test Plan:
- RUSTC_WRAPPER= CARGO_BUILD_RUSTC_WRAPPER= cargo clippy --manifest-path
  rust/Cargo.toml --all-targets -- -D warnings
- RUSTC_WRAPPER= CARGO_BUILD_RUSTC_WRAPPER= cargo test --manifest-path
  rust/Cargo.toml zstdmt_compress
- make -B -C lib lib-mt
- git diff --cached --check
This commit is contained in:
ddidderr
2026-07-12 10:13:41 +02:00
parent e75574690b
commit 45cb1f509e
4 changed files with 681 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/compress/zstdmt_compress.c
+73 -170
View File
@@ -94,28 +94,36 @@ typedef struct buffer_s {
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);
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 ZSTDMT_bufferPool_s {
ZSTD_pthread_mutex_t poolMutex;
size_t bufferSize;
unsigned totalBuffers;
unsigned nbBuffers;
ZSTDMT_RustBufferPool* rustPool;
ZSTD_customMem cMem;
Buffer* buffers;
} ZSTDMT_bufferPool;
static void ZSTDMT_freeBufferPool(ZSTDMT_bufferPool* bufPool)
{
DEBUGLOG(3, "ZSTDMT_freeBufferPool (address:%08X)", (U32)(size_t)bufPool);
if (!bufPool) return; /* compatibility with free on NULL */
if (bufPool->buffers) {
unsigned u;
for (u=0; u<bufPool->totalBuffers; u++) {
DEBUGLOG(4, "free buffer %2u (address:%08X)", u, (U32)(size_t)bufPool->buffers[u].start);
ZSTD_customFree(bufPool->buffers[u].start, bufPool->cMem);
}
ZSTD_customFree(bufPool->buffers, bufPool->cMem);
}
ZSTD_pthread_mutex_destroy(&bufPool->poolMutex);
ZSTDMT_rust_buffer_pool_free(bufPool->rustPool);
ZSTD_customFree(bufPool, bufPool->cMem);
}
@@ -124,18 +132,11 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned maxNbBuffers, ZSTD_cu
ZSTDMT_bufferPool* const bufPool =
(ZSTDMT_bufferPool*)ZSTD_customCalloc(sizeof(ZSTDMT_bufferPool), cMem);
if (bufPool==NULL) return NULL;
if (ZSTD_pthread_mutex_init(&bufPool->poolMutex, NULL)) {
bufPool->rustPool = ZSTDMT_rust_buffer_pool_create(maxNbBuffers, cMem);
if (bufPool->rustPool == NULL) {
ZSTD_customFree(bufPool, cMem);
return NULL;
}
bufPool->buffers = (Buffer*)ZSTD_customCalloc(maxNbBuffers * sizeof(Buffer), cMem);
if (bufPool->buffers==NULL) {
ZSTDMT_freeBufferPool(bufPool);
return NULL;
}
bufPool->bufferSize = 64 KB;
bufPool->totalBuffers = maxNbBuffers;
bufPool->nbBuffers = 0;
bufPool->cMem = cMem;
return bufPool;
}
@@ -143,16 +144,8 @@ static ZSTDMT_bufferPool* ZSTDMT_createBufferPool(unsigned maxNbBuffers, ZSTD_cu
/* only works at initialization, not during compression */
static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)
{
size_t const poolSize = sizeof(*bufPool);
size_t const arraySize = bufPool->totalBuffers * sizeof(Buffer);
unsigned u;
size_t totalBufferSize = 0;
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
for (u=0; u<bufPool->totalBuffers; u++)
totalBufferSize += bufPool->buffers[u].capacity;
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
return poolSize + arraySize + totalBufferSize;
if (bufPool == NULL) return 0;
return sizeof(*bufPool) + ZSTDMT_rust_buffer_pool_sizeof(bufPool->rustPool);
}
/* ZSTDMT_setBufferSize() :
@@ -161,28 +154,21 @@ static size_t ZSTDMT_sizeof_bufferPool(ZSTDMT_bufferPool* bufPool)
* as they become freely interchangeable, reducing malloc/free usages and memory fragmentation */
static void ZSTDMT_setBufferSize(ZSTDMT_bufferPool* const bufPool, size_t const bSize)
{
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
DEBUGLOG(4, "ZSTDMT_setBufferSize: bSize = %u", (U32)bSize);
bufPool->bufferSize = bSize;
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
ZSTDMT_rust_buffer_pool_set_size(bufPool->rustPool, bSize);
}
static ZSTDMT_bufferPool* ZSTDMT_expandBufferPool(ZSTDMT_bufferPool* srcBufPool, unsigned maxNbBuffers)
{
if (srcBufPool==NULL) return NULL;
if (srcBufPool->totalBuffers >= maxNbBuffers) /* good enough */
return srcBufPool;
/* need a larger buffer pool */
{ ZSTD_customMem const cMem = srcBufPool->cMem;
size_t const bSize = srcBufPool->bufferSize; /* forward parameters */
ZSTDMT_bufferPool* newBufPool;
ZSTDMT_freeBufferPool(srcBufPool);
newBufPool = ZSTDMT_createBufferPool(maxNbBuffers, cMem);
if (newBufPool==NULL) return newBufPool;
ZSTDMT_setBufferSize(newBufPool, bSize);
return newBufPool;
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() :
@@ -191,38 +177,10 @@ static ZSTDMT_bufferPool* ZSTDMT_expandBufferPool(ZSTDMT_bufferPool* srcBufPool,
* note: allocation may fail, in this case, start==NULL and size==0 */
static Buffer ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
{
size_t const bSize = bufPool->bufferSize;
DEBUGLOG(5, "ZSTDMT_getBuffer: bSize = %u", (U32)bufPool->bufferSize);
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers) { /* try to use an existing buffer */
Buffer const buf = bufPool->buffers[--(bufPool->nbBuffers)];
size_t const availBufferSize = buf.capacity;
bufPool->buffers[bufPool->nbBuffers] = g_nullBuffer;
if ((availBufferSize >= bSize) & ((availBufferSize>>3) <= bSize)) {
/* large enough, but not too much */
DEBUGLOG(5, "ZSTDMT_getBuffer: provide buffer %u of size %u",
bufPool->nbBuffers, (U32)buf.capacity);
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
return buf;
}
/* size conditions not respected : scratch this buffer, create new one */
DEBUGLOG(5, "ZSTDMT_getBuffer: existing buffer does not meet size conditions => freeing");
ZSTD_customFree(buf.start, bufPool->cMem);
}
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
/* create new buffer */
DEBUGLOG(5, "ZSTDMT_getBuffer: create a new buffer");
{ Buffer buffer;
void* const start = ZSTD_customMalloc(bSize, bufPool->cMem);
buffer.start = start; /* note : start can be NULL if malloc fails ! */
buffer.capacity = (start==NULL) ? 0 : bSize;
if (start==NULL) {
DEBUGLOG(5, "ZSTDMT_getBuffer: buffer allocation failure !!");
} else {
DEBUGLOG(5, "ZSTDMT_getBuffer: created buffer of size %u", (U32)bSize);
}
return buffer;
}
ZSTDMT_RustBuffer const rustBuffer =
ZSTDMT_rust_buffer_pool_get(bufPool->rustPool);
Buffer buffer = { rustBuffer.start, rustBuffer.capacity };
return buffer;
}
#if ZSTD_RESIZE_SEQPOOL
@@ -233,41 +191,19 @@ static Buffer ZSTDMT_getBuffer(ZSTDMT_bufferPool* bufPool)
*/
static Buffer ZSTDMT_resizeBuffer(ZSTDMT_bufferPool* bufPool, Buffer buffer)
{
size_t const bSize = bufPool->bufferSize;
if (buffer.capacity < bSize) {
void* const start = ZSTD_customMalloc(bSize, bufPool->cMem);
Buffer newBuffer;
newBuffer.start = start;
newBuffer.capacity = start == NULL ? 0 : bSize;
if (start != NULL) {
assert(newBuffer.capacity >= buffer.capacity);
ZSTD_memcpy(newBuffer.start, buffer.start, buffer.capacity);
DEBUGLOG(5, "ZSTDMT_resizeBuffer: created buffer of size %u", (U32)bSize);
return newBuffer;
}
DEBUGLOG(5, "ZSTDMT_resizeBuffer: buffer allocation failure !!");
}
return 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)
{
DEBUGLOG(5, "ZSTDMT_releaseBuffer");
if (buf.start == NULL) return; /* compatible with release on NULL */
ZSTD_pthread_mutex_lock(&bufPool->poolMutex);
if (bufPool->nbBuffers < bufPool->totalBuffers) {
bufPool->buffers[bufPool->nbBuffers++] = buf; /* stored for later use */
DEBUGLOG(5, "ZSTDMT_releaseBuffer: stored buffer of size %u in slot %u",
(U32)buf.capacity, (U32)(bufPool->nbBuffers-1));
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
return;
}
ZSTD_pthread_mutex_unlock(&bufPool->poolMutex);
/* Reached bufferPool capacity (note: should not happen) */
DEBUGLOG(5, "ZSTDMT_releaseBuffer: pool capacity reached => freeing ");
ZSTD_customFree(buf.start, bufPool->cMem);
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.
@@ -308,9 +244,6 @@ static Buffer seqToBuffer(RawSeqStore_t seq)
static RawSeqStore_t ZSTDMT_getSeq(ZSTDMT_seqPool* seqPool)
{
if (seqPool->bufferSize == 0) {
return kNullRawSeqStore;
}
return bufferToSeq(ZSTDMT_getBuffer(seqPool));
}
@@ -353,25 +286,28 @@ static ZSTDMT_seqPool* ZSTDMT_expandSeqPool(ZSTDMT_seqPool* pool, U32 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);
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 {
ZSTD_pthread_mutex_t poolMutex;
int totalCCtx;
int availCCtx;
ZSTDMT_RustCCtxPool* rustPool;
int totalCCtx; /* kept for the existing MT parameter diagnostics */
ZSTD_customMem cMem;
ZSTD_CCtx** cctxs;
} 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;
ZSTD_pthread_mutex_destroy(&pool->poolMutex);
if (pool->cctxs) {
int cid;
for (cid=0; cid<pool->totalCCtx; cid++)
ZSTD_freeCCtx(pool->cctxs[cid]); /* free compatible with NULL */
ZSTD_customFree(pool->cctxs, pool->cMem);
}
ZSTDMT_rust_cctx_pool_free(pool->rustPool);
ZSTD_customFree(pool, pool->cMem);
}
@@ -384,20 +320,13 @@ static ZSTDMT_CCtxPool* ZSTDMT_createCCtxPool(int nbWorkers,
(ZSTDMT_CCtxPool*) ZSTD_customCalloc(sizeof(ZSTDMT_CCtxPool), cMem);
assert(nbWorkers > 0);
if (!cctxPool) return NULL;
if (ZSTD_pthread_mutex_init(&cctxPool->poolMutex, 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->cctxs = (ZSTD_CCtx**)ZSTD_customCalloc(nbWorkers * sizeof(ZSTD_CCtx*), cMem);
if (!cctxPool->cctxs) {
ZSTDMT_freeCCtxPool(cctxPool);
return NULL;
}
cctxPool->cMem = cMem;
cctxPool->cctxs[0] = ZSTD_createCCtx_advanced(cMem);
if (!cctxPool->cctxs[0]) { ZSTDMT_freeCCtxPool(cctxPool); return NULL; }
cctxPool->availCCtx = 1; /* at least one cctx for single-thread mode */
DEBUGLOG(3, "cctxPool created, with %u workers", nbWorkers);
return cctxPool;
}
@@ -406,59 +335,33 @@ static ZSTDMT_CCtxPool* ZSTDMT_expandCCtxPool(ZSTDMT_CCtxPool* srcPool,
int nbWorkers)
{
if (srcPool==NULL) return NULL;
if (nbWorkers <= srcPool->totalCCtx) return srcPool; /* good enough */
/* need a larger cctx pool */
{ ZSTD_customMem const cMem = srcPool->cMem;
ZSTDMT_freeCCtxPool(srcPool);
return ZSTDMT_createCCtxPool(nbWorkers, cMem);
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)
{
ZSTD_pthread_mutex_lock(&cctxPool->poolMutex);
{ unsigned const nbWorkers = cctxPool->totalCCtx;
size_t const poolSize = sizeof(*cctxPool);
size_t const arraySize = cctxPool->totalCCtx * sizeof(ZSTD_CCtx*);
size_t totalCCtxSize = 0;
unsigned u;
for (u=0; u<nbWorkers; u++) {
totalCCtxSize += ZSTD_sizeof_CCtx(cctxPool->cctxs[u]);
}
ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);
assert(nbWorkers > 0);
return poolSize + arraySize + totalCCtxSize;
}
if (cctxPool == NULL) return 0;
return sizeof(*cctxPool) + ZSTDMT_rust_cctx_pool_sizeof(cctxPool->rustPool);
}
static ZSTD_CCtx* ZSTDMT_getCCtx(ZSTDMT_CCtxPool* cctxPool)
{
DEBUGLOG(5, "ZSTDMT_getCCtx");
ZSTD_pthread_mutex_lock(&cctxPool->poolMutex);
if (cctxPool->availCCtx) {
cctxPool->availCCtx--;
{ ZSTD_CCtx* const cctx = cctxPool->cctxs[cctxPool->availCCtx];
ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);
return cctx;
} }
ZSTD_pthread_mutex_unlock(&cctxPool->poolMutex);
DEBUGLOG(5, "create one more CCtx");
return ZSTD_createCCtx_advanced(cctxPool->cMem); /* note : can be NULL, when creation fails ! */
return ZSTDMT_rust_cctx_pool_get(cctxPool->rustPool);
}
static void ZSTDMT_releaseCCtx(ZSTDMT_CCtxPool* pool, ZSTD_CCtx* cctx)
{
if (cctx==NULL) return; /* compatibility with release on NULL */
ZSTD_pthread_mutex_lock(&pool->poolMutex);
if (pool->availCCtx < pool->totalCCtx)
pool->cctxs[pool->availCCtx++] = cctx;
else {
/* pool overflow : should not happen, since totalCCtx==nbWorkers */
DEBUGLOG(4, "CCtx pool overflow : free cctx");
ZSTD_freeCCtx(cctx);
}
ZSTD_pthread_mutex_unlock(&pool->poolMutex);
ZSTDMT_rust_cctx_pool_release(pool->rustPool, cctx);
}
/* ==== Serial State ==== */